mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-16 09:38:41 +00:00
cb49dd67dc
Group them by bounded context instead of by technology, one file per
port, and align the directory name with the AWS Prescriptive Guidance
layout (entrypoints / domain-with-ports / adapters).
app/infra/persistence/feedback.py
-> app/adapters/feedback/feedback_repository.py owned persistence
-> app/adapters/feedback/run_lookup.py anti-corruption layer
`persistence/` promised a technology-first classification that its own
contents contradicted: RunStoreRunLookup lived there while its docstring
said "no new SQL". Splitting per port makes that distinction structural.
SqlFeedbackRepository and _tz_aware move unchanged -- verified by
comparing their AST against the original rather than by eye. run_lookup.py
additionally gains a RunStore annotation behind TYPE_CHECKING (the module
is imported lazily by the composition root, so this keeps the runtime
import cost at zero), a docstring stating that this context owns no table
and writes no SQL against it, and a TODO recording the condition under
which the body is replaced: when the run context publishes a contract of
its own, the RunLookup port itself does not move.
Each module docstring opens with a fixed marker so the two kinds of
secondary adapter stay greppable:
grep -rl "anti-corruption layer" app/adapters/
Filenames deliberately carry no sql_ / acl_ prefix: a prefix encodes an
implementation property, so switching storage would force a rename even
though the port -- and therefore the import path -- has not changed. The
class name already carries it. A prefix earns its place once one port has
several production implementations, which is not yet the case here.
app/infra/ held nothing else and is removed.
182 lines
7.5 KiB
Python
182 lines
7.5 KiB
Python
"""Contract tests for the FeedbackRepository port.
|
|
|
|
Every implementation (SQL adapter, in-memory fake) runs the same suite, so
|
|
port semantics and implementations cannot drift apart. The in-memory fake
|
|
lives in ``feedback_fakes`` because service-level tests reuse it as a fast
|
|
zero-IO double.
|
|
"""
|
|
|
|
import pytest
|
|
from feedback_fakes import InMemoryFeedbackRepository
|
|
|
|
from deerflow.domain.feedback import Feedback
|
|
from deerflow.domain.feedback.ports import FeedbackRepository
|
|
|
|
|
|
class FeedbackRepositoryContract:
|
|
"""Shared port-semantics suite; subclasses provide a ``repo`` fixture."""
|
|
|
|
@pytest.mark.anyio
|
|
async def test_satisfies_port(self, repo):
|
|
assert isinstance(repo, FeedbackRepository)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_save_creates(self, repo):
|
|
fb = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
assert fb.feedback_id
|
|
assert fb.rating == 1
|
|
assert fb.created_at.tzinfo is not None
|
|
|
|
@pytest.mark.anyio
|
|
async def test_save_updates_keeping_identity(self, repo):
|
|
first = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
second = await repo.save(
|
|
Feedback.create(
|
|
run_id="r1",
|
|
thread_id="t1",
|
|
rating=-1,
|
|
user_id="u1",
|
|
comment="changed my mind",
|
|
tags=["incorrect", "slow"],
|
|
)
|
|
)
|
|
assert second.feedback_id == first.feedback_id
|
|
assert second.rating == -1
|
|
assert second.comment == "changed my mind"
|
|
assert second.tags == ("incorrect", "slow")
|
|
|
|
@pytest.mark.anyio
|
|
async def test_save_different_users_separate(self, repo):
|
|
a = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
b = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=-1, user_id="u2"))
|
|
assert a.feedback_id != b.feedback_id
|
|
|
|
@pytest.mark.anyio
|
|
async def test_latest_per_run_in_thread_scopes_thread_and_owner(self, repo):
|
|
await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
await repo.save(Feedback.create(run_id="r2", thread_id="t1", rating=-1, user_id="u1"))
|
|
await repo.save(Feedback.create(run_id="r3", thread_id="t2", rating=1, user_id="u1"))
|
|
await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=-1, user_id="u2"))
|
|
|
|
grouped = await repo.latest_per_run_in_thread("t1", user_id="u1")
|
|
|
|
assert set(grouped) == {"r1", "r2"}
|
|
assert grouped["r1"].rating == 1
|
|
assert grouped["r2"].rating == -1
|
|
|
|
@pytest.mark.anyio
|
|
async def test_latest_per_run_in_thread_empty(self, repo):
|
|
assert await repo.latest_per_run_in_thread("t1", user_id="u1") == {}
|
|
|
|
@pytest.mark.anyio
|
|
async def test_latest_for_runs_scopes_selection(self, repo):
|
|
await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
await repo.save(Feedback.create(run_id="r2", thread_id="t1", rating=-1, user_id="u1"))
|
|
await repo.save(Feedback.create(run_id="r3", thread_id="t1", rating=1, user_id="u1"))
|
|
|
|
grouped = await repo.latest_for_runs("t1", {"r1", "r2"}, user_id="u1")
|
|
|
|
assert set(grouped) == {"r1", "r2"}
|
|
|
|
@pytest.mark.anyio
|
|
async def test_latest_for_runs_empty_set(self, repo):
|
|
assert await repo.latest_for_runs("t1", set(), user_id="u1") == {}
|
|
|
|
@pytest.mark.anyio
|
|
async def test_remove_for_run(self, repo):
|
|
await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
assert await repo.remove_for_run("t1", "r1", user_id="u1") is True
|
|
assert await repo.latest_per_run_in_thread("t1", user_id="u1") == {}
|
|
|
|
@pytest.mark.anyio
|
|
async def test_remove_for_run_nonexistent(self, repo):
|
|
assert await repo.remove_for_run("t1", "r1", user_id="u1") is False
|
|
|
|
@pytest.mark.anyio
|
|
async def test_remove_for_run_other_user_denied(self, repo):
|
|
await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
|
|
assert await repo.remove_for_run("t1", "r1", user_id="u2") is False
|
|
assert set(await repo.latest_per_run_in_thread("t1", user_id="u1")) == {"r1"}
|
|
|
|
|
|
class TestSqlFeedbackRepository(FeedbackRepositoryContract):
|
|
@pytest.fixture
|
|
async def repo(self, tmp_path):
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
import deerflow.persistence.feedback.model # noqa: F401 (register FeedbackRow)
|
|
from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
|
|
from deerflow.persistence.base import Base
|
|
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield SqlFeedbackRepository(async_sessionmaker(engine, expire_on_commit=False))
|
|
await engine.dispose()
|
|
|
|
|
|
class TestInMemoryFeedbackRepository(FeedbackRepositoryContract):
|
|
@pytest.fixture
|
|
def repo(self):
|
|
return InMemoryFeedbackRepository()
|
|
|
|
|
|
# -- Follow-up association (unrelated to the feedback repository) --
|
|
|
|
|
|
class TestFollowUpAssociation:
|
|
@pytest.mark.anyio
|
|
async def test_run_records_follow_up_via_memory_store(self):
|
|
"""MemoryRunStore stores follow_up_to_run_id in kwargs."""
|
|
from deerflow.runtime.runs.store.memory import MemoryRunStore
|
|
|
|
store = MemoryRunStore()
|
|
await store.put("r1", thread_id="t1", status="success")
|
|
# MemoryRunStore doesn't have follow_up_to_run_id as a top-level param,
|
|
# but it can be passed via metadata
|
|
await store.put("r2", thread_id="t1", metadata={"follow_up_to_run_id": "r1"})
|
|
run = await store.get("r2")
|
|
assert run["metadata"]["follow_up_to_run_id"] == "r1"
|
|
|
|
@pytest.mark.anyio
|
|
async def test_human_message_has_follow_up_metadata(self):
|
|
"""human_message event metadata includes follow_up_to_run_id."""
|
|
from deerflow.runtime.events.store.memory import MemoryRunEventStore
|
|
|
|
event_store = MemoryRunEventStore()
|
|
await event_store.put(
|
|
thread_id="t1",
|
|
run_id="r2",
|
|
event_type="human_message",
|
|
category="message",
|
|
content="Tell me more about that",
|
|
metadata={"follow_up_to_run_id": "r1"},
|
|
)
|
|
messages = await event_store.list_messages("t1")
|
|
assert messages[0]["metadata"]["follow_up_to_run_id"] == "r1"
|
|
|
|
@pytest.mark.anyio
|
|
async def test_follow_up_auto_detection_logic(self):
|
|
"""Simulate the auto-detection: latest successful run becomes follow_up_to."""
|
|
from deerflow.runtime.runs.store.memory import MemoryRunStore
|
|
|
|
store = MemoryRunStore()
|
|
await store.put("r1", thread_id="t1", status="success")
|
|
await store.put("r2", thread_id="t1", status="error")
|
|
|
|
# Auto-detect: list_by_thread returns newest first
|
|
recent = await store.list_by_thread("t1", limit=1)
|
|
follow_up = None
|
|
if recent and recent[0].get("status") == "success":
|
|
follow_up = recent[0]["run_id"]
|
|
# r2 (error) is newest, so no follow_up detected
|
|
assert follow_up is None
|
|
|
|
# Now add a successful run
|
|
await store.put("r3", thread_id="t1", status="success")
|
|
recent = await store.list_by_thread("t1", limit=1)
|
|
follow_up = None
|
|
if recent and recent[0].get("status") == "success":
|
|
follow_up = recent[0]["run_id"]
|
|
assert follow_up == "r3"
|