mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
test(schedule): make the domain suite fail when the domain breaks
All 102 domain cases were green, and four of them would have stayed green through the exact regression they were named after. Assertions that could not fail ------------------------------ `test_a_claimed_task_is_marked_running_before_dispatch` asserted the lease was released *after* dispatch -- the opposite of the ordering its name and docstring describe. The claim is what makes a task uneditable while it is being dispatched, so the only place that ordering is observable is inside the launch; the launcher double now reads the repository from there. `test_active_statuses_are_exactly_queued_and_running` restated the constant it was checking, so editing the constant edits the assertion with it. Replaced by `is_active` over all six statuses, which also covers `RUNNING` and the two terminal statuses that had none. `test_reuse_thread_with_an_empty_thread_falls_back_to_a_fresh_one` compared its result against `task.thread_id`, which is `None` on the default task -- it asserted "not None" against a method whose body is `str(uuid.uuid4())`. Now asserts the fresh-thread semantics it is named for: a real uuid, distinct per call. `test_a_task_deleted_mid_flight_is_not_an_error` had no assert at all. That path does have observable behaviour: the hook writes the run record before it reads the task, so a task deleted mid-flight must still leave a finalized record and a freed active slot. Contracts stated in a docstring and nowhere else ------------------------------------------------ - a cron overlap must not leave `last_error` behind (service.py:455 branches on it; only the `once` half was covered, so dropping the branch was free) - a failed launch replaces the launch bookkeeping instead of carrying it over the way a skip does -- which is what `last_run_id=None` in `_fail` means for a task that had already run successfully - `SchedulePolicy`'s defaults are the permissive ones, so a deployment that configured no policy cannot have a business constraint invented for it - transitions leave `updated_at` to the repository, rather than becoming a second source of truth for the same column ACTIVE_RUN_STATUSES' promised assertion --------------------------------------- Its docstring says the check that it stays in lockstep with the partial unique index's predicate "lives in a separate test module rather than the domain tests". It did not exist, in that module or any other. `test_scheduled_task_ models.py` now reads both dialect predicates off `__table_args__` and compares the values it extracts against the constant. The domain suite cannot do this -- it is deliberately dependency-free and cannot import an ORM model -- so the new domain case names where the other half of the rule lives. Removed ------- Three duplicates: an `INTERRUPTED -> CANCELLED` case the parametrize directly above it already made, a trailing-Z case identical in path to the aware-run_at case beside it (the `from_primitives` one is the real one, because it parses a string), and an `ensure_launchable == next_after` case whose value another case already asserts outright. Their reasoning moved into comments where it still applies. The second copy of the tautology, in `test_schedule_fakes.py`, goes with it. Verification ------------ A green run is not evidence for this kind of change, so every new or rewritten assertion was checked by mutation -- break the production rule, confirm the guarding test fails. All nine caught, re-run after `ruff format` to confirm the reformat did not soften any of them. Net +207/-31 across four test files; 425 passed, 3 skipped for the schedule and composition suites. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
rayhpeng
parent
34ce7c259b
commit
152e82e25b
@ -13,12 +13,13 @@ through the service's integration tests.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from dataclasses import replace
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.domain.schedule.model import (
|
||||
ACTIVE_RUN_STATUSES,
|
||||
ContextMode,
|
||||
InvalidContextModeError,
|
||||
InvalidScheduleError,
|
||||
@ -101,16 +102,13 @@ class TestScheduleSpecInvariants:
|
||||
assert spec.run_at.astimezone(UTC) == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
|
||||
|
||||
def test_aware_run_at_is_left_alone(self):
|
||||
"""Covers the shape the frontend submits (`zonedLocalToUtcIso`, trailing
|
||||
Z): already aware, so localization must leave it alone rather than
|
||||
reinterpreting it in the task's timezone. Parsing that spelling is
|
||||
`from_primitives`' job and is tested there."""
|
||||
run_at = datetime(2026, 8, 1, 9, 0, tzinfo=UTC)
|
||||
assert ScheduleSpec.once_at(run_at, "Asia/Shanghai").run_at == run_at
|
||||
|
||||
def test_zulu_iso_input_is_accepted_as_the_same_instant(self):
|
||||
"""The shape the frontend actually submits: `zonedLocalToUtcIso` output,
|
||||
which carries a trailing Z. It is already aware, so localization must
|
||||
leave it alone rather than reinterpreting it in the task's timezone."""
|
||||
spec = ScheduleSpec.once_at(datetime.fromisoformat("2026-08-01T01:00:00Z"), "Asia/Shanghai")
|
||||
assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- A2. from_primitives
|
||||
|
||||
@ -249,9 +247,26 @@ class TestEnsureLaunchable:
|
||||
spec = once_spec(after_seconds=3600)
|
||||
assert spec.ensure_launchable(NOW.replace(tzinfo=None), MIN_60) == spec.ensure_launchable(NOW, MIN_60)
|
||||
|
||||
def test_matches_next_after_for_a_valid_once_schedule(self):
|
||||
spec = once_spec(after_seconds=3600)
|
||||
assert spec.ensure_launchable(NOW, MIN_60) == spec.next_after(NOW)
|
||||
|
||||
class TestSchedulePolicyDefaults:
|
||||
""" "Nobody configured a policy" must not invent a business constraint.
|
||||
|
||||
Real values only ever arrive from the composition root (spec.py:22-25). The
|
||||
defaults being the permissive ones is what keeps an unconfigured deployment
|
||||
from silently rejecting schedules a configured one would accept -- so they
|
||||
are pinned here rather than left to whatever the dataclass happens to say.
|
||||
"""
|
||||
|
||||
def test_defaults_are_permissive(self):
|
||||
policy = SchedulePolicy()
|
||||
assert policy.min_once_delay_seconds == 0
|
||||
assert policy.max_concurrent_runs == 1
|
||||
assert policy.lease_seconds == 60
|
||||
|
||||
def test_the_default_delay_floor_admits_an_imminent_once_schedule(self):
|
||||
"""The consequence of the constant above, stated as behaviour: with no
|
||||
configured policy a one-second-away one-shot is legal."""
|
||||
assert once_spec(after_seconds=1).ensure_launchable(NOW, SchedulePolicy()) == NOW + timedelta(seconds=1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- D. value semantics
|
||||
@ -412,6 +427,8 @@ class TestStatusAfterCompletion:
|
||||
("outcome", "expected"),
|
||||
[
|
||||
(RunStatus.SUCCESS, TaskStatus.COMPLETED),
|
||||
# INTERRUPTED is CANCELLED rather than FAILED: a user cancel or a
|
||||
# same-thread takeover carries no execution failure.
|
||||
(RunStatus.INTERRUPTED, TaskStatus.CANCELLED),
|
||||
(RunStatus.FAILED, TaskStatus.FAILED),
|
||||
],
|
||||
@ -419,11 +436,6 @@ class TestStatusAfterCompletion:
|
||||
def test_once_maps_each_terminal_outcome(self, outcome, expected):
|
||||
assert make_task(schedule=once_spec()).status_after_completion(outcome) is expected
|
||||
|
||||
def test_interrupted_is_cancelled_not_failed(self):
|
||||
"""A user cancel or same-thread takeover carries no execution failure."""
|
||||
task = make_task(schedule=once_spec())
|
||||
assert task.status_after_completion(RunStatus.INTERRUPTED) is TaskStatus.CANCELLED
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- G. mutability gate
|
||||
|
||||
@ -454,6 +466,16 @@ class TestMutabilityGate:
|
||||
task.paused()
|
||||
assert task.status is TaskStatus.ENABLED
|
||||
|
||||
def test_transitions_leave_updated_at_to_the_repository(self):
|
||||
"""The repository stamps `updated_at` on write (task.py:54-57). A second
|
||||
clock read here would make the domain impure and a competing source of
|
||||
truth for the same column."""
|
||||
task = make_task(status=TaskStatus.ENABLED)
|
||||
assert task.paused().updated_at == task.updated_at
|
||||
assert task.paused().resumed().updated_at == task.updated_at
|
||||
assert task.with_schedule(cron_spec("0 10 * * *"), now=NOW, policy=NO_DELAY).updated_at == task.updated_at
|
||||
assert task.with_context(ContextMode.FRESH_THREAD_PER_RUN, None).updated_at == task.updated_at
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- H. with_schedule
|
||||
|
||||
@ -525,12 +547,20 @@ class TestResolveExecutionThread:
|
||||
|
||||
def test_reuse_thread_with_an_empty_thread_falls_back_to_a_fresh_one(self):
|
||||
"""Unreachable for new aggregates (__post_init__ forbids it) but rows
|
||||
predating that invariant can still carry this shape."""
|
||||
task = make_task()
|
||||
predating that invariant can still carry this shape.
|
||||
|
||||
Asserting against the fresh-thread semantics rather than against `None`:
|
||||
the fallback has to mint a real, distinct thread per call, which is what
|
||||
the fresh-thread branch means.
|
||||
"""
|
||||
legacy = ScheduledTask.__new__(ScheduledTask)
|
||||
object.__setattr__(legacy, "context_mode", ContextMode.REUSE_THREAD)
|
||||
object.__setattr__(legacy, "thread_id", None)
|
||||
assert ScheduledTask.resolve_execution_thread(legacy) != task.thread_id
|
||||
|
||||
first = ScheduledTask.resolve_execution_thread(legacy)
|
||||
second = ScheduledTask.resolve_execution_thread(legacy)
|
||||
assert uuid.UUID(first), "a real thread id, not a placeholder"
|
||||
assert first != second, "a fresh thread per call, like FRESH_THREAD_PER_RUN"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- K. ScheduledRun
|
||||
@ -555,5 +585,27 @@ class TestScheduledRun:
|
||||
second = ScheduledRun.queued(task_id="task-1", thread_id="t", scheduled_for=NOW, trigger=TriggerKind.MANUAL)
|
||||
assert first.record_id != second.record_id
|
||||
|
||||
def test_active_statuses_are_exactly_queued_and_running(self):
|
||||
assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING)
|
||||
@pytest.mark.parametrize(
|
||||
("status", "active"),
|
||||
[
|
||||
(RunStatus.QUEUED, True),
|
||||
(RunStatus.RUNNING, True),
|
||||
(RunStatus.SUCCESS, False),
|
||||
(RunStatus.FAILED, False),
|
||||
(RunStatus.SKIPPED, False),
|
||||
(RunStatus.INTERRUPTED, False),
|
||||
],
|
||||
)
|
||||
def test_only_queued_and_running_occupy_the_active_slot(self, status, active):
|
||||
"""Stated as behaviour over every status rather than as an equality
|
||||
against ACTIVE_RUN_STATUSES -- restating the constant proves nothing,
|
||||
since editing it would edit the assertion with it. The other half of
|
||||
this rule, that the set matches the partial unique index's predicate,
|
||||
cannot be checked from a dependency-free domain test and lives in
|
||||
test_scheduled_task_models.py.
|
||||
"""
|
||||
run = replace(
|
||||
ScheduledRun.queued(task_id="task-1", thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED),
|
||||
status=status,
|
||||
)
|
||||
assert run.is_active is active
|
||||
|
||||
@ -32,7 +32,6 @@ from app.adapters.schedule.scheduled_run_repository import SqlScheduledRunReposi
|
||||
from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.domain.schedule.model import (
|
||||
ACTIVE_RUN_STATUSES,
|
||||
ActiveRunConflictError,
|
||||
ContextMode,
|
||||
RunStatus,
|
||||
@ -532,9 +531,6 @@ class TestRunStatusWrites:
|
||||
assert by_id[active.record_id].error == "gateway restarted"
|
||||
assert by_id[done.record_id].status is RunStatus.SKIPPED
|
||||
|
||||
async def test_active_statuses_are_exactly_queued_and_running(self):
|
||||
assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING)
|
||||
|
||||
|
||||
class TestLauncherAndThreadLookup:
|
||||
"""Fake-only: these two ports have no SQL implementation -- one starts a
|
||||
|
||||
@ -176,6 +176,11 @@ class TestFullLifecycle:
|
||||
assert after_skip.run_count == 1, "a skip is not an execution"
|
||||
assert after_skip.last_run_id == after_launch.last_run_id, "bookkeeping carried over"
|
||||
assert after_skip.status is TaskStatus.ENABLED
|
||||
# Only a lost one-shot occurrence is worth surfacing; a cron task simply
|
||||
# waits for its next turn, so the skip must not leave a user-visible
|
||||
# error behind (service.py:455). The `once` half is covered by
|
||||
# TestOnceTaskDispatch.test_a_skipped_once_task_is_failed_not_completed.
|
||||
assert after_skip.last_error is None, "a routine cron overlap is not an error to report"
|
||||
|
||||
# -- the first run finally finishes ----------------------------------
|
||||
record = next(r for r in runs.all_runs() if r.status is RunStatus.RUNNING)
|
||||
@ -271,6 +276,47 @@ class TestDispatchOutcomes:
|
||||
assert "provider exploded" in result.error
|
||||
assert runs.all_runs()[0].status is RunStatus.FAILED
|
||||
|
||||
async def test_a_failed_launch_replaces_the_previous_run_bookkeeping(self):
|
||||
"""A failed launch is still an execution *attempt*, so it overwrites the
|
||||
launch bookkeeping rather than carrying it over the way a skip does
|
||||
(contrast `_finalize_skip`, which passes the current values back).
|
||||
|
||||
`record_launch` assigns every field unconditionally, so this is what
|
||||
`last_run_id=None` in `_fail` actually means for a task that had already
|
||||
run successfully: the id of the previous, unrelated run must not be left
|
||||
pointing at a launch that never happened.
|
||||
"""
|
||||
launcher = FakeRunLauncher()
|
||||
runs = InMemoryScheduledRunRepository()
|
||||
service = make_service(runs=runs, launcher=launcher)
|
||||
task = await create_cron_task(service)
|
||||
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
|
||||
succeeded = await service.get_task(task.task_id, user_id="user-1")
|
||||
assert succeeded.last_run_id == "run-1", "precondition: a successful launch was recorded"
|
||||
|
||||
# The slot has to be free, or the next dispatch is an overlap instead.
|
||||
await service.handle_run_completion(
|
||||
RunOutcome(
|
||||
task_id=task.task_id,
|
||||
record_id=runs.all_runs()[0].record_id,
|
||||
run_id="run-1",
|
||||
user_id="user-1",
|
||||
status=RunStatus.SUCCESS,
|
||||
error=None,
|
||||
),
|
||||
now=NOW,
|
||||
)
|
||||
launcher.fail_with = LaunchFailedError("provider exploded")
|
||||
reloaded = await service.get_task(task.task_id, user_id="user-1")
|
||||
|
||||
await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED)
|
||||
|
||||
after = await service.get_task(task.task_id, user_id="user-1")
|
||||
assert after.last_run_id is None, "no run started, so no run id to point at"
|
||||
assert after.last_error == "provider exploded"
|
||||
assert after.last_run_at == NOW, "the attempt itself is timestamped"
|
||||
assert after.run_count == 1, "a failed launch is not a completed execution"
|
||||
|
||||
async def test_busy_thread_on_the_scheduled_path_degrades_to_a_skip(self):
|
||||
launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
|
||||
service = make_service(launcher=launcher)
|
||||
@ -412,16 +458,34 @@ class TestRunOnceBudget:
|
||||
assert len(results) == 2
|
||||
assert all(r.outcome is DispatchOutcome.LAUNCHED for r in results)
|
||||
|
||||
async def test_a_claimed_task_is_marked_running_before_dispatch(self):
|
||||
async def test_the_claim_is_held_across_the_launch_and_released_after(self):
|
||||
"""Claiming is what makes the task uneditable while it is being
|
||||
dispatched, so the claim must land before the launch."""
|
||||
dispatched, so the claim has to be live *at the moment of the launch* --
|
||||
not merely taken at some point and released by the end.
|
||||
|
||||
The launch is the only place that ordering is observable, so the
|
||||
launcher double reads the repository from inside `launch`. Asserting
|
||||
only on the end state would pass even if the claim were taken after the
|
||||
run had already started.
|
||||
"""
|
||||
tasks = InMemoryScheduledTaskRepository()
|
||||
service = make_service(tasks=tasks)
|
||||
observed: dict = {}
|
||||
|
||||
class _ObservingLauncher(FakeRunLauncher):
|
||||
async def launch(self, **kwargs):
|
||||
task_id = kwargs["metadata"]["scheduled_task_id"]
|
||||
observed["lease"] = tasks.lease_of(task_id)
|
||||
observed["status"] = (await tasks.get(task_id, user_id="user-1")).status
|
||||
return await super().launch(**kwargs)
|
||||
|
||||
service = make_service(tasks=tasks, launcher=_ObservingLauncher())
|
||||
task = await create_cron_task(service)
|
||||
|
||||
await service.run_once(now=task.next_run_at + timedelta(seconds=1))
|
||||
|
||||
assert tasks.lease_of(task.task_id) == (None, None), "the claim is released after dispatch"
|
||||
assert observed["status"] is TaskStatus.RUNNING, "the claim marks the task running before the launch"
|
||||
assert observed["lease"][1] is not None, "and the lease is still held while the run starts"
|
||||
assert tasks.lease_of(task.task_id) == (None, None), "released only once the dispatch is done"
|
||||
|
||||
|
||||
# ==================================================================== completion
|
||||
@ -493,8 +557,28 @@ class TestRunCompletion:
|
||||
assert after.status is TaskStatus.ENABLED, "the schedule outlives any single run"
|
||||
assert after.last_error == "boom"
|
||||
|
||||
async def test_a_task_deleted_mid_flight_is_not_an_error(self):
|
||||
service, task, record = await self._launched_once_task()
|
||||
async def test_a_task_deleted_mid_flight_still_finalizes_its_run(self):
|
||||
"""A task deleted while its run was in flight has nothing to write back,
|
||||
and that is not an error -- but the *run* record must still be closed.
|
||||
|
||||
The hook writes the record first and only then reads the task, so
|
||||
asserting on the record is what keeps that ordering honest; a test that
|
||||
only checked "no exception" would stay green if the record write were
|
||||
dropped entirely.
|
||||
"""
|
||||
runs = InMemoryScheduledRunRepository()
|
||||
service = make_service(runs=runs)
|
||||
task = await service.create_task(
|
||||
user_id="user-1",
|
||||
title="one shot",
|
||||
prompt="go",
|
||||
schedule=once_spec(),
|
||||
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
|
||||
thread_id=None,
|
||||
now=NOW,
|
||||
)
|
||||
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
|
||||
record = runs.all_runs()[0]
|
||||
await service.delete_task(task.task_id, user_id="user-1")
|
||||
|
||||
await service.handle_run_completion(
|
||||
@ -509,6 +593,11 @@ class TestRunCompletion:
|
||||
now=NOW,
|
||||
)
|
||||
|
||||
stored = runs.all_runs()[0]
|
||||
assert stored.status is RunStatus.SUCCESS, "the execution record is finalized regardless"
|
||||
assert stored.finished_at == NOW
|
||||
assert await runs.count_active() == 0, "the active slot is freed"
|
||||
|
||||
async def test_a_cron_task_survives_a_launch_write_landing_mid_completion(self):
|
||||
"""The interleaving `dispatch_task` already admits is possible.
|
||||
|
||||
|
||||
@ -1,4 +1,9 @@
|
||||
import re
|
||||
|
||||
from sqlalchemy import Index
|
||||
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.domain.schedule.model import ACTIVE_RUN_STATUSES
|
||||
from deerflow.persistence.models import ScheduledTaskRow, ScheduledTaskRunRow
|
||||
|
||||
|
||||
@ -17,3 +22,37 @@ def test_app_config_exposes_scheduler_section():
|
||||
def test_scheduled_task_models_registered():
|
||||
assert ScheduledTaskRow.__tablename__ == "scheduled_tasks"
|
||||
assert ScheduledTaskRunRow.__tablename__ == "scheduled_task_runs"
|
||||
|
||||
|
||||
def _active_run_index() -> Index:
|
||||
return next(arg for arg in ScheduledTaskRunRow.__table_args__ if isinstance(arg, Index) and arg.name == "uq_scheduled_task_run_active")
|
||||
|
||||
|
||||
def test_active_run_index_arbitrates_one_active_run_per_task():
|
||||
"""The index is the atomic arbiter of the overlap rule, so its shape is a
|
||||
contract, not a detail: unique, keyed on `task_id` alone."""
|
||||
index = _active_run_index()
|
||||
assert index.unique is True
|
||||
assert [column.name for column in index.expressions] == ["task_id"]
|
||||
|
||||
|
||||
def test_active_run_index_predicate_matches_the_domain_constant():
|
||||
"""`ACTIVE_RUN_STATUSES` and this predicate must stay in lockstep.
|
||||
|
||||
The domain's fast path (`has_active`) and the index disagree the moment
|
||||
they drift, which silently decouples the overlap check from its arbiter.
|
||||
The domain tests cannot assert this -- they are deliberately
|
||||
dependency-free and cannot import an ORM model -- so the assertion the
|
||||
`ACTIVE_RUN_STATUSES` docstring promises lives here.
|
||||
|
||||
Both dialect predicates are checked: `create_all` renders the SQLite one
|
||||
and production renders the Postgres one, so a drift in either is real.
|
||||
"""
|
||||
index = _active_run_index()
|
||||
expected = {str(status) for status in ACTIVE_RUN_STATUSES}
|
||||
assert expected == {"queued", "running"}, "domain constant changed -- update the ORM predicates below"
|
||||
|
||||
predicates = {key: str(value) for key, value in index.dialect_kwargs.items() if key.endswith("_where")}
|
||||
assert set(predicates) == {"sqlite_where", "postgresql_where"}, "a dialect lost its partial-index predicate"
|
||||
for dialect, predicate in predicates.items():
|
||||
assert set(re.findall(r"'([^']+)'", predicate)) == expected, f"{dialect} predicate drifted from ACTIVE_RUN_STATUSES"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user