d411572607
Target/Scenario/Run/Campaign repositories repeated the same __init__/list_all/get/create/delete skeleton (~120 lines). Extract a generic BaseRepository[M, DB]: subclasses declare the table + ordering column and implement instance-method _to_db/_from_db converters (so Scenario's _from_db can reach self.session for model bindings). Bespoke paths (Scenario create/delete, all update) stay per-subclass. Adds test_repository.py locking the CRUD + JSON round-trip contract.
114 lines
3.3 KiB
Python
114 lines
3.3 KiB
Python
"""CRUD + serialization round-trip characterization for the repository layer.
|
|
|
|
These lock the create → get → list_all → delete contract shared by the
|
|
Target / Run / Campaign repositories (the BaseRepository skeleton), plus the
|
|
summary/plan JSON round-trip, so the generic-base refactor stays behaviour-
|
|
preserving. Scenario's bespoke create/update (binding validation, versioning)
|
|
is covered by its own tests.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from agenteval.models import (
|
|
Campaign,
|
|
CampaignPlanEntry,
|
|
ChannelType,
|
|
EvalRun,
|
|
EvalTarget,
|
|
PlatformType,
|
|
RunStatus,
|
|
RunSummary,
|
|
TargetStatus,
|
|
)
|
|
from agenteval.storage.repository import (
|
|
CampaignRepository,
|
|
RunRepository,
|
|
TargetRepository,
|
|
)
|
|
|
|
|
|
def _make_target(tid: str = "t-1") -> EvalTarget:
|
|
return EvalTarget(
|
|
id=tid,
|
|
name="target",
|
|
platform=PlatformType.AI_DIGITAL_EMPLOYEE,
|
|
channel_type=ChannelType.TUTU_API,
|
|
channel_config={
|
|
"base_url": "x",
|
|
"token": "x",
|
|
"tenant": "x",
|
|
"chat_channel_id": "x",
|
|
"chat_contact_id": "x",
|
|
},
|
|
)
|
|
|
|
|
|
def test_target_crud_round_trip(db_session):
|
|
repo = TargetRepository(db_session)
|
|
repo.create(_make_target())
|
|
|
|
fetched = repo.get("t-1")
|
|
assert fetched is not None
|
|
assert fetched.name == "target"
|
|
assert fetched.channel_config["base_url"] == "x"
|
|
assert fetched.status == TargetStatus.ACTIVE or fetched.status is not None
|
|
|
|
assert [t.id for t in repo.list_all()] == ["t-1"]
|
|
|
|
assert repo.delete("t-1") is True
|
|
assert repo.get("t-1") is None
|
|
assert repo.delete("t-1") is False
|
|
|
|
|
|
def test_target_list_all_newest_first(db_session):
|
|
repo = TargetRepository(db_session)
|
|
older = _make_target("t-old")
|
|
older.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
newer = _make_target("t-new")
|
|
newer.created_at = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
|
repo.create(older)
|
|
repo.create(newer)
|
|
assert [t.id for t in repo.list_all()] == ["t-new", "t-old"]
|
|
|
|
|
|
def test_run_summary_json_round_trip(db_session):
|
|
TargetRepository(db_session).create(_make_target())
|
|
repo = RunRepository(db_session)
|
|
repo.create(
|
|
EvalRun(
|
|
id="r-1",
|
|
target_id="t-1",
|
|
scenario_id="s-1",
|
|
status=RunStatus.COMPLETED,
|
|
summary=RunSummary(total_cases=2, passed_cases=1, pass_rate=0.5),
|
|
)
|
|
)
|
|
fetched = repo.get("r-1")
|
|
assert fetched is not None
|
|
assert fetched.summary is not None
|
|
assert fetched.summary.total_cases == 2
|
|
assert fetched.summary.pass_rate == 0.5
|
|
|
|
|
|
def test_campaign_crud_and_plan_round_trip(db_session):
|
|
TargetRepository(db_session).create(_make_target())
|
|
repo = CampaignRepository(db_session)
|
|
repo.create(
|
|
Campaign(
|
|
id="cp-1",
|
|
name="campaign",
|
|
target_id="t-1",
|
|
window_seconds=3600,
|
|
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2)],
|
|
)
|
|
)
|
|
fetched = repo.get("cp-1")
|
|
assert fetched is not None
|
|
assert fetched.plan[0].scenario_id == "s-1"
|
|
assert fetched.plan[0].count == 2
|
|
assert [c.id for c in repo.list_all()] == ["cp-1"]
|
|
|
|
# Campaign inherits the shared delete() from the base repository.
|
|
assert repo.delete("cp-1") is True
|
|
assert repo.get("cp-1") is None
|