guke
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""埋点健康度聚合:纯差分函数 + 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
|