AgentEvalTool/tests/integration/test_campaign_comparison_api.py

"""Integration tests for /api/campaigns/{id}/comparison (v0.8 tickets 01+02)."""

from datetime import timedelta

import pytest
from agenteval.models import (
    CampaignStatus,
    Case,
    CaseType,
    ChannelType,
    EvalRun,
    EvalTarget,
    PlatformType,
    RunStatus,
    RunSummary,
    Scenario,
    TargetStatus,
)
from agenteval.storage.db import CampaignAnalysisDB, CampaignPeriodComparisonDB, ModelConfigDB, utc_now
from agenteval.storage.model_config_repository import ModelConfigRepository
from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository, TargetRepository
from agenteval.web.app import app
from httpx import ASGITransport, AsyncClient
from sqlmodel import select


@pytest.fixture()
def seeded_db(db_session, monkeypatch):
    from agenteval.storage import db as db_module
    from agenteval.storage import repository as repo_module
    from agenteval.web import app as app_module
    from agenteval.web.routers import campaigns as campaigns_module

    monkeypatch.setattr(app_module, "init_db", lambda: None)
    monkeypatch.setattr(campaigns_module, "start_campaign", lambda *a, **k: None)

    monkeypatch.setattr(db_module, "get_session", lambda: db_session)
    monkeypatch.setattr(repo_module, "get_session", lambda: db_session)

    from agenteval.web.deps import get_db

    def _test_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = _test_get_db

    TargetRepository(db_session).create(EvalTarget(
        id="t-1", name="mock-target",
        platform=PlatformType.AI_DIGITAL_EMPLOYEE,
        channel_type=ChannelType.TUTU_API,
        channel_config={"base_url": "http://mock", "token": "x"},
        status=TargetStatus.ACTIVE,
    ))
    ScenarioRepository(db_session).create(Scenario(
        id="s-1", name="mock-scenario",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    ))

    yield db_session

    app.dependency_overrides.clear()


@pytest.fixture()
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


def _payload(name: str = "24h-cycle") -> dict:
    return {
        "name": name,
        "target_id": "t-1",
        "window_seconds": 86400,
        "time_scale": 1.0,
        "plan": [{"scenario_id": "s-1", "offset_seconds": 0, "count": 1}],
    }


async def _create_completed_campaign(client, session, name: str, *, completed_at) -> str:
    campaign_id = (await client.post("/api/campaigns", json=_payload(name))).json()["id"]
    repo = CampaignRepository(session)
    campaign = repo.get(campaign_id)
    campaign.status = CampaignStatus.COMPLETED
    campaign.completed_at = completed_at
    repo.update(campaign)
    return campaign_id


def _complete_analysis(session, campaign_id: str) -> None:
    session.add(CampaignAnalysisDB(campaign_id=campaign_id, status="completed"))
    session.commit()


def _seed_run(session, run_id: str, campaign_id: str, *, pass_rate: float, latency: float) -> None:
    RunRepository(session).create(EvalRun(
        id=run_id, target_id="t-1", scenario_id="s-1", campaign_id=campaign_id,
        status=RunStatus.COMPLETED, started_at=utc_now(),
        summary=RunSummary(total_cases=2, pass_rate=pass_rate, avg_latency_ms=latency),
    ))


async def test_get_comparison_missing_campaign_404(client, seeded_db):
    resp = await client.get("/api/campaigns/nope/comparison")
    assert resp.status_code == 404


async def test_get_comparison_without_baseline_returns_nulls(client, seeded_db):
    campaign_id = await _create_completed_campaign(
        client, seeded_db, "第一期", completed_at=utc_now() - timedelta(hours=1),
    )
    resp = await client.get(f"/api/campaigns/{campaign_id}/comparison")
    assert resp.status_code == 200
    body = resp.json()
    assert body["status"] == "none"
    assert body["comparison"] is None
    assert body["auto_baseline"] is None
    assert body["metric_diff"] is None


async def test_get_comparison_accelerated_line_has_no_baseline(client, seeded_db):
    completed = utc_now() - timedelta(hours=1)
    await _create_completed_campaign(client, seeded_db, "上期正式线", completed_at=completed - timedelta(hours=2))
    # 本期为加速调试线：即便有同指纹正式线也不自动配对
    payload = _payload("本期加速线")
    payload["time_scale"] = 10.0
    campaign_id = (await client.post("/api/campaigns", json=payload)).json()["id"]
    repo = CampaignRepository(seeded_db)
    campaign = repo.get(campaign_id)
    campaign.status = CampaignStatus.COMPLETED
    campaign.completed_at = completed
    repo.update(campaign)

    resp = await client.get(f"/api/campaigns/{campaign_id}/comparison")
    assert resp.status_code == 200
    body = resp.json()
    assert body["status"] == "none"
    assert body["auto_baseline"] is None
    assert body["metric_diff"] is None


async def test_get_comparison_returns_baseline_and_metric_diff(client, seeded_db):
    completed_now = utc_now()
    baseline_id = await _create_completed_campaign(
        client, seeded_db, "上期", completed_at=completed_now - timedelta(hours=2),
    )
    _complete_analysis(seeded_db, baseline_id)
    current_id = await _create_completed_campaign(
        client, seeded_db, "本期", completed_at=completed_now,
    )

    _seed_run(seeded_db, "run-base", baseline_id, pass_rate=0.5, latency=800.0)
    _seed_run(seeded_db, "run-cur", current_id, pass_rate=0.9, latency=500.0)

    resp = await client.get(f"/api/campaigns/{current_id}/comparison")
    assert resp.status_code == 200
    body = resp.json()

    assert body["auto_baseline"]["id"] == baseline_id
    assert body["auto_baseline"]["name"] == "上期"
    assert body["auto_baseline"]["completed_at"]

    diff = body["metric_diff"]
    assert diff["overall"]["pass_rate"] == {"baseline": 0.5, "current": 0.9, "delta": 0.4}
    assert diff["overall"]["availability"]["baseline"] == 1.0
    assert diff["overall"]["availability"]["delta"] == 0.0
    assert diff["overall"]["avg_latency_ms"] == {"baseline": 800.0, "current": 500.0, "delta": -300.0}
    assert [s["scenario_id"] for s in diff["scenarios"]] == ["s-1"]
    assert diff["scenarios"][0]["pass_rate"]["delta"] == 0.4


async def test_get_comparison_baseline_without_analysis_is_skipped(client, seeded_db):
    completed_now = utc_now()
    baseline_id = await _create_completed_campaign(
        client, seeded_db, "上期无分析", completed_at=completed_now - timedelta(hours=2),
    )
    _seed_run(seeded_db, "run-base", baseline_id, pass_rate=0.5, latency=800.0)
    current_id = await _create_completed_campaign(
        client, seeded_db, "本期", completed_at=completed_now,
    )
    _seed_run(seeded_db, "run-cur", current_id, pass_rate=0.9, latency=500.0)

    resp = await client.get(f"/api/campaigns/{current_id}/comparison")
    assert resp.status_code == 200
    body = resp.json()
    assert body["status"] == "none"
    assert body["auto_baseline"] is None
    assert body["metric_diff"] is None


# ── ticket 02: POST + GET 合并 ────────────────────────────────────────────


def _seed_analysis_default(session) -> None:
    ModelConfigRepository(session).create(ModelConfigDB(
        id="mc-1", name="analysis-cfg", provider="openai_compatible", capability="chat",
        endpoint_url="https://models.example.com/v1/chat/completions", model_name="m",
        is_analysis_default=True,
    ))


def _complete_analysis_with_result(session, campaign_id: str) -> None:
    row = CampaignAnalysisDB(campaign_id=campaign_id, status="completed")
    row.set_result({
        "overall": "整体达标",
        "problems": [],
        "scenario_narratives": [{"scenario_id": "s-1", "narrative": "表现稳定"}],
        "suggestions": [{"priority": 1, "text": "保持"}],
    })
    existing = session.exec(
        select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == campaign_id)
    ).first()
    if existing:
        session.delete(existing)
        session.commit()
    session.add(row)
    session.commit()


def _complete_comparison_row(session, campaign_id: str, baseline_id: str) -> None:
    row = CampaignPeriodComparisonDB(
        campaign_id=campaign_id, baseline_campaign_id=baseline_id,
        status="completed", model_config_id="mc-1", triggered_by="manual",
    )
    row.set_result({
        "trend": "improving",
        "summary": "通过率提升",
        "problem_evolution": [{"status": "resolved", "title": "答非所问", "detail": "已缓解", "scenario_ids": ["s-1"]}],
        "suggestion_tracking": [{"text": "保持", "status": "addressed", "note": "已落实"}],
    })
    existing = session.exec(
        select(CampaignPeriodComparisonDB).where(CampaignPeriodComparisonDB.campaign_id == campaign_id)
    ).first()
    if existing:
        session.delete(existing)
        session.commit()
    session.add(row)
    session.commit()


async def _two_period_setup(client, seeded_db):
    """上期（completed + 分析）+ 本期（completed + 分析），返回 (baseline_id, current_id)。"""
    completed_now = utc_now()
    baseline_id = await _create_completed_campaign(
        client, seeded_db, "上期", completed_at=completed_now - timedelta(hours=2),
    )
    _complete_analysis_with_result(seeded_db, baseline_id)
    current_id = await _create_completed_campaign(
        client, seeded_db, "本期", completed_at=completed_now,
    )
    _complete_analysis_with_result(seeded_db, current_id)
    return baseline_id, current_id


async def test_post_comparison_rejects_non_terminal(client, seeded_db):
    _seed_analysis_default(seeded_db)
    payload = _payload("运行中")
    campaign_id = (await client.post("/api/campaigns", json=payload)).json()["id"]
    repo = CampaignRepository(seeded_db)
    campaign = repo.get(campaign_id)
    campaign.status = CampaignStatus.RUNNING
    repo.update(campaign)

    resp = await client.post(f"/api/campaigns/{campaign_id}/comparison")
    assert resp.status_code == 400


async def test_post_comparison_requires_current_analysis(client, seeded_db):
    _seed_analysis_default(seeded_db)
    campaign_id = await _create_completed_campaign(
        client, seeded_db, "无分析本期", completed_at=utc_now(),
    )
    resp = await client.post(f"/api/campaigns/{campaign_id}/comparison")
    assert resp.status_code == 400
    assert "智能分析" in resp.json()["detail"]


async def test_post_comparison_requires_model(client, seeded_db):
    baseline_id, current_id = await _two_period_setup(client, seeded_db)
    # 不配置分析默认模型
    resp = await client.post(f"/api/campaigns/{current_id}/comparison")
    assert resp.status_code == 400
    assert "分析" in resp.json()["detail"]


async def test_post_comparison_without_baseline_guides_manual_selection(client, seeded_db):
    _seed_analysis_default(seeded_db)
    # 本期有分析但无同指纹历史活动
    campaign_id = await _create_completed_campaign(
        client, seeded_db, "孤立本期", completed_at=utc_now(),
    )
    _complete_analysis_with_result(seeded_db, campaign_id)
    resp = await client.post(f"/api/campaigns/{campaign_id}/comparison")
    assert resp.status_code == 400
    assert "基线" in resp.json()["detail"]


async def test_post_comparison_auto_baseline_success(client, seeded_db, monkeypatch):
    from agenteval.web.routers import campaigns as campaigns_module

    _seed_analysis_default(seeded_db)
    baseline_id, current_id = await _two_period_setup(client, seeded_db)
    _seed_run(seeded_db, "run-base", baseline_id, pass_rate=0.5, latency=800.0)
    _seed_run(seeded_db, "run-cur", current_id, pass_rate=0.9, latency=500.0)

    monkeypatch.setattr(
        campaigns_module,
        "start_campaign_comparison",
        lambda cid, *, triggered_by, baseline_campaign_id=None: _complete_comparison_row(
            seeded_db, cid, baseline_campaign_id
        ),
    )

    resp = await client.post(f"/api/campaigns/{current_id}/comparison")
    assert resp.status_code == 200
    assert resp.json() == {"status": "generating"}

    got = (await client.get(f"/api/campaigns/{current_id}/comparison")).json()
    assert got["status"] == "completed"
    assert got["comparison"]["baseline_campaign_id"] == baseline_id
    assert got["comparison"]["baseline"]["name"] == "上期"
    assert got["comparison"]["result"]["trend"] == "improving"
    assert got["comparison"]["model_config_id"] == "mc-1"
    # metric_diff 按生效基线（=存储基线）现算
    assert got["metric_diff"]["overall"]["pass_rate"]["delta"] == 0.4


async def test_post_comparison_manual_baseline_across_fingerprints(client, seeded_db, monkeypatch):
    from agenteval.web.routers import campaigns as campaigns_module

    _seed_analysis_default(seeded_db)
    # 本期：24h 窗口（无同指纹历史活动 → 无自动基线）
    current_id = await _create_completed_campaign(client, seeded_db, "本期", completed_at=utc_now())
    _complete_analysis_with_result(seeded_db, current_id)
    # 手动基线：不同窗口（不同指纹）的历史活动，有 completed 分析
    other_payload = _payload("异指纹基线")
    other_payload["window_seconds"] = 3600
    other_id = (await client.post("/api/campaigns", json=other_payload)).json()["id"]
    repo = CampaignRepository(seeded_db)
    other = repo.get(other_id)
    other.status = CampaignStatus.COMPLETED
    other.completed_at = utc_now() - timedelta(hours=2)
    repo.update(other)
    _complete_analysis_with_result(seeded_db, other_id)

    monkeypatch.setattr(
        campaigns_module,
        "start_campaign_comparison",
        lambda cid, *, triggered_by, baseline_campaign_id=None: _complete_comparison_row(
            seeded_db, cid, baseline_campaign_id
        ),
    )

    # 不指定基线 → 无自动基线 400
    resp = await client.post(f"/api/campaigns/{current_id}/comparison")
    assert resp.status_code == 400

    # 手动指定跨指纹基线 → 成功
    resp = await client.post(
        f"/api/campaigns/{current_id}/comparison",
        json={"baseline_campaign_id": other_id},
    )
    assert resp.status_code == 200

    got = (await client.get(f"/api/campaigns/{current_id}/comparison")).json()
    assert got["comparison"]["baseline_campaign_id"] == other_id


async def test_post_comparison_manual_baseline_without_analysis_400(client, seeded_db):
    _seed_analysis_default(seeded_db)
    baseline_id, current_id = await _two_period_setup(client, seeded_db)
    # 再建一个无分析的历史活动作为手动基线
    ghost_id = await _create_completed_campaign(
        client, seeded_db, "无分析基线", completed_at=utc_now() - timedelta(hours=3),
    )

    resp = await client.post(
        f"/api/campaigns/{current_id}/comparison",
        json={"baseline_campaign_id": ghost_id},
    )
    assert resp.status_code == 400
    assert "基线" in resp.json()["detail"]


async def test_post_comparison_rerun_upserts_without_new_row(client, seeded_db, monkeypatch):
    from agenteval.web.routers import campaigns as campaigns_module

    _seed_analysis_default(seeded_db)
    baseline_id, current_id = await _two_period_setup(client, seeded_db)

    monkeypatch.setattr(
        campaigns_module,
        "start_campaign_comparison",
        lambda cid, *, triggered_by, baseline_campaign_id=None: _complete_comparison_row(
            seeded_db, cid, baseline_campaign_id
        ),
    )

    await client.post(f"/api/campaigns/{current_id}/comparison")
    await client.post(f"/api/campaigns/{current_id}/comparison")

    rows = seeded_db.exec(
        select(CampaignPeriodComparisonDB).where(CampaignPeriodComparisonDB.campaign_id == current_id)
    ).all()
    assert len(rows) == 1