Spaces:
Paused
Paused
File size: 8,146 Bytes
a2bbd94 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | """Async run lifecycle, the query API (cache behavior + routing), 409 gating,
containers/forensic drill-down, review endpoints and performance SLOs —
all through the real HTTP API."""
from __future__ import annotations
import time
import pytest
from fastapi.testclient import TestClient
@pytest.fixture(scope="module")
def client(mega_env):
from app_backend import app
with TestClient(app) as c:
yield c
@pytest.fixture(scope="module")
def api_run(client, mega_env):
"""Create a fresh run via the API and wait for the async lifecycle to publish."""
pop_id = mega_env["meta"]["population_id"]
res = client.post("/api/megaeval/runs",
json={"population_id": pop_id, "model_version": "model-api",
"worker_delay_s": 0.05, "workers": 4})
assert res.status_code == 200
run = res.json()
assert run["status"] in ("queued", "running")
seen_statuses = {run["status"]}
deadline = time.time() + 60
while time.time() < deadline:
r = client.get(f"/api/megaeval/runs/{run['run_id']}").json()
seen_statuses.add(r["status"])
if r["status"] in ("published", "failed"):
break
time.sleep(0.2)
assert r["status"] == "published", r.get("error")
return {"run": r, "seen_statuses": seen_statuses}
def test_async_lifecycle_and_progress(api_run):
r = api_run["run"]
assert api_run["seen_statuses"] & {"queued", "running", "reducing", "materializing"}
assert r["percent"] == 100.0
assert r["objects_processed"] == r["objects_total"] > 0
assert r["throughput_objs_per_s"] > 0
assert r["headline"]["n"] == r["objects_total"]
lin = r["lineage"]
for field in ("evaluation_id", "dataset_version", "model_version", "model_checkpoint",
"label_version", "evaluator_code_version", "metric_version",
"threshold_config", "sampling_config", "seed", "hardware", "timestamp"):
assert field in lin
def test_results_blocked_until_published(client, mega_env):
pop_id = mega_env["meta"]["population_id"]
res = client.post("/api/megaeval/runs",
json={"population_id": pop_id, "model_version": "model-slow",
"worker_delay_s": 0.4, "workers": 2})
run_id = res.json()["run_id"]
blocked = client.get(f"/api/megaeval/runs/{run_id}/funnel")
assert blocked.status_code == 409
deadline = time.time() + 90
while time.time() < deadline:
if client.get(f"/api/megaeval/runs/{run_id}").json()["status"] == "published":
break
time.sleep(0.3)
assert client.get(f"/api/megaeval/runs/{run_id}/funnel").status_code == 200
def test_query_api_cache_and_latency_slos(client, api_run):
run_id = api_run["run"]["run_id"]
body = {"evaluation_id": run_id,
"filters": {"class": ["pedestrian"], "lighting": ["night"]},
"metrics": ["n", "recall", "precision", "mean_iou"],
"group_by": []}
r1 = client.post("/api/evaluations/query", json=body).json()
assert r1["meta"]["source"] == "cube"
assert r1["meta"]["cache_hit"] is False
assert r1["meta"]["exact"] is True
assert r1["rows"][0]["n"] > 0
assert r1["meta"]["latency_ms"] < 3000 # filtered cohort query SLO
r2 = client.post("/api/evaluations/query", json=body).json()
assert r2["meta"]["cache_hit"] is True
assert r2["meta"]["source"] == "cache"
assert r2["meta"]["latency_ms"] < 100 # cached query SLO
assert r2["rows"] == r1["rows"]
dash = client.post("/api/evaluations/query",
json={"evaluation_id": run_id, "group_by": ["class"],
"metrics": ["n", "precision", "recall", "f1"]}).json()
assert dash["meta"]["latency_ms"] < 2000 # aggregate dashboard SLO
assert len(dash["rows"]) == 6
stats = client.get("/api/megaeval/cache").json()
assert stats["hits"] >= 1
def test_query_sketch_metrics_marked_approximate(client, api_run):
run_id = api_run["run"]["run_id"]
res = client.post("/api/evaluations/query",
json={"evaluation_id": run_id,
"metrics": ["n", "recall", "confidence_p50", "iou_p90"]}).json()
assert res["meta"]["exact"] is False
assert set(res["meta"]["approximate_fields"]) == {"confidence_p50", "iou_p90"}
assert 0 <= res["rows"][0]["confidence_p50"] <= 1
def test_containers_sort_presets_and_forensic_drilldown(client, api_run):
run_id = api_run["run"]["run_id"]
worst = client.get(f"/api/megaeval/runs/{run_id}/containers",
params={"sort": "worst_recall", "limit": 20}).json()
assert worst["total"] > 0 and len(worst["rows"]) == 20
recalls = [r["recall"] for r in worst["rows"] if r["recall"] is not None]
assert recalls == sorted(recalls)
risky = client.get(f"/api/megaeval/runs/{run_id}/containers",
params={"sort": "highest_risk", "limit": 5}).json()["rows"]
risks = [r["risk_score"] for r in risky]
assert risks == sorted(risks, reverse=True)
assert risky[0]["status"] in ("ok", "warn", "critical")
t0 = time.perf_counter()
objs = client.get(
f"/api/megaeval/runs/{run_id}/containers/{risky[0]['container_id']}/objects").json()
elapsed_ms = (time.perf_counter() - t0) * 1000
assert elapsed_ms < 1000 # object investigation SLO
assert objs["objects"], "forensic drill-down must return annotation-level rows"
outcomes = {o["outcome"] for o in objs["objects"]}
assert outcomes <= {"TP", "FN", "FP", "LOCALIZATION"}
gt_objs = [o for o in objs["objects"] if o["outcome"] != "FP"]
assert len(gt_objs) == risky[0]["n_objects"]
def test_review_endpoints_full_cycle(client, api_run):
run_id = api_run["run"]["run_id"]
state = client.get(f"/api/megaeval/runs/{run_id}/review").json()
assert state["planned"] is False
plan = client.post(f"/api/megaeval/runs/{run_id}/review/plan",
json={"target_n": 600}).json()
assert plan["planned"] and not plan["executed"]
assert plan["funnel"]["statistically_selected"] > 0
done = client.post(f"/api/megaeval/runs/{run_id}/review/execute").json()
assert done["executed"]
res = done["results"]
for metric in ("precision", "recall"):
est = res[metric]
assert est["ci_low"] < est["estimate"] < est["ci_high"]
funnel = client.get(f"/api/megaeval/runs/{run_id}/funnel").json()
stages = {s["stage"]: s["count"] for s in funnel["stages"]}
assert stages["Human Verified"] == done["funnel"]["reviewed"]
assert funnel["estimated_precision"]["estimate"] is not None
def test_error_search_compare_shift_why_similarity_endpoints(client, api_run, mega_env):
run_id = api_run["run"]["run_id"]
es = client.post("/api/megaeval/errors/search",
json={"run_id": run_id, "error_types": ["FN"],
"filters": {"class": ["pedestrian"]},
"risk_min": 0.5}).json()
assert es["matched_errors"] > 0 and es["worst_containers"]
cmp = client.post("/api/megaeval/compare",
json={"candidate_run_id": mega_env["bad"].run_id,
"baseline_run_id": run_id}).json()
assert cmp["recommendation"] == "DO_NOT_PROMOTE"
sh = client.get(f"/api/megaeval/runs/{run_id}/shift").json()
assert sh["shifts"]
why = client.post("/api/megaeval/why",
json={"run_id": run_id, "filters": {"lighting": ["night"]},
"metric": "recall"}).json()
assert why["failure_count"] > 0 and why["factors"]
cid = es["worst_containers"][0]["container_id"]
sim = client.post("/api/megaeval/similarity",
json={"run_id": run_id, "container_id": cid, "k": 5}).json()
assert len(sim["results"]) == 5
dist = client.get(f"/api/megaeval/runs/{run_id}/distributions").json()
assert dist["exact"] is False
assert "confidence" in dist and "50" in dist["confidence"]["percentiles"]
assert dist["containers_exact"] > 0
|