From f8ed8e21922f400aeb73ba0505fed6209617051d Mon Sep 17 00:00:00 2001 From: lda Date: Thu, 10 Sep 2026 02:17:27 +0700 Subject: [PATCH] test: harden scheduling review boundaries --- tests/artifacts/test_store.py | 45 +++++++++- tests/core/test_input_sources.py | 2 +- tests/scheduling/test_history_reconcile.py | 86 +++++++++++++++++- tests/scheduling/test_lifecycle.py | 6 -- tests/scheduling/test_lock_identity.py | 1 + tests/scheduling/test_preparation.py | 4 +- tests/wf_api/test_resume_attempt.py | 6 +- tests/wf_api/test_schedules.py | 87 ++++++++++++++++++- .../wf_transport_rpc_http/test_rpc_models.py | 5 ++ .../test_schedules_rpc.py | 14 +-- 10 files changed, 231 insertions(+), 25 deletions(-) diff --git a/tests/artifacts/test_store.py b/tests/artifacts/test_store.py index b9cb499a..f2cd9e00 100644 --- a/tests/artifacts/test_store.py +++ b/tests/artifacts/test_store.py @@ -1,7 +1,10 @@ from __future__ import annotations import json +import threading from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any import pytest @@ -69,7 +72,7 @@ def test_file_store_round_trips_deployment(tmp_path) -> None: def test_concurrent_deployment_saves_advance_revision_without_lost_updates( - tmp_path, + tmp_path, monkeypatch ) -> None: store = FileWorkflowArtifactStore(tmp_path) store.save_deployment( @@ -88,9 +91,45 @@ def test_concurrent_deployment_saves_advance_revision_without_lost_updates( for version in range(2, 10) ] - with ThreadPoolExecutor(max_workers=len(updates)) as executor: - list(executor.map(store.save_deployment, updates)) + start_gate = threading.Barrier(len(updates)) + read_gate = threading.Barrier(len(updates)) + read_state_lock = threading.Lock() + active_reads = 0 + max_active_reads = 0 + deployment_path = (store.deployments_dir / "concurrent.personal.json").resolve() + original_read_text = Path.read_text + def coordinated_read(path: Path, *args: Any, **kwargs: Any) -> str: + nonlocal active_reads, max_active_reads + if path.resolve() != deployment_path: + return original_read_text(path, *args, **kwargs) + with read_state_lock: + active_reads += 1 + max_active_reads = max(max_active_reads, active_reads) + try: + # A broken implementation reaches this gate from every worker + # after reading the same revision. The real store's lock lets + # only one worker enter the read/modify/write window, so the + # gate times out once and later workers proceed immediately. + try: + read_gate.wait(timeout=0.5) + except threading.BrokenBarrierError: + pass + return original_read_text(path, *args, **kwargs) + finally: + with read_state_lock: + active_reads -= 1 + + monkeypatch.setattr(Path, "read_text", coordinated_read) + + def save(deployment: WorkflowDeployment) -> None: + start_gate.wait(timeout=10) + store.save_deployment(deployment) + + with ThreadPoolExecutor(max_workers=len(updates)) as executor: + list(executor.map(save, updates)) + + assert max_active_reads == 1 assert store.get_deployment("concurrent.personal").revision == 9 diff --git a/tests/core/test_input_sources.py b/tests/core/test_input_sources.py index 928df4d3..7d010eee 100644 --- a/tests/core/test_input_sources.py +++ b/tests/core/test_input_sources.py @@ -9,6 +9,7 @@ from wf_core.models.input_bindings import ( ObjectExpression, PathExpression, validate_input_expression_limits, + walk_expression_paths, ) from wf_core.runtime.input_bindings import ( resolve_input_expression, @@ -18,7 +19,6 @@ from wf_core.runtime.input_sources import ( GraphSourceResolver, MappingSourceResolver, resolve_composed_expression, - walk_expression_paths, ) diff --git a/tests/scheduling/test_history_reconcile.py b/tests/scheduling/test_history_reconcile.py index 5c3ed764..30d54d13 100644 --- a/tests/scheduling/test_history_reconcile.py +++ b/tests/scheduling/test_history_reconcile.py @@ -9,10 +9,12 @@ anew. from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime, timedelta, timezone from pathlib import Path from typing import Any, cast +import pytest + from tests.artifacts.test_run_store import artifact as _artifact from tests.artifacts.test_run_store import deployment as _deployment from tests.scheduling.controlled import ( @@ -30,7 +32,11 @@ from wf_artifacts.runs.models import ResumeAttempt from wf_artifacts.runs.store import FileRunStore from wf_core import RunState, RunStatus from wf_scheduling import recovery as sched_recovery -from wf_scheduling.history import FileScheduleHistoryRecorder +from wf_scheduling.history import ( + FileScheduleHistoryRecorder, + HistoryEntry, + entry_occurrence_id, +) from wf_scheduling.models import Schedule from wf_scheduling.ownership import SchedulerOwnership from wf_scheduling.poll import Scheduler @@ -136,6 +142,80 @@ def _recover( ownership.release() +def test_interval_identity_normalizes_offsets_and_deduplicates( + tmp_path: Path, +) -> None: + store = FileScheduleStore(tmp_path / "sched") + recorder = FileScheduleHistoryRecorder(store) + utc_start = datetime(2026, 9, 8, 12, 0, tzinfo=UTC) + utc_end = datetime(2026, 9, 8, 13, 0, tzinfo=UTC) + offset = timezone(timedelta(hours=2)) + + recorder.record( + HistoryEntry( + schedule_id="a", + kind="interval-summary", + interval_start=utc_start, + interval_end=utc_end, + interval_count=2, + created_at=utc_start, + ) + ) + recorder.record( + HistoryEntry( + schedule_id="a", + kind="interval-summary", + interval_start=utc_start.astimezone(offset), + interval_end=utc_end.astimezone(offset), + interval_count=2, + created_at=utc_end, + ) + ) + + rows = cast( + list[dict[str, Any]], store.list_occurrences("a", limit=10)["occurrences"] + ) + assert len(rows) == 1 + row = rows[0] + assert row["occurrence_id"] == ( + "a|summary|2026-09-08T12:00:00+00:00|2026-09-08T13:00:00+00:00" + ) + assert row["interval_start"] == "2026-09-08T12:00:00Z" + assert row["interval_end"] == "2026-09-08T13:00:00Z" + + +@pytest.mark.parametrize( + ("interval_start", "interval_end"), + [ + (datetime(2026, 9, 8, 12, 0), datetime(2026, 9, 8, 13, 0, tzinfo=UTC)), + (datetime(2026, 9, 8, 12, 0, tzinfo=UTC), datetime(2026, 9, 8, 13, 0)), + (None, datetime(2026, 9, 8, 13, 0, tzinfo=UTC)), + (datetime(2026, 9, 8, 12, 0, tzinfo=UTC), None), + ], +) +def test_interval_identity_rejects_naive_or_half_specified_bounds( + interval_start: datetime | None, + interval_end: datetime | None, +) -> None: + with pytest.raises(ValueError): + entry_occurrence_id( + HistoryEntry( + schedule_id="a", + kind="interval-summary", + interval_start=interval_start, + interval_end=interval_end, + ), + datetime(2026, 9, 8, 14, 0, tzinfo=UTC), + ) + + +def test_history_identity_without_interval_still_uses_created_time() -> None: + created = datetime(2026, 9, 8, 14, 0, tzinfo=UTC) + assert entry_occurrence_id( + HistoryEntry(schedule_id="a", kind="interval-summary"), created + ) == "a|summary|2026-09-08T14:00:00+00:00" + + def test_failed_run_persists_reason_without_callback(tmp_path: Path) -> None: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") @@ -466,7 +546,7 @@ def test_consumed_write_failure_recovers_without_dup(tmp_path: Path) -> None: sched_store, run_store, ownership, script={"*": "hang"} ) sched.sources["a"] = OneShotSource(intended) - with __import__("pytest").raises(OSError, match="injected consumed"): + with pytest.raises(OSError, match="injected consumed"): sched.poll(intended) assert [a.id for a in run_store.list_admissions()] != [] first = run_store.list_admissions()[0].id diff --git a/tests/scheduling/test_lifecycle.py b/tests/scheduling/test_lifecycle.py index 86c86750..24efdc91 100644 --- a/tests/scheduling/test_lifecycle.py +++ b/tests/scheduling/test_lifecycle.py @@ -261,8 +261,6 @@ async def test_settlement_does_not_block_the_event_loop( settlement_finished.set() monkeypatch.setattr(Scheduler, "record_stopped_execution", _blocking_settle) - release_timer = threading.Timer(1.0, release.set) - release_timer.start() try: await service.start() service.schedule_store.create_schedule(_sched_model("a", intended)) @@ -275,7 +273,6 @@ async def test_settlement_does_not_block_the_event_loop( await _wait_for(lambda: service.live_executions == 0) finally: release.set() - release_timer.cancel() await service.stop() @@ -303,8 +300,6 @@ async def test_shutdown_joins_cancelled_settlement_before_releasing_ownership( original(self, run_id, state, now) monkeypatch.setattr(Scheduler, "record_stopped_execution", _blocking_settle) - release_timer = threading.Timer(1.0, release.set) - release_timer.start() try: await service.start() service.schedule_store.create_schedule(_sched_model("a", intended)) @@ -328,7 +323,6 @@ async def test_shutdown_joins_cancelled_settlement_before_releasing_ownership( ).status.value == ("completed") finally: release.set() - release_timer.cancel() await service.stop() diff --git a/tests/scheduling/test_lock_identity.py b/tests/scheduling/test_lock_identity.py index 80302d2c..88eb8b1d 100644 --- a/tests/scheduling/test_lock_identity.py +++ b/tests/scheduling/test_lock_identity.py @@ -123,6 +123,7 @@ def test_competing_processes_share_the_canonical_lock(tmp_path: Path) -> None: ) proc = subprocess.run( [sys.executable, str(script_path)], + cwd=Path(__file__).resolve().parents[2], capture_output=True, text=True, timeout=60, diff --git a/tests/scheduling/test_preparation.py b/tests/scheduling/test_preparation.py index 70bb7612..31e63029 100644 --- a/tests/scheduling/test_preparation.py +++ b/tests/scheduling/test_preparation.py @@ -112,8 +112,8 @@ def test_production_scheduler_has_no_fixture_seams() -> None: def test_preparer_contract_is_a_protocol() -> None: - assert issubclass(InvocationPreparer, object) - assert issubclass(RunDispatcher, object) + assert typing.is_protocol(InvocationPreparer) + assert typing.is_protocol(RunDispatcher) def test_occurrence_bindings_resolve_into_admission(tmp_path: Path) -> None: diff --git a/tests/wf_api/test_resume_attempt.py b/tests/wf_api/test_resume_attempt.py index 36007231..8f685f52 100644 --- a/tests/wf_api/test_resume_attempt.py +++ b/tests/wf_api/test_resume_attempt.py @@ -82,8 +82,10 @@ def test_active_attempt_marker_is_durable(tmp_path: Path) -> None: # executing. Echo runs complete immediately so restore fails first on # non-interrupted status; the ACTIVE guard is exercised on interrupted # runs in recovery tests (T10). - assert store.get_resume_attempt(run_id) is not None - assert store.get_resume_attempt(run_id).state == "ACTIVE" # type: ignore[union-attr] + reopened = FileRunStore(store.root) + attempt = reopened.get_resume_attempt(run_id) + assert attempt is not None + assert attempt.state == "ACTIVE" def test_stopped_checkpoint_echoes_attempt_id(tmp_path: Path) -> None: diff --git a/tests/wf_api/test_schedules.py b/tests/wf_api/test_schedules.py index dd75e4ed..998e26bd 100644 --- a/tests/wf_api/test_schedules.py +++ b/tests/wf_api/test_schedules.py @@ -8,12 +8,18 @@ tests/wf_api/test_run_lifecycle.py). from __future__ import annotations +import asyncio +import threading from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any import pytest +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, +) from wf_api import WorkflowApi from wf_api.durable_context import durable_workflow_api from wf_api.models import TraceRange @@ -36,9 +42,12 @@ from wf_artifacts import ( from wf_authoring import NodeSpec from wf_core import InterruptRequest, RunState, RunStatus from wf_platform import CapabilitySource +from wf_scheduling.calendar import OneShotSource from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry from wf_scheduling.models import PendingCandidate from wf_scheduling.ownership import SchedulerOwnership +from wf_scheduling.poll import Scheduler +from wf_scheduling.prepare import SchedulePreparer from wf_scheduling.recovery import recover from wf_scheduling.store import ( FileScheduleStore, @@ -516,6 +525,74 @@ async def test_update_stale_revision_rejected(tmp_path: Path) -> None: assert _history_rows(sched_store, "s") == [] +async def test_admin_mutation_waits_for_poll_transaction(tmp_path: Path) -> None: + """A poll transition and an API edit cannot interleave store writes.""" + api, sched_store, run_store, artifact_store, _ = _harness( + tmp_path / "admin_poll" + ) + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.save_consumed("s", intended - timedelta(hours=1)) + + entered = threading.Event() + release = threading.Event() + block_once = True + original_save_consumed = sched_store.save_consumed + + def block_poll_watermark(schedule_id: str, consumed_through: datetime) -> None: + nonlocal block_once + if block_once: + block_once = False + entered.set() + if not release.wait(timeout=5): + raise TimeoutError("poll transaction did not get released") + original_save_consumed(schedule_id, consumed_through) + + sched_store.save_consumed = block_poll_watermark # type: ignore[method-assign] + ownership = SchedulerOwnership(tmp_path / "admin_poll", owner="poll").acquire() + scheduler = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources={"s": OneShotSource(intended)}, + capacity=1, + preparer=SchedulePreparer( + DictDeployments({"dep.personal": {"rev": 1, "required": []}}), + lambda sched: create_pinned_environment( + deployment=artifact_store.get_deployment(sched.deployment_id), + artifact=artifact_store.get_artifact("sched-art", 1), + tree=SavedSubgraphTree(artifacts_by_ref={}, diagnostics=[]), + ), + ), + dispatcher=ScriptedDispatcher({"*": "complete"}), + ownership=ownership, + ) + release_timer = threading.Timer(0.2, release.set) + try: + poll_task = asyncio.create_task(asyncio.to_thread(scheduler.poll, intended)) + assert await asyncio.to_thread(entered.wait, 5) + edit_task = asyncio.create_task( + api.update_schedule( + schedule_id="s", + expected_revision=1, + overlap="parallel", + max_active_runs=2, + ) + ) + release_timer.start() + poll_result, edited = await asyncio.gather(poll_task, edit_task) + finally: + release.set() + release_timer.cancel() + ownership.release() + + assert poll_result["s"].startswith("admit:") + assert edited["revision"] == 2 + assert run_store.list_admissions()[0].schedule_revision == 1 + assert sched_store.get_schedule("s").revision == 2 + + async def test_create_initializes_consumed_no_backfill(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "create_consumed") @@ -959,16 +1036,20 @@ async def test_occurrences_pagination_visits_tied_entries_once(tmp_path: Path) - seen: list[tuple[str, str | None]] = [] cursor: str | None = None - while True: + for _ in range(10): page = await api.list_schedule_occurrences( schedule_id="s", cursor=cursor, limit=1 ) assert page["total"] == 3 for row in page["occurrences"]: seen.append((row["kind"], row["checkpoint_id"])) - cursor = page["next_cursor"] - if cursor is None: + next_cursor = page["next_cursor"] + if next_cursor is None: break + assert next_cursor != cursor + cursor = next_cursor + else: + pytest.fail("occurrence pagination did not terminate") assert seen == [ ("admitted", None), ("interrupted", "run-1.000001"), diff --git a/tests/wf_transport_rpc_http/test_rpc_models.py b/tests/wf_transport_rpc_http/test_rpc_models.py index 9fb1afd1..474f4911 100644 --- a/tests/wf_transport_rpc_http/test_rpc_models.py +++ b/tests/wf_transport_rpc_http/test_rpc_models.py @@ -95,6 +95,11 @@ def test_update_schedule_params_match_create_numeric_constraints() -> None: UpdateScheduleParams( schedule_id="schedule", expected_revision=1, max_steps=cast(Any, "10") ) + for revision in (0, -1): + with pytest.raises(ValidationError): + UpdateScheduleParams( + schedule_id="schedule", expected_revision=revision + ) def test_schedule_policy_params_reject_unknown_values() -> None: diff --git a/tests/wf_transport_rpc_http/test_schedules_rpc.py b/tests/wf_transport_rpc_http/test_schedules_rpc.py index 6a7f1ae5..56200fbc 100644 --- a/tests/wf_transport_rpc_http/test_schedules_rpc.py +++ b/tests/wf_transport_rpc_http/test_schedules_rpc.py @@ -308,19 +308,23 @@ async def test_rpc_schedule_occurrences_tied_entries_traverse_once(tmp_path) -> ) seen: list[tuple[str, object]] = [] - cursor: object = None - while True: + cursor: str | None = None + for _ in range(10): page = await client.list_schedule_occurrences( schedule_id="s", cursor=cursor, - limit=1, # type: ignore[arg-type] + limit=1, ) assert page["total"] == 2 for row in page["occurrences"]: seen.append((row["kind"], row["checkpoint_id"])) - cursor = page["next_cursor"] - if cursor is None: + next_cursor = page["next_cursor"] + if next_cursor is None: break + assert next_cursor != cursor + cursor = next_cursor + else: + pytest.fail("occurrence pagination did not terminate") assert seen == [("admitted", None), ("interrupted", "run-1.000001")]