From 99c038a04fbdcfe8fe77f7bce8e0f19d5aaab888 Mon Sep 17 00:00:00 2001 From: lda Date: Wed, 9 Sep 2026 17:33:03 +0700 Subject: [PATCH] sched: truthful executing marks in capacity test shortcuts; format wave files --- src/wf_scheduling/resume_gate.py | 2 +- src/wf_scheduling/store.py | 8 ++------ tests/scheduling/test_lifecycle.py | 14 +++----------- tests/scheduling/test_poll.py | 9 +++++++-- tests/scheduling/test_schedule_store.py | 4 +++- tests/wf_server/test_scheduler_integration.py | 18 ++++++++++++++---- .../test_schedules_rpc.py | 4 +++- 7 files changed, 33 insertions(+), 26 deletions(-) diff --git a/src/wf_scheduling/resume_gate.py b/src/wf_scheduling/resume_gate.py index 48882a50..bc094bf4 100644 --- a/src/wf_scheduling/resume_gate.py +++ b/src/wf_scheduling/resume_gate.py @@ -124,5 +124,5 @@ class SchedulerResumeGate: checkpoint_id=checkpoint_id, now=service.clock(), ) - except (SecondOwnerError, OSError, ValueError): + except SecondOwnerError, OSError, ValueError: return False diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py index a5673e73..d50ae267 100644 --- a/src/wf_scheduling/store.py +++ b/src/wf_scheduling/store.py @@ -175,9 +175,7 @@ class FileScheduleStore: highest = -1 for index, item in enumerate(entries): ordinal = item.get("seq") - effective = ( - ordinal if isinstance(ordinal, int) else index - ) + effective = ordinal if isinstance(ordinal, int) else index if effective > highest: highest = effective if record.seq is None: @@ -310,9 +308,7 @@ class FileScheduleStore: last = page[-1] _, _, _, last_ordinal = page_keys[-1] iso = "" if last.resolved_at is None else last.resolved_at.isoformat() - next_cursor: str | None = ( - f"{iso}|{last.occurrence_id}|{int(last_ordinal)}" - ) + next_cursor: str | None = f"{iso}|{last.occurrence_id}|{int(last_ordinal)}" else: next_cursor = None return { diff --git a/tests/scheduling/test_lifecycle.py b/tests/scheduling/test_lifecycle.py index 1ea95076..118d48e1 100644 --- a/tests/scheduling/test_lifecycle.py +++ b/tests/scheduling/test_lifecycle.py @@ -559,9 +559,7 @@ async def test_store_deployment_directory_contract(tmp_path: Path) -> None: directory.required_inputs("missing.deployment") -async def _poll_two( - service: SchedulerService, intended: datetime -) -> tuple[str, str]: +async def _poll_two(service: SchedulerService, intended: datetime) -> tuple[str, str]: service.schedule_store.create_schedule(_sched_model("a", intended)) service.schedule_store.create_schedule(_sched_model("b", intended)) result = await service.poll_once(intended + timedelta(seconds=1)) @@ -644,9 +642,7 @@ async def test_live_settlement_failure_spares_healthy_sibling( def _flaky_settle(self: Any, run_id: str, state: Any, now: datetime) -> None: try: - owned_by_b = ( - self.run_store.get_admission(run_id).schedule_id == "b" - ) + owned_by_b = self.run_store.get_admission(run_id).schedule_id == "b" except KeyError: owned_by_b = False if owned_by_b and not fired["done"]: @@ -707,11 +703,7 @@ async def test_live_torn_settlement_reconciles_genuine_result( ) except KeyError: owned_by_b = False - if ( - kwargs.get("kind") == "completed" - and owned_by_b - and not fired["done"] - ): + if kwargs.get("kind") == "completed" and owned_by_b and not fired["done"]: fired["done"] = True raise RuntimeError("injected history failure") original(self, **kwargs) diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index 65580c72..1dc72087 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -193,7 +193,11 @@ def test_parallel_limits_and_interrupted_slots() -> None: sched.poll(t0) assert sched._task_load() == 1 first = runs.list_runs()[0] + # A waiting interruption holds no task slot: production dispatch + # clears the executing mark when persisting the stopped result, so + # the simulated interruption clears it too. runs.save_run(first.model_copy(update={"status": StoredRunStatus.INTERRUPTED})) + runs.clear_executing(first.id) assert sched._task_load() == 0 sched.poll(t0 + timedelta(minutes=5)) assert len(runs.list_runs()) == 2 @@ -205,6 +209,7 @@ def test_parallel_limits_and_interrupted_slots() -> None: store.save_schedule(sched_model) second = [r for r in runs.list_runs() if r.id != first.id][0] runs.save_run(second.model_copy(update={"status": StoredRunStatus.INTERRUPTED})) + runs.clear_executing(second.id) sched.poll(t0 + timedelta(minutes=15)) assert any( r["kind"] == "skipped-overlap" and "12:15" in str(r["resolved_at"]) @@ -287,6 +292,7 @@ def test_fairness_slow_schedule_not_starved(tmp_path: Path) -> None: ) for run in runs.list_runs(): runs.save_run(run.model_copy(update={"status": StoredRunStatus.COMPLETED})) + runs.clear_executing(run.id) sched.poll(t0 + timedelta(seconds=30)) assert any(r["kind"] == "admitted" for r in _history(store, "slow")) sched.ownership.release() @@ -390,9 +396,8 @@ def test_record_resumed_stopped_result_is_idempotent_and_attributed( tmp_path: Path, ) -> None: """Live resumed results record exactly once with schedule attribution (B3).""" - from wf_api.run_lifecycle import persist_admission - from tests.scheduling.controlled import fixture_environment + from wf_api.run_lifecycle import persist_admission sched, store, runs, sources = _harness(tmp_path, script={"*": "interrupt"}) t0 = ts(2026, 9, 8, 12, 0) diff --git a/tests/scheduling/test_schedule_store.py b/tests/scheduling/test_schedule_store.py index 06057c86..eb5dabfc 100644 --- a/tests/scheduling/test_schedule_store.py +++ b/tests/scheduling/test_schedule_store.py @@ -168,7 +168,9 @@ def test_history_pagination_over_resolved_utc(tmp_path: Path) -> None: assert "|" in str(page["next_cursor"]) assert len(page["occurrences"]) == 2 # type: ignore[arg-type] second = store.list_occurrences( - "a", cursor=page["next_cursor"], limit=2 # type: ignore[arg-type] + "a", + cursor=page["next_cursor"], + limit=2, # type: ignore[arg-type] ) assert second["next_cursor"] is None assert len(second["occurrences"]) == 1 # type: ignore[arg-type] diff --git a/tests/wf_server/test_scheduler_integration.py b/tests/wf_server/test_scheduler_integration.py index eec88f7a..5405aa0f 100644 --- a/tests/wf_server/test_scheduler_integration.py +++ b/tests/wf_server/test_scheduler_integration.py @@ -1081,7 +1081,9 @@ async def test_scheduled_resume_rejected_when_capacity_saturated( _gate_open.clear() root = tmp_path / "store" server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) - await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"]) + await _seed( + server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"] + ) await _seed(server, "gated", "gated.default", _gate_plan("gated"), ["ok"]) store = FileScheduleStore(root) store.create_schedule(_one_shot("asker", "ask1.default", datetime.now(UTC))) @@ -1277,7 +1279,9 @@ async def test_scheduled_resume_records_completion_live(tmp_path: Path) -> None: """ root = tmp_path / "store" server = build_local_static_workflow_server(root) - await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"]) + await _seed( + server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"] + ) store = FileScheduleStore(root) store.create_schedule(_one_shot("asker", "ask1.default", datetime.now(UTC))) service = _scheduler(server, capacity=2) @@ -1398,7 +1402,9 @@ async def test_paused_and_deleted_schedules_keep_resume_history( """Pausing/deleting never suppresses retained resume history (B3).""" root = tmp_path / "store" server = build_local_static_workflow_server(root) - await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"]) + await _seed( + server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"] + ) store = FileScheduleStore(root) store.create_schedule(_one_shot("pausable", "ask1.default", datetime.now(UTC))) store.create_schedule(_one_shot("doomed", "ask1.default", datetime.now(UTC))) @@ -1427,7 +1433,11 @@ async def test_paused_and_deleted_schedules_keep_resume_history( run_id=rid, resume_payload={}, resume_outcome="submitted" ) assert resumed["status"] == "completed" - assert await _kinds(root, "pausable") == ["admitted", "interrupted", "completed"] + assert await _kinds(root, "pausable") == [ + "admitted", + "interrupted", + "completed", + ] assert await _kinds(root, "doomed") == ["admitted", "interrupted", "completed"] finally: await service.stop() diff --git a/tests/wf_transport_rpc_http/test_schedules_rpc.py b/tests/wf_transport_rpc_http/test_schedules_rpc.py index 0fcfb0c1..43824e13 100644 --- a/tests/wf_transport_rpc_http/test_schedules_rpc.py +++ b/tests/wf_transport_rpc_http/test_schedules_rpc.py @@ -306,7 +306,9 @@ async def test_rpc_schedule_occurrences_tied_entries_traverse_once(tmp_path) -> cursor: object = None while True: page = await client.list_schedule_occurrences( - schedule_id="s", cursor=cursor, limit=1 # type: ignore[arg-type] + schedule_id="s", + cursor=cursor, + limit=1, # type: ignore[arg-type] ) assert page["total"] == 2 for row in page["occurrences"]: