43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
"""Tests for the conditional Campaign start lifecycle seam."""
|
|
|
|
from agenteval.evaluation.campaign_lifecycle import start_campaign
|
|
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus
|
|
from agenteval.storage.repository import CampaignRepository
|
|
|
|
|
|
def test_start_planned_campaign_is_atomic_and_launches_after_commit(db_session):
|
|
CampaignRepository(db_session).create(
|
|
Campaign(
|
|
id="campaign-1",
|
|
name="campaign",
|
|
target_id="t-1",
|
|
status=CampaignStatus.PLANNED,
|
|
window_seconds=60,
|
|
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
|
|
)
|
|
)
|
|
observed = []
|
|
|
|
def launch(campaign_id, session):
|
|
observed.append(CampaignRepository(session).get(campaign_id).status)
|
|
|
|
started = start_campaign(db_session, "campaign-1", launch=launch)
|
|
|
|
assert started.status is CampaignStatus.RUNNING
|
|
assert started.started_at is not None
|
|
assert observed == [CampaignStatus.RUNNING]
|
|
|
|
|
|
def test_start_terminal_campaign_is_noop(db_session):
|
|
CampaignRepository(db_session).create(
|
|
Campaign(
|
|
id="campaign-1",
|
|
name="campaign",
|
|
target_id="t-1",
|
|
status=CampaignStatus.COMPLETED,
|
|
window_seconds=60,
|
|
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
|
|
)
|
|
)
|
|
assert start_campaign(db_session, "campaign-1") is None
|