guke
34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
"""自报计数上报端点测试。"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def _payload(**over) -> dict:
|
|
base = {
|
|
"device_id": "dev-1", "epoch_id": "ep-1", "sent_at": 1700000000000,
|
|
"app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
|
|
"batches_attempted": 10, "batches_ok": 9, "batches_fail": 1,
|
|
"retries": 1, "queue_depth": 2,
|
|
"events": [
|
|
{"event": "video_play", "attempted": 100, "drop_capture": 1,
|
|
"delivered": 95, "drop_undelivered": 2},
|
|
],
|
|
}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def test_selfstat_ingest_ok(client: TestClient) -> None:
|
|
r = client.post("/api/v1/analytics/selfstat", json=_payload())
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert isinstance(body["snapshot_id"], int)
|
|
|
|
|
|
def test_selfstat_ingest_empty_events(client: TestClient) -> None:
|
|
r = client.post("/api/v1/analytics/selfstat", json=_payload(events=[]))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["ok"] is True
|