shaguabijia-app-server/tests/test_analytics_health.py

"""埋点健康度聚合:纯差分函数 + admin 端点。"""
from __future__ import annotations

from datetime import datetime, timezone

from app.admin.repositories.analytics_health import _cn_day, _rates, diff_snapshots


def _row(device, epoch, event, ts, **cum) -> dict:
    base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}
    base.update(cum)
    return {
        "device_id": device, "epoch_id": epoch, "event": event,
        "created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc),
        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
        **base,
    }


def test_diff_first_row_is_full_cumulative() -> None:
    rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)]
    out = diff_snapshots(rows)
    assert len(out) == 1
    assert out[0]["d_attempted"] == 100
    assert out[0]["d_delivered"] == 90


def test_diff_consecutive_delta() -> None:
    rows = [
        _row("d", "e", "video_play", 1, attempted=100, delivered=90),
        _row("d", "e", "video_play", 2, attempted=150, delivered=140),
    ]
    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
    assert out[1]["d_attempted"] == 50
    assert out[1]["d_delivered"] == 50


def test_diff_epoch_reset_new_partition() -> None:
    rows = [
        _row("d", "e1", "video_play", 1, attempted=100),
        _row("d", "e2", "video_play", 2, attempted=5),
    ]
    out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)}
    assert out["e1"] == 100
    assert out["e2"] == 5


def test_diff_clamps_negative_on_reorder() -> None:
    rows = [
        _row("d", "e", "video_play", 1, attempted=100),
        _row("d", "e", "video_play", 2, attempted=80),
    ]
    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
    assert out[1]["d_attempted"] == 0


def test_rates_computes_both_formulas() -> None:
    out = _rates({"attempted": 150, "drop_capture": 0, "delivered": 140, "drop_undelivered": 10})
    assert out["track_success_rate"] == 1.0
    assert out["report_success_rate"] == 140 / 150


def test_rates_zero_denominator_yields_none() -> None:
    out = _rates({"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0})
    assert out["track_success_rate"] is None
    assert out["report_success_rate"] is None


def test_cn_day_beijing_boundary() -> None:
    # UTC 15:59 → 北京 23:59 同日;UTC 16:00 → 北京 次日 00:00
    assert _cn_day(datetime(2026, 7, 1, 15, 59, tzinfo=timezone.utc)) == "2026-07-01"
    assert _cn_day(datetime(2026, 7, 1, 16, 0, tzinfo=timezone.utc)) == "2026-07-02"


def test_diff_same_timestamp_ordered_by_id() -> None:
    ts = datetime(2026, 7, 1, 1, 0, tzinfo=timezone.utc)
    # 相同 created_at,乱序传入(id=2 在前);应按 id 排序 → id=1(cum100) 在前
    rows = [
        {"id": 2, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
         "app_ver": "v", "oem": "o", "os": "s",
         "attempted": 150, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
        {"id": 1, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
         "app_ver": "v", "oem": "o", "os": "s",
         "attempted": 100, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
    ]
    out = diff_snapshots(rows)
    assert out[0]["d_attempted"] == 100  # id=1 sorts first → its cumulative
    assert out[1]["d_attempted"] == 50   # id=2 second → 150-100


import pytest
from fastapi.testclient import TestClient

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.db.session import SessionLocal


@pytest.fixture()
def admin_client() -> TestClient:
    return TestClient(admin_app)


@pytest.fixture()
def admin_token() -> str:
    db = SessionLocal()
    try:
        if admin_repo.get_by_username(db, "health_admin") is None:
            admin_repo.create_admin(db, username="health_admin", password="pw", role="super_admin")
    finally:
        db.close()
    c = TestClient(admin_app)
    r = c.post("/admin/api/auth/login", json={"username": "health_admin", "password": "pw"})
    return r.json()["access_token"]


def _auth(t: str) -> dict:
    return {"Authorization": f"Bearer {t}"}


def _seed_two_snapshots(client: TestClient) -> None:
    for attempted, delivered in ((100, 90), (150, 140)):
        client.post("/api/v1/analytics/selfstat", json={
            "device_id": "d-health", "epoch_id": "e-health",
            "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
            "events": [{"event": "video_play", "attempted": attempted,
                        "drop_capture": 0, "delivered": delivered, "drop_undelivered": 0}],
        })


def test_health_overview_requires_auth(admin_client: TestClient) -> None:
    r = admin_client.get("/admin/api/analytics-health/overview",
                         params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"})
    assert r.status_code == 401


def test_health_overview(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
    _seed_two_snapshots(client)  # 播种走公开 app(ingest 端点在 app,不在 admin_app)
    r = admin_client.get(
        "/admin/api/analytics-health/overview",
        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"},
        headers=_auth(admin_token),
    )
    assert r.status_code == 200, r.text
    data = r.json()
    # 两快照累计 100→150,差分后 attempted 总 150(首快照 100 + 增量 50)
    assert data["attempted"] == 150
    assert data["track_success_rate"] == 1.0


def test_health_breakdown(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
    _seed_two_snapshots(client)  # 播种走公开 app
    r = admin_client.get(
        "/admin/api/analytics-health/breakdown",
        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z", "dim": "event"},
        headers=_auth(admin_token),
    )
    assert r.status_code == 200, r.text
    rows = r.json()
    assert any(row["key"] == "video_play" for row in rows)