38849d46f1
全行 update 逼着调用方承诺「持有最新完整快照」，巡检水位、活动取消、 调度循环三方并发写入会互相覆盖（后写者抹掉先写者的 summary/status）。 CampaignRepository 新增三个窄口径原子操作：touch_patrol_watermark、 mark_cancelled、save_scheduler_state，各自只写自己的列；三处调用点迁移。
103 lines
3.8 KiB
Python
103 lines
3.8 KiB
Python
"""CampaignRepository 窄口径原子更新直测（架构保养候选 1）。
|
||
|
||
全行 update 逼着调用方承诺「持有最新完整快照」，巡检水位 / 取消 /
|
||
调度循环并发写入会互相覆盖。三个窄方法各自只动自己的列：
|
||
touch_patrol_watermark（水位）、mark_cancelled（终态迁移）、
|
||
save_scheduler_state（调度进度）。
|
||
"""
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
import pytest
|
||
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, CampaignSummary, SchedulerState
|
||
from agenteval.storage.db import as_utc
|
||
from agenteval.storage.repository import CampaignRepository
|
||
from sqlmodel import Session, SQLModel, create_engine
|
||
|
||
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
||
T1 = datetime(2026, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
@pytest.fixture()
|
||
def db_session(tmp_path):
|
||
from agenteval.storage.db import ( # noqa: F401
|
||
CampaignDB,
|
||
EvalResultDB,
|
||
EvalRunDB,
|
||
EvalTargetDB,
|
||
FileCategoryDB,
|
||
FileRecordDB,
|
||
ScenarioDB,
|
||
TurnDB,
|
||
)
|
||
engine = create_engine(
|
||
f"sqlite:///{tmp_path / 'narrow.db'}",
|
||
connect_args={"check_same_thread": False},
|
||
)
|
||
SQLModel.metadata.create_all(engine)
|
||
session = Session(engine)
|
||
try:
|
||
yield session
|
||
finally:
|
||
session.close()
|
||
engine.dispose()
|
||
|
||
|
||
def _seed(db_session, status=CampaignStatus.RUNNING) -> Campaign:
|
||
campaign = CampaignRepository(db_session).create(Campaign(
|
||
name="cycle", target_id="t-1", window_seconds=3600,
|
||
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
|
||
status=status, started_at=T0,
|
||
summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0])),
|
||
))
|
||
return campaign
|
||
|
||
|
||
def test_touch_patrol_watermark_only_touches_watermark(db_session):
|
||
campaign = _seed(db_session)
|
||
CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
|
||
|
||
fresh = CampaignRepository(db_session).get(campaign.id)
|
||
assert as_utc(fresh.last_patrolled_at) == T1
|
||
# 其余列原样：窄口径不覆写并发写入的 status / summary
|
||
assert fresh.status == CampaignStatus.RUNNING
|
||
assert fresh.summary.scheduler.spawned_indices == [0]
|
||
assert fresh.name == "cycle" and as_utc(fresh.started_at) == T0
|
||
|
||
|
||
def test_mark_cancelled_only_touches_status_and_completed_at(db_session):
|
||
campaign = _seed(db_session)
|
||
CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
|
||
|
||
updated = CampaignRepository(db_session).mark_cancelled(campaign.id, T1)
|
||
assert updated.status == CampaignStatus.CANCELLED
|
||
assert as_utc(updated.completed_at) == T1
|
||
|
||
fresh = CampaignRepository(db_session).get(campaign.id)
|
||
assert fresh.status == CampaignStatus.CANCELLED
|
||
assert as_utc(fresh.completed_at) == T1
|
||
# 水位与调度进度不被终态迁移抹掉
|
||
assert as_utc(fresh.last_patrolled_at) == T1
|
||
assert fresh.summary.scheduler.spawned_indices == [0]
|
||
|
||
|
||
def test_save_scheduler_state_only_touches_summary(db_session):
|
||
campaign = _seed(db_session)
|
||
CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
|
||
|
||
new_summary = CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1, 2]))
|
||
CampaignRepository(db_session).save_scheduler_state(campaign.id, new_summary)
|
||
|
||
fresh = CampaignRepository(db_session).get(campaign.id)
|
||
assert fresh.summary.scheduler.spawned_indices == [0, 1, 2]
|
||
# 状态与水位不被调度进度覆写
|
||
assert fresh.status == CampaignStatus.RUNNING
|
||
assert as_utc(fresh.last_patrolled_at) == T1
|
||
|
||
|
||
def test_narrow_updates_ignore_missing_campaign(db_session):
|
||
repo = CampaignRepository(db_session)
|
||
repo.touch_patrol_watermark("missing", T1)
|
||
repo.save_scheduler_state("missing", CampaignSummary())
|
||
assert repo.mark_cancelled("missing", T1) is None
|