5ecb30876e
- ruff --fix 自动修正 44 项：移除未用 import（pytest 等）、import 块排序归一（I001） - 手工修复剩余 5 项：test_cascade.py 两处未用赋值（F841）；test_s2_rules_and_logic.py 中部 import 移至文件顶部（E402 ×3） - 无行为变更：全量 492 项测试通过
76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
"""CampaignSummary VO — typed campaign scheduler state, restart-safe (ADR-0003).
|
|
|
|
Mirrors RunSummary's typed treatment: Campaign.summary is no longer a bare
|
|
dict. Legacy dict summaries coerce; unknown top-level keys survive
|
|
(extra=allow) so older records keep parsing.
|
|
"""
|
|
|
|
from agenteval.models import (
|
|
Campaign,
|
|
CampaignPlanEntry,
|
|
CampaignSummary,
|
|
SchedulerState,
|
|
)
|
|
from agenteval.storage.repository import CampaignRepository, TargetRepository
|
|
|
|
from tests.unit.test_repository import _make_target
|
|
|
|
|
|
def _campaign(**summary_kw) -> Campaign:
|
|
kw = dict(
|
|
name="c",
|
|
target_id="t-1",
|
|
window_seconds=3600,
|
|
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
|
|
)
|
|
kw.update(summary_kw)
|
|
return Campaign(**kw)
|
|
|
|
|
|
def test_summary_defaults_are_empty():
|
|
s = CampaignSummary()
|
|
assert s.scheduler.spawned_indices == []
|
|
assert s.scheduler.errors == []
|
|
|
|
|
|
def test_scheduler_state_holds_progress():
|
|
s = SchedulerState(spawned_indices=[0, 2], errors=[{"index": 1, "error": "boom"}])
|
|
assert s.spawned_indices == [0, 2]
|
|
assert s.errors[0]["error"] == "boom"
|
|
|
|
|
|
def test_legacy_dict_summary_coerces():
|
|
c = _campaign(summary={"scheduler": {"spawned_indices": [0, 1], "errors": [{"index": 2, "error": "x"}]}})
|
|
assert isinstance(c.summary, CampaignSummary)
|
|
assert c.summary.scheduler.spawned_indices == [0, 1]
|
|
assert c.summary.scheduler.errors[0]["error"] == "x"
|
|
|
|
|
|
def test_assigning_dict_coerces_to_vo():
|
|
c = _campaign()
|
|
c.summary = {"scheduler": {"spawned_indices": [3]}}
|
|
assert isinstance(c.summary, CampaignSummary)
|
|
assert c.summary.scheduler.spawned_indices == [3]
|
|
|
|
|
|
def test_unknown_top_level_keys_survive():
|
|
c = _campaign(summary={"scheduler": {}, "future_axis": {"availability": 0.9}})
|
|
assert isinstance(c.summary, CampaignSummary)
|
|
assert c.summary.model_extra["future_axis"] == {"availability": 0.9}
|
|
|
|
|
|
def test_summary_survives_repository_round_trip(db_session):
|
|
TargetRepository(db_session).create(_make_target())
|
|
repo = CampaignRepository(db_session)
|
|
repo.create(
|
|
_campaign(
|
|
id="cp-1",
|
|
summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1], errors=[{"index": 2}])),
|
|
)
|
|
)
|
|
fetched = repo.get("cp-1")
|
|
assert fetched is not None
|
|
assert isinstance(fetched.summary, CampaignSummary)
|
|
assert fetched.summary.scheduler.spawned_indices == [0, 1]
|
|
assert fetched.summary.scheduler.errors == [{"index": 2}]
|