mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-27 00:17:53 +00:00
refactor(harness): drop legacy feedback repository
The ORM row stays behind (single alembic chain until the infrastructure move); everything else is served by the domain slice. Repository tests become a contract suite that runs the same cases against the SQL adapter and an in-memory fake; ownership and timezone tests use explicit user_id instead of the AUTO-sentinel context.
This commit is contained in:
rayhpeng
parent
70613f8996
commit
808e31e2ee
@ -1,6 +1,11 @@
|
||||
"""Feedback persistence — ORM and SQL repository."""
|
||||
"""Feedback persistence — ORM row only.
|
||||
|
||||
The legacy SQL repository was replaced by the hexagonal slice:
|
||||
port `deerflow.domain.feedback.ports.FeedbackRepository`, adapter
|
||||
`app.infra.persistence.feedback.SqlFeedbackRepository`. The ORM row stays
|
||||
here until PR-N moves engine/models/migrations into app/infra.
|
||||
"""
|
||||
|
||||
from deerflow.persistence.feedback.model import FeedbackRow
|
||||
from deerflow.persistence.feedback.sql import FeedbackRepository
|
||||
|
||||
__all__ = ["FeedbackRepository", "FeedbackRow"]
|
||||
__all__ = ["FeedbackRow"]
|
||||
|
||||
@ -1,240 +0,0 @@
|
||||
"""SQLAlchemy-backed feedback storage.
|
||||
|
||||
Each method acquires its own short-lived session.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import case, func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from deerflow.persistence.feedback.model import FeedbackRow
|
||||
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
|
||||
from deerflow.utils.time import coerce_iso
|
||||
|
||||
|
||||
class FeedbackRepository:
|
||||
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
||||
self._sf = session_factory
|
||||
|
||||
@staticmethod
|
||||
def _row_to_dict(row: FeedbackRow) -> dict:
|
||||
d = row.to_dict()
|
||||
val = d.get("created_at")
|
||||
if isinstance(val, datetime):
|
||||
# SQLite drops tzinfo on read; normalize via ``coerce_iso`` so output is always tz-aware.
|
||||
d["created_at"] = coerce_iso(val)
|
||||
return d
|
||||
|
||||
async def create(
|
||||
self,
|
||||
*,
|
||||
run_id: str,
|
||||
thread_id: str,
|
||||
rating: int,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
message_id: str | None = None,
|
||||
comment: str | None = None,
|
||||
) -> dict:
|
||||
"""Create a feedback record. rating must be +1 or -1."""
|
||||
if rating not in (1, -1):
|
||||
raise ValueError(f"rating must be +1 or -1, got {rating}")
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.create")
|
||||
row = FeedbackRow(
|
||||
feedback_id=str(uuid.uuid4()),
|
||||
run_id=run_id,
|
||||
thread_id=thread_id,
|
||||
user_id=resolved_user_id,
|
||||
message_id=message_id,
|
||||
rating=rating,
|
||||
comment=comment,
|
||||
created_at=datetime.now(UTC),
|
||||
)
|
||||
async with self._sf() as session:
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
await session.refresh(row)
|
||||
return self._row_to_dict(row)
|
||||
|
||||
async def get(
|
||||
self,
|
||||
feedback_id: str,
|
||||
*,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> dict | None:
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.get")
|
||||
async with self._sf() as session:
|
||||
row = await session.get(FeedbackRow, feedback_id)
|
||||
if row is None:
|
||||
return None
|
||||
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
||||
return None
|
||||
return self._row_to_dict(row)
|
||||
|
||||
async def list_by_run(
|
||||
self,
|
||||
thread_id: str,
|
||||
run_id: str,
|
||||
*,
|
||||
limit: int = 100,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> list[dict]:
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_run")
|
||||
stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id, FeedbackRow.run_id == run_id)
|
||||
if resolved_user_id is not None:
|
||||
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
||||
stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
|
||||
async with self._sf() as session:
|
||||
result = await session.execute(stmt)
|
||||
return [self._row_to_dict(r) for r in result.scalars()]
|
||||
|
||||
async def list_by_thread(
|
||||
self,
|
||||
thread_id: str,
|
||||
*,
|
||||
limit: int = 100,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> list[dict]:
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_thread")
|
||||
stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
|
||||
if resolved_user_id is not None:
|
||||
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
||||
stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
|
||||
async with self._sf() as session:
|
||||
result = await session.execute(stmt)
|
||||
return [self._row_to_dict(r) for r in result.scalars()]
|
||||
|
||||
async def delete(
|
||||
self,
|
||||
feedback_id: str,
|
||||
*,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> bool:
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.delete")
|
||||
async with self._sf() as session:
|
||||
row = await session.get(FeedbackRow, feedback_id)
|
||||
if row is None:
|
||||
return False
|
||||
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
||||
return False
|
||||
await session.delete(row)
|
||||
await session.commit()
|
||||
return True
|
||||
|
||||
async def upsert(
|
||||
self,
|
||||
*,
|
||||
run_id: str,
|
||||
thread_id: str,
|
||||
rating: int,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
comment: str | None = None,
|
||||
) -> dict:
|
||||
"""Create or update feedback for (thread_id, run_id, user_id). rating must be +1 or -1."""
|
||||
if rating not in (1, -1):
|
||||
raise ValueError(f"rating must be +1 or -1, got {rating}")
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.upsert")
|
||||
async with self._sf() as session:
|
||||
stmt = select(FeedbackRow).where(
|
||||
FeedbackRow.thread_id == thread_id,
|
||||
FeedbackRow.run_id == run_id,
|
||||
FeedbackRow.user_id == resolved_user_id,
|
||||
)
|
||||
result = await session.execute(stmt)
|
||||
row = result.scalar_one_or_none()
|
||||
if row is not None:
|
||||
row.rating = rating
|
||||
row.comment = comment
|
||||
row.created_at = datetime.now(UTC)
|
||||
else:
|
||||
row = FeedbackRow(
|
||||
feedback_id=str(uuid.uuid4()),
|
||||
run_id=run_id,
|
||||
thread_id=thread_id,
|
||||
user_id=resolved_user_id,
|
||||
rating=rating,
|
||||
comment=comment,
|
||||
created_at=datetime.now(UTC),
|
||||
)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
await session.refresh(row)
|
||||
return self._row_to_dict(row)
|
||||
|
||||
async def delete_by_run(
|
||||
self,
|
||||
*,
|
||||
thread_id: str,
|
||||
run_id: str,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> bool:
|
||||
"""Delete the current user's feedback for a run. Returns True if a record was deleted."""
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.delete_by_run")
|
||||
async with self._sf() as session:
|
||||
stmt = select(FeedbackRow).where(
|
||||
FeedbackRow.thread_id == thread_id,
|
||||
FeedbackRow.run_id == run_id,
|
||||
FeedbackRow.user_id == resolved_user_id,
|
||||
)
|
||||
result = await session.execute(stmt)
|
||||
row = result.scalar_one_or_none()
|
||||
if row is None:
|
||||
return False
|
||||
await session.delete(row)
|
||||
await session.commit()
|
||||
return True
|
||||
|
||||
async def list_by_thread_grouped(
|
||||
self,
|
||||
thread_id: str,
|
||||
*,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> dict[str, dict]:
|
||||
"""Return feedback grouped by run_id for a thread: {run_id: feedback_dict}."""
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_thread_grouped")
|
||||
stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
|
||||
if resolved_user_id is not None:
|
||||
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
||||
async with self._sf() as session:
|
||||
result = await session.execute(stmt)
|
||||
return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
|
||||
|
||||
async def list_by_run_ids(
|
||||
self,
|
||||
thread_id: str,
|
||||
run_ids: set[str],
|
||||
*,
|
||||
user_id: str | None | _AutoSentinel = AUTO,
|
||||
) -> dict[str, dict]:
|
||||
"""Return feedback for only the selected runs in one thread."""
|
||||
if not run_ids:
|
||||
return {}
|
||||
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_run_ids")
|
||||
stmt = select(FeedbackRow).where(
|
||||
FeedbackRow.thread_id == thread_id,
|
||||
FeedbackRow.run_id.in_(run_ids),
|
||||
)
|
||||
if resolved_user_id is not None:
|
||||
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
||||
async with self._sf() as session:
|
||||
result = await session.execute(stmt)
|
||||
return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
|
||||
|
||||
async def aggregate_by_run(self, thread_id: str, run_id: str) -> dict:
|
||||
"""Aggregate feedback stats for a run using database-side counting."""
|
||||
stmt = select(
|
||||
func.count().label("total"),
|
||||
func.coalesce(func.sum(case((FeedbackRow.rating == 1, 1), else_=0)), 0).label("positive"),
|
||||
func.coalesce(func.sum(case((FeedbackRow.rating == -1, 1), else_=0)), 0).label("negative"),
|
||||
).where(FeedbackRow.thread_id == thread_id, FeedbackRow.run_id == run_id)
|
||||
async with self._sf() as session:
|
||||
row = (await session.execute(stmt)).one()
|
||||
return {
|
||||
"run_id": run_id,
|
||||
"total": row.total,
|
||||
"positive": row.positive,
|
||||
"negative": row.negative,
|
||||
}
|
||||
@ -1,258 +1,154 @@
|
||||
"""Tests for FeedbackRepository and follow-up association.
|
||||
"""Contract tests for the FeedbackRepository port.
|
||||
|
||||
Uses temp SQLite DB for ORM tests.
|
||||
Every implementation (SQL adapter, in-memory fake) runs the same suite, so
|
||||
port semantics and implementations cannot drift apart. The in-memory fake
|
||||
also serves service-level tests as a fast zero-IO double.
|
||||
"""
|
||||
|
||||
from dataclasses import replace
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.persistence.feedback import FeedbackRepository
|
||||
from deerflow.domain.feedback import Feedback
|
||||
from deerflow.domain.feedback.ports import FeedbackRepository
|
||||
|
||||
|
||||
async def _make_feedback_repo(tmp_path):
|
||||
from deerflow.persistence.engine import get_session_factory, init_engine
|
||||
class InMemoryFeedbackRepository:
|
||||
"""Zero-IO fake keyed by aggregate identity (thread, run, user)."""
|
||||
|
||||
url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}"
|
||||
await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
|
||||
return FeedbackRepository(get_session_factory())
|
||||
def __init__(self) -> None:
|
||||
self._rows: dict[tuple[str, str, str | None], Feedback] = {}
|
||||
|
||||
async def save(self, feedback: Feedback) -> Feedback:
|
||||
key = (feedback.thread_id, feedback.run_id, feedback.user_id)
|
||||
existing = self._rows.get(key)
|
||||
if existing is not None:
|
||||
feedback = replace(feedback, feedback_id=existing.feedback_id)
|
||||
self._rows[key] = feedback
|
||||
return feedback
|
||||
|
||||
async def latest_per_run_in_thread(self, thread_id: str, *, user_id: str | None) -> dict[str, Feedback]:
|
||||
return {fb.run_id: fb for (thread, _run, user), fb in self._rows.items() if thread == thread_id and (user_id is None or user == user_id)}
|
||||
|
||||
async def latest_for_runs(self, thread_id: str, run_ids: set[str], *, user_id: str | None) -> dict[str, Feedback]:
|
||||
if not run_ids:
|
||||
return {}
|
||||
per_run = await self.latest_per_run_in_thread(thread_id, user_id=user_id)
|
||||
return {run_id: fb for run_id, fb in per_run.items() if run_id in run_ids}
|
||||
|
||||
async def remove_for_run(self, thread_id: str, run_id: str, *, user_id: str | None) -> bool:
|
||||
return self._rows.pop((thread_id, run_id, user_id), None) is not None
|
||||
|
||||
|
||||
async def _cleanup():
|
||||
from deerflow.persistence.engine import close_engine
|
||||
|
||||
await close_engine()
|
||||
|
||||
|
||||
# -- FeedbackRepository --
|
||||
|
||||
|
||||
class TestFeedbackRepository:
|
||||
@pytest.mark.anyio
|
||||
async def test_create_positive(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
record = await repo.create(run_id="r1", thread_id="t1", rating=1)
|
||||
assert record["feedback_id"]
|
||||
assert record["rating"] == 1
|
||||
assert record["run_id"] == "r1"
|
||||
assert record["thread_id"] == "t1"
|
||||
assert "created_at" in record
|
||||
await _cleanup()
|
||||
class FeedbackRepositoryContract:
|
||||
"""Shared port-semantics suite; subclasses provide a ``repo`` fixture."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_negative_with_comment(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
record = await repo.create(
|
||||
run_id="r1",
|
||||
thread_id="t1",
|
||||
rating=-1,
|
||||
comment="Response was inaccurate",
|
||||
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 record["rating"] == -1
|
||||
assert record["comment"] == "Response was inaccurate"
|
||||
await _cleanup()
|
||||
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_create_with_message_id(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
record = await repo.create(run_id="r1", thread_id="t1", rating=1, message_id="msg-42")
|
||||
assert record["message_id"] == "msg-42"
|
||||
await _cleanup()
|
||||
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_create_with_owner(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
record = await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-1")
|
||||
assert record["user_id"] == "user-1"
|
||||
await _cleanup()
|
||||
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"))
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_invalid_rating_zero(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
with pytest.raises(ValueError):
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=0)
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_invalid_rating_five(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
with pytest.raises(ValueError):
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=5)
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
created = await repo.create(run_id="r1", thread_id="t1", rating=1)
|
||||
fetched = await repo.get(created["feedback_id"])
|
||||
assert fetched is not None
|
||||
assert fetched["feedback_id"] == created["feedback_id"]
|
||||
assert fetched["rating"] == 1
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_nonexistent(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
assert await repo.get("nonexistent") is None
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_by_run(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-1")
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=-1, user_id="user-2")
|
||||
await repo.create(run_id="r2", thread_id="t1", rating=1, user_id="user-1")
|
||||
results = await repo.list_by_run("t1", "r1", user_id=None)
|
||||
assert len(results) == 2
|
||||
assert all(r["run_id"] == "r1" for r in results)
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_by_thread(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=1)
|
||||
await repo.create(run_id="r2", thread_id="t1", rating=-1)
|
||||
await repo.create(run_id="r3", thread_id="t2", rating=1)
|
||||
results = await repo.list_by_thread("t1")
|
||||
assert len(results) == 2
|
||||
assert all(r["thread_id"] == "t1" for r in results)
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
created = await repo.create(run_id="r1", thread_id="t1", rating=1)
|
||||
deleted = await repo.delete(created["feedback_id"])
|
||||
assert deleted is True
|
||||
assert await repo.get(created["feedback_id"]) is None
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_nonexistent(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
deleted = await repo.delete("nonexistent")
|
||||
assert deleted is False
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aggregate_by_run(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-1")
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-2")
|
||||
await repo.create(run_id="r1", thread_id="t1", rating=-1, user_id="user-3")
|
||||
stats = await repo.aggregate_by_run("t1", "r1")
|
||||
assert stats["total"] == 3
|
||||
assert stats["positive"] == 2
|
||||
assert stats["negative"] == 1
|
||||
assert stats["run_id"] == "r1"
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aggregate_empty(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
stats = await repo.aggregate_by_run("t1", "r1")
|
||||
assert stats["total"] == 0
|
||||
assert stats["positive"] == 0
|
||||
assert stats["negative"] == 0
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_creates_new(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
record = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
|
||||
assert record["rating"] == 1
|
||||
assert record["feedback_id"]
|
||||
assert record["user_id"] == "u1"
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_updates_existing(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
first = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
|
||||
second = await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u1", comment="changed my mind")
|
||||
assert second["feedback_id"] == first["feedback_id"]
|
||||
assert second["rating"] == -1
|
||||
assert second["comment"] == "changed my mind"
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_different_users_separate(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
r1 = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
|
||||
r2 = await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u2")
|
||||
assert r1["feedback_id"] != r2["feedback_id"]
|
||||
assert r1["rating"] == 1
|
||||
assert r2["rating"] == -1
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_invalid_rating(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
with pytest.raises(ValueError):
|
||||
await repo.upsert(run_id="r1", thread_id="t1", rating=0, user_id="u1")
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_by_run(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
|
||||
deleted = await repo.delete_by_run(thread_id="t1", run_id="r1", user_id="u1")
|
||||
assert deleted is True
|
||||
results = await repo.list_by_run("t1", "r1", user_id="u1")
|
||||
assert len(results) == 0
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_by_run_nonexistent(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
deleted = await repo.delete_by_run(thread_id="t1", run_id="r1", user_id="u1")
|
||||
assert deleted is False
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_by_thread_grouped(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
|
||||
await repo.upsert(run_id="r2", thread_id="t1", rating=-1, user_id="u1")
|
||||
await repo.upsert(run_id="r3", thread_id="t2", rating=1, user_id="u1")
|
||||
grouped = await repo.list_by_thread_grouped("t1", user_id="u1")
|
||||
assert "r1" in grouped
|
||||
assert "r2" in grouped
|
||||
assert "r3" not in grouped
|
||||
assert grouped["r1"]["rating"] == 1
|
||||
assert grouped["r2"]["rating"] == -1
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_by_thread_grouped_empty(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
grouped = await repo.list_by_thread_grouped("t1", user_id="u1")
|
||||
assert grouped == {}
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_by_run_ids_is_thread_and_owner_scoped(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
|
||||
await repo.upsert(run_id="r2", thread_id="t1", rating=-1, user_id="u1")
|
||||
await repo.upsert(run_id="r3", thread_id="t1", rating=1, user_id="u1")
|
||||
await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u2")
|
||||
await repo.upsert(run_id="r2", thread_id="t2", rating=1, user_id="u1")
|
||||
|
||||
grouped = await repo.list_by_run_ids("t1", {"r1", "r2"}, user_id="u1")
|
||||
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
|
||||
await _cleanup()
|
||||
assert grouped["r1"].rating == 1
|
||||
assert grouped["r2"].rating == -1
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_by_run_ids_empty_skips_query(self, tmp_path):
|
||||
repo = await _make_feedback_repo(tmp_path)
|
||||
async def test_latest_per_run_in_thread_empty(self, repo):
|
||||
assert await repo.latest_per_run_in_thread("t1", user_id="u1") == {}
|
||||
|
||||
assert await repo.list_by_run_ids("t1", set(), user_id="u1") == {}
|
||||
await _cleanup()
|
||||
@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"}
|
||||
|
||||
|
||||
# -- Follow-up association --
|
||||
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.infra.persistence.feedback 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:
|
||||
|
||||
96
backend/tests/test_feedback_service.py
Normal file
96
backend/tests/test_feedback_service.py
Normal file
@ -0,0 +1,96 @@
|
||||
"""FeedbackService tests with injected fakes — zero IO, no engine.
|
||||
|
||||
The seam the ports create is what makes these tests possible: the service
|
||||
is exercised end-to-end against dict-backed doubles.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from test_feedback import InMemoryFeedbackRepository
|
||||
|
||||
from deerflow.domain.feedback import FeedbackService, InvalidRatingError, RunNotFoundError
|
||||
|
||||
|
||||
class FakeRunLookup:
|
||||
"""RunLookup double backed by a run_id -> thread_id mapping."""
|
||||
|
||||
def __init__(self, runs: dict[str, str]):
|
||||
self._runs = runs
|
||||
|
||||
async def thread_of(self, run_id: str) -> str | None:
|
||||
return self._runs.get(run_id)
|
||||
|
||||
|
||||
def _service(runs: dict[str, str] | None = None) -> FeedbackService:
|
||||
return FeedbackService(
|
||||
repository=InMemoryFeedbackRepository(),
|
||||
runs=FakeRunLookup(runs if runs is not None else {"r1": "t1"}),
|
||||
)
|
||||
|
||||
|
||||
class TestRateRun:
|
||||
@pytest.mark.anyio
|
||||
async def test_stores_rating(self):
|
||||
svc = _service()
|
||||
fb = await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
assert fb.rating == 1
|
||||
assert fb.user_id == "u1"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_idempotent_upsert_keeps_identity(self):
|
||||
svc = _service()
|
||||
first = await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
second = await svc.rate_run("t1", "r1", rating=-1, comment="meh", user_id="u1")
|
||||
assert second.feedback_id == first.feedback_id
|
||||
assert second.rating == -1
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_enrich_with_tags(self):
|
||||
svc = _service()
|
||||
await svc.rate_run("t1", "r1", rating=-1, comment=None, user_id="u1")
|
||||
fb = await svc.rate_run("t1", "r1", rating=-1, comment="wrong", user_id="u1", tags=["incorrect"])
|
||||
assert fb.tags == ("incorrect",)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_unknown_run_rejected(self):
|
||||
svc = _service({})
|
||||
with pytest.raises(RunNotFoundError):
|
||||
await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_thread_run_rejected(self):
|
||||
"""A run id belonging to another thread must not be ratable through
|
||||
this thread's URL (ownership check)."""
|
||||
svc = _service({"r1": "other-thread"})
|
||||
with pytest.raises(RunNotFoundError):
|
||||
await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_invalid_rating_propagates_before_persistence(self):
|
||||
svc = _service()
|
||||
with pytest.raises(InvalidRatingError):
|
||||
await svc.rate_run("t1", "r1", rating=0, comment=None, user_id="u1")
|
||||
|
||||
|
||||
class TestRetractAndReads:
|
||||
@pytest.mark.anyio
|
||||
async def test_retract(self):
|
||||
svc = _service()
|
||||
await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
assert await svc.retract_run_rating("t1", "r1", user_id="u1") is True
|
||||
assert await svc.retract_run_rating("t1", "r1", user_id="u1") is False
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_latest_per_run_in_thread(self):
|
||||
svc = _service({"r1": "t1", "r2": "t1"})
|
||||
await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
await svc.rate_run("t1", "r2", rating=-1, comment=None, user_id="u1")
|
||||
grouped = await svc.latest_per_run_in_thread("t1", user_id="u1")
|
||||
assert {run_id: fb.rating for run_id, fb in grouped.items()} == {"r1": 1, "r2": -1}
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_latest_for_runs(self):
|
||||
svc = _service({"r1": "t1", "r2": "t1"})
|
||||
await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
|
||||
await svc.rate_run("t1", "r2", rating=-1, comment=None, user_id="u1")
|
||||
grouped = await svc.latest_for_runs("t1", {"r2"}, user_id="u1")
|
||||
assert set(grouped) == {"r2"}
|
||||
@ -334,53 +334,32 @@ async def test_run_events_cross_user_delete_denied(tmp_path):
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.no_auto_user
|
||||
async def test_feedback_cross_user_isolation(tmp_path):
|
||||
"""Ownership is an explicit user_id filter on the new feedback port —
|
||||
the AUTO-sentinel context no longer applies (hexagonal slice)."""
|
||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
||||
from deerflow.domain.feedback import Feedback
|
||||
from deerflow.persistence.engine import get_session_factory
|
||||
from deerflow.persistence.feedback import FeedbackRepository
|
||||
|
||||
cleanup = await _make_engines(tmp_path)
|
||||
try:
|
||||
repo = FeedbackRepository(get_session_factory())
|
||||
repo = SqlFeedbackRepository(get_session_factory())
|
||||
|
||||
# User A submits positive feedback.
|
||||
with _as_user(USER_A):
|
||||
a_feedback = await repo.create(
|
||||
run_id="run-a1",
|
||||
thread_id="t-alpha",
|
||||
rating=1,
|
||||
comment="A liked this",
|
||||
)
|
||||
|
||||
# User B submits negative feedback.
|
||||
with _as_user(USER_B):
|
||||
b_feedback = await repo.create(
|
||||
run_id="run-b1",
|
||||
thread_id="t-beta",
|
||||
rating=-1,
|
||||
comment="B disliked this",
|
||||
)
|
||||
await repo.save(Feedback.create(run_id="run-a1", thread_id="t-alpha", rating=1, user_id=USER_A.id, comment="A liked this"))
|
||||
await repo.save(Feedback.create(run_id="run-b1", thread_id="t-beta", rating=-1, user_id=USER_B.id, comment="B disliked this"))
|
||||
|
||||
# User A must see only A's feedback.
|
||||
with _as_user(USER_A):
|
||||
retrieved = await repo.get(a_feedback["feedback_id"])
|
||||
assert retrieved is not None
|
||||
assert retrieved["comment"] == "A liked this"
|
||||
a_view = await repo.latest_per_run_in_thread("t-alpha", user_id=USER_A.id)
|
||||
assert set(a_view) == {"run-a1"}
|
||||
assert a_view["run-a1"].comment == "A liked this"
|
||||
|
||||
# CRITICAL: cannot read B's feedback by id.
|
||||
leaked = await repo.get(b_feedback["feedback_id"])
|
||||
assert leaked is None, "User A leaked User B's feedback"
|
||||
|
||||
# list_by_run for B's run must be empty.
|
||||
empty = await repo.list_by_run("t-beta", "run-b1")
|
||||
assert empty == []
|
||||
# CRITICAL: A cannot see B's feedback through B's thread.
|
||||
leaked = await repo.latest_per_run_in_thread("t-beta", user_id=USER_A.id)
|
||||
assert leaked == {}, "User A leaked User B's feedback"
|
||||
|
||||
# User B must see only B's feedback.
|
||||
with _as_user(USER_B):
|
||||
leaked = await repo.get(a_feedback["feedback_id"])
|
||||
assert leaked is None, "User B leaked User A's feedback"
|
||||
|
||||
b_list = await repo.list_by_run("t-beta", "run-b1")
|
||||
assert len(b_list) == 1
|
||||
assert b_list[0]["comment"] == "B disliked this"
|
||||
assert await repo.latest_per_run_in_thread("t-alpha", user_id=USER_B.id) == {}
|
||||
b_view = await repo.latest_per_run_in_thread("t-beta", user_id=USER_B.id)
|
||||
assert b_view["run-b1"].comment == "B disliked this"
|
||||
finally:
|
||||
await cleanup()
|
||||
|
||||
@ -388,25 +367,21 @@ async def test_feedback_cross_user_isolation(tmp_path):
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.no_auto_user
|
||||
async def test_feedback_cross_user_delete_denied(tmp_path):
|
||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
||||
from deerflow.domain.feedback import Feedback
|
||||
from deerflow.persistence.engine import get_session_factory
|
||||
from deerflow.persistence.feedback import FeedbackRepository
|
||||
|
||||
cleanup = await _make_engines(tmp_path)
|
||||
try:
|
||||
repo = FeedbackRepository(get_session_factory())
|
||||
repo = SqlFeedbackRepository(get_session_factory())
|
||||
await repo.save(Feedback.create(run_id="run-a1", thread_id="t-alpha", rating=1, user_id=USER_A.id))
|
||||
|
||||
with _as_user(USER_A):
|
||||
fb = await repo.create(run_id="run-a1", thread_id="t-alpha", rating=1)
|
||||
# User B tries to retract A's feedback — must return False (no-op).
|
||||
deleted = await repo.remove_for_run("t-alpha", "run-a1", user_id=USER_B.id)
|
||||
assert deleted is False, "User B deleted User A's feedback"
|
||||
|
||||
# User B tries to delete A's feedback — must return False (no-op).
|
||||
with _as_user(USER_B):
|
||||
deleted = await repo.delete(fb["feedback_id"])
|
||||
assert deleted is False, "User B deleted User A's feedback"
|
||||
|
||||
# A's feedback still retrievable.
|
||||
with _as_user(USER_A):
|
||||
row = await repo.get(fb["feedback_id"])
|
||||
assert row is not None
|
||||
# A's feedback still present.
|
||||
assert set(await repo.latest_per_run_in_thread("t-alpha", user_id=USER_A.id)) == {"run-a1"}
|
||||
finally:
|
||||
await cleanup()
|
||||
|
||||
|
||||
@ -76,12 +76,17 @@ async def test_run_repository_emits_tz_aware_timestamps(tmp_path):
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_feedback_repository_emits_tz_aware_timestamps(tmp_path):
|
||||
from deerflow.persistence.feedback import FeedbackRepository
|
||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
||||
from deerflow.domain.feedback import Feedback
|
||||
|
||||
repo = FeedbackRepository(await _init_sqlite(tmp_path))
|
||||
repo = SqlFeedbackRepository(await _init_sqlite(tmp_path))
|
||||
try:
|
||||
record = await repo.create(run_id="r-tz", thread_id="t-tz", rating=1, user_id="u1")
|
||||
_assert_tz_aware(record["created_at"], context="feedback.create.created_at")
|
||||
fb = await repo.save(Feedback.create(run_id="r-tz", thread_id="t-tz", rating=1, user_id="u1"))
|
||||
# The port exchanges domain objects (datetime), not ISO strings;
|
||||
# serialization to ISO happens in the router. Assert tz-awareness here.
|
||||
assert fb.created_at.tzinfo is not None, "feedback.save.created_at lacks tzinfo"
|
||||
read_back = (await repo.latest_per_run_in_thread("t-tz", user_id="u1"))["r-tz"]
|
||||
assert read_back.created_at.tzinfo is not None, "feedback read-back lacks tzinfo"
|
||||
finally:
|
||||
await _cleanup()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user