shaguabijia-app-server/tests/test_comparison_admin_summary.py

"""比价记录页后端概览聚合（admin 口径：外部缺失记为成功）。"""
from __future__ import annotations

from datetime import UTC, date, datetime

import pytest
from sqlalchemy.dialects import postgresql

from app.admin.repositories import queries
from app.admin.repositories.comparison_outcome import admin_success_sql
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord


def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
    stmt = queries._comparison_duration_aggregate_stmt(
        [], admin_success_sql(), (0.05, 0.5, 0.95, 0.99)
    )
    sql = str(
        stmt.compile(
            dialect=postgresql.dialect(),
            compile_kwargs={"literal_binds": True},
        )
    )
    assert sql.count("percentile_cont") == 4
    # 口径按原始结局 coalesce，而非直接读 status 列
    assert "coalesce" in sql.lower()


def test_summary_counts_external_gaps_as_success() -> None:
    db = SessionLocal()
    try:
        # (trace_id, status 列[已 normalize], record_status[原始结局], total_ms, cost, saved)
        rows = [
            ("sum-success", "success", "success", 1000, 1.0, 100),
            ("sum-below-min", "success", "below_minimum", 2000, 2.0, 0),
            ("sum-store-closed", "failed", "store_closed", 3000, None, 0),
            ("sum-store-not-found", "failed", "store_not_found", 4000, None, 0),
            ("sum-failed", "failed", "failed", 100_000, 3.0, 0),
            ("sum-cancelled", "cancelled", None, 5000, None, 0),
            ("sum-running", "running", None, 6000, None, 0),
        ]
        for trace_id, status, record_status, total_ms, cost, saved in rows:
            db.add(ComparisonRecord(
                trace_id=trace_id,
                status=status,
                total_ms=total_ms,
                llm_cost_yuan=cost,
                saved_amount_cents=saved,
                raw_payload={"record_status": record_status} if record_status else None,
                created_at=datetime(2038, 1, 15, 12, tzinfo=UTC),
            ))
        db.flush()

        summary = queries.comparison_records_summary(
            db, date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
        )

        # admin 成功 = success + below_minimum + store_closed + store_not_found = 4
        assert summary["started"] == 7
        assert summary["success"] == 4
        assert summary["completed"] == 5           # 4 成功 + 1 纯 failed
        assert summary["cancelled"] == 1
        assert summary["success_rate"] == pytest.approx(4 / 6)  # 分母 started - cancelled
        assert summary["avg_token_cost"] == pytest.approx(2.0)  # (1+2+3)/3
        assert summary["lower_price_rate"] == pytest.approx(1 / 4)  # 仅 sum-success saved>0
        # 耗时统计集 = admin 成功的 total_ms [1000,2000,3000,4000]
        assert summary["avg_duration_ms"] == 2500
        assert summary["p5_duration_ms"] == 1150
        assert summary["p50_duration_ms"] == 2500
        assert summary["p95_duration_ms"] == 3850
        assert summary["p99_duration_ms"] == 3970
        assert summary["cancelled_p50_ms"] == 5000
    finally:
        db.rollback()
        db.close()


def test_list_status_filter_uses_admin_outcome() -> None:
    """列表「状态」筛选走 admin 口径：筛成功含 6 类、筛失败仅纯技术故障；并验日期边界排除。"""
    db = SessionLocal()
    try:
        rows = [
            ("flt-success", "success", "success"),
            ("flt-below-min", "success", "below_minimum"),
            ("flt-store-not-found", "failed", "store_not_found"),
            ("flt-failed", "failed", "failed"),
            ("flt-cancelled", "cancelled", None),
        ]
        for trace_id, status, record_status in rows:
            db.add(ComparisonRecord(
                trace_id=trace_id,
                status=status,
                raw_payload={"record_status": record_status} if record_status else None,
                created_at=datetime(2039, 3, 10, 12, tzinfo=UTC),
            ))
        # 日期边界：窗口外一条 admin 成功记录（次日），应被日期筛选排除
        db.add(ComparisonRecord(
            trace_id="flt-out-of-window",
            status="success",
            raw_payload={"record_status": "success"},
            created_at=datetime(2039, 3, 11, 12, tzinfo=UTC),
        ))
        db.flush()

        succ, _c1, succ_total = queries.list_comparison_records(
            db, status="success", date_from=date(2039, 3, 10), date_to=date(2039, 3, 10)
        )
        # 3 条当天 admin 成功（含 store_not_found）；窗口外 flt-out-of-window 被日期排除
        assert succ_total == 3
        assert {it.trace_id for it in succ} == {
            "flt-success", "flt-below-min", "flt-store-not-found",
        }

        fail, _c2, fail_total = queries.list_comparison_records(
            db, status="failed", date_from=date(2039, 3, 10), date_to=date(2039, 3, 10)
        )
        assert fail_total == 1
        assert {it.trace_id for it in fail} == {"flt-failed"}

        _canc, _c3, canc_total = queries.list_comparison_records(
            db, status="cancelled", date_from=date(2039, 3, 10), date_to=date(2039, 3, 10)
        )
        assert canc_total == 1
    finally:
        db.rollback()
        db.close()