mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
7852421c68
Adds the three remaining adapters plus the poller. Nothing is wired yet -- the composition root is the next commit -- so this is additive and the legacy `app/scheduler/service.py` still serves production. `run_launcher.py` is the pivot of the whole slice. The Gateway signals a busy thread two ways -- `ConflictError` from the run manager, or an `HTTPException(409)` from the route-level path -- which is why the legacy scheduler service imported fastapi to tell them apart. Both are one domain fact, and saying so here is what lets that import disappear without the busy/failed distinction disappearing with it. Everything else becomes `LaunchFailedError`, because the port promises the domain that nothing but its two errors escapes. `CancelledError` is deliberately not caught: shutdown is control flow, not a launch outcome. `thread_lookup.py` narrows `ThreadMetaStore` to the one question this context asks. `require_existing=True` is load-bearing -- the store's default treats an absent row as accessible, which is right for a thread not yet written and wrong for binding a task to it. Both inherit their port explicitly, matching every other adapter in the codebase including feedback's own anti-corruption layer, and both carry the TODO naming the published contract that would replace them once the upstream context has been through a slice of its own. `run_outcome_mapping.py` implements no port: it is the inbound translation the composition root will install on the completion hook, and it owns the filtering the legacy hook did inline. Returning None means "this run is none of the schedule context's business", so the service is simply never called and needs no guard clauses. `poller.py` keeps the two behaviours the legacy loop got right: a failing poll must not end the loop (one transient "database is locked" used to stop scheduling for the rest of the process life), and reconciliation must not block startup. One deliberate behaviour change: the legacy `start()` swept stale runs and stuck once-tasks under separate try/excepts, so the first failing did not stop the second. `reconcile_on_startup` is one call that lets failures propagate -- the domain's position is that fatality is the caller's policy -- so the poller's single except means a failed first sweep now skips the second. Both end up logged and non-fatal, as before. Tests: 50 new cases across the four modules, each port method called and asserted on its return value. That is not decoration: inheriting a Protocol means a misspelled method silently inherits its `...` body and returns None, so the suite was verified by mutation -- renaming `launch` and `exists_for_user` turns 16 and 6 cases red respectively.
72 lines
3.1 KiB
Python
72 lines
3.1 KiB
Python
"""Contract tests for the thread-lookup anti-corruption layer.
|
|
|
|
The port asks one question -- "does this thread exist AND may this user use
|
|
it?" -- and deliberately answers both halves with a single bool, so a caller
|
|
cannot probe for the existence of threads they cannot see. These tests pin that
|
|
the adapter does not accidentally widen it back into two answers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.adapters.schedule.thread_lookup import ThreadStoreThreadLookup
|
|
|
|
|
|
class _RecordingThreadStore:
|
|
"""Stands in for `ThreadMetaStore`, recording how it was asked."""
|
|
|
|
def __init__(self, owners: dict[str, str] | None = None) -> None:
|
|
self._owners = dict(owners or {})
|
|
self.calls: list[tuple[str, str, bool]] = []
|
|
|
|
async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
|
|
self.calls.append((thread_id, user_id, require_existing))
|
|
owner = self._owners.get(thread_id)
|
|
if owner is None:
|
|
# Mirrors the real store: absent rows pass a non-strict check.
|
|
return not require_existing
|
|
return owner == user_id
|
|
|
|
|
|
class TestTheOneQuestion:
|
|
@pytest.mark.asyncio
|
|
async def test_an_owned_thread_exists_for_its_user(self):
|
|
lookup = ThreadStoreThreadLookup(_RecordingThreadStore({"thread-1": "user-1"}))
|
|
assert await lookup.exists_for_user("thread-1", "user-1") is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_someone_elses_thread_does_not(self):
|
|
lookup = ThreadStoreThreadLookup(_RecordingThreadStore({"thread-1": "user-1"}))
|
|
assert await lookup.exists_for_user("thread-1", "user-2") is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_missing_thread_does_not(self):
|
|
"""This is the half `require_existing` buys: without it the store
|
|
treats an absent row as accessible, and a task could be bound to a
|
|
thread that does not exist."""
|
|
lookup = ThreadStoreThreadLookup(_RecordingThreadStore())
|
|
assert await lookup.exists_for_user("thread-nope", "user-1") is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_and_forbidden_are_indistinguishable(self):
|
|
lookup = ThreadStoreThreadLookup(_RecordingThreadStore({"thread-1": "user-1"}))
|
|
missing = await lookup.exists_for_user("thread-nope", "user-2")
|
|
forbidden = await lookup.exists_for_user("thread-1", "user-2")
|
|
assert missing == forbidden is False
|
|
|
|
|
|
class TestHowTheStoreIsAsked:
|
|
@pytest.mark.asyncio
|
|
async def test_require_existing_is_always_set(self):
|
|
store = _RecordingThreadStore({"thread-1": "user-1"})
|
|
await ThreadStoreThreadLookup(store).exists_for_user("thread-1", "user-1")
|
|
assert store.calls == [("thread-1", "user-1", True)]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_result_is_a_real_bool(self):
|
|
"""The port is typed `bool`; a truthy row object leaking through would
|
|
satisfy the domain's `if` and still be the wrong contract."""
|
|
store = _RecordingThreadStore({"thread-1": "user-1"})
|
|
assert await ThreadStoreThreadLookup(store).exists_for_user("thread-1", "user-1") is True
|