spb/company-atlas

1"""Admin API: auth, overview/quality/costs shapes, sensor actions, company creation, queue, reviews, event corrections, cache."""2from __future__ import annotations34import pytest5import test_api_support as support67client = support.client8fixture_data = support.fixture_data9ADMIN = support.ADMIN1011from companyatlas.api.common import cache12from companyatlas.db import execute, fetch_one, fetch_val, transaction1314pytestmark = pytest.mark.asyncio(loop_scope="session")15V = "/api/v1/admin"161718async def test_admin_requires_token(client):  # type: ignore[no-untyped-def]19    for path in ("/overview", "/connectors", "/sensors", "/companies", "/failures", "/queue", "/llm", "/reviews", "/quality", "/costs"):20        r = await client.get(f"{V}{path}")21        assert r.status_code == 401 and r.json() == {"detail": "admin token required"}, path22    assert (await client.get(f"{V}/overview", headers={"X-CA-Admin-Token": "wrong"})).status_code == 40123    assert (await client.post(f"{V}/cache/clear")).status_code == 401242526async def test_overview_connectors_quality_costs(client, fixture_data):  # type: ignore[no-untyped-def]27    r = await client.get(f"{V}/overview", headers=ADMIN)28    assert r.status_code == 200 and r.headers["cache-control"] == "no-store" and r.headers["x-ratelimit-tier"] == "admin"29    body = r.json()30    for key in ("companies_by_status", "sensors_by_status", "sensors_by_tier", "queue", "llm", "failures_24h_by_class", "fetch_rate_1h", "change_rate_1h",31                "meaningful_rate_1h", "storage", "workers", "cost_today"):32        assert key in body, key33    assert body["queue"]["dead"] >= 1 and body["failures_24h_by_class"].get("HTTP_5XX", 0) >= 1 and set(body["cost_today"]) >= {"fetch", "browser", "llm"}34    conns = (await client.get(f"{V}/connectors", headers=ADMIN)).json()["items"]35    mine = next(c for c in conns if c["id"] == fixture_data["connector"])36    assert mine["sensors_active"] == 3 and {"success_rate_24h", "avg_latency_ms", "change_rate_24h", "errors_24h", "last_run_at"} <= set(mine)37    q = (await client.get(f"{V}/quality", headers=ADMIN)).json()38    for key in ("coverage", "freshness", "duplicate_rate", "event_confidence_avg", "unknown_surfaces", "failed_sensors", "calibration"):39        assert key in q40    assert set(q["calibration"]) == {"correct", "duplicate", "noise", "misclassified"}41    c = (await client.get(f"{V}/costs?days=7", headers=ADMIN)).json()42    assert {"items", "per_1000_companies", "per_million_observations", "per_meaningful_event"} <= set(c)434445async def test_sensors_list_filters_and_actions(client, fixture_data):  # type: ignore[no-untyped-def]46    sid = fixture_data["sensor_beta"]47    r = await client.get(f"{V}/sensors", headers=ADMIN, params={"company": fixture_data["beta_slug"]})48    body = r.json()49    assert r.status_code == 200 and body["total"] == 1 and body["items"][0]["company"]["slug"] == fixture_data["beta_slug"]50    assert (await client.get(f"{V}/sensors", headers=ADMIN, params={"filter": "healthy", "connector": fixture_data["connector"]})).json()["total"] == 351    assert (await client.get(f"{V}/sensors", headers=ADMIN, params={"filter": "high_activity", "connector": fixture_data["connector"]})).json()["total"] == 352    assert (await client.get(f"{V}/sensors", headers=ADMIN, params={"filter": "nope"})).status_code == 42253    r = await client.post(f"{V}/sensors/{sid}/pause", headers=ADMIN)54    assert r.status_code == 200 and r.json()["sensor"]["status"] == "paused"55    assert (await client.get(f"{V}/sensors", headers=ADMIN, params={"status": "paused", "domain": f"{fixture_data['beta_slug']}.example"})).json()["total"] == 156    r = await client.post(f"{V}/sensors/{sid}/resume", headers=ADMIN)57    assert r.json()["sensor"]["status"] == "active" and r.json()["sensor"]["consecutive_failures"] == 058    r = await client.post(f"{V}/sensors/{sid}/set_interval", headers=ADMIN, json={"interval_s": 3600})59    assert r.json()["interval_s"] == 3600 and r.json()["sensor"]["current_interval_s"] == 3600 and r.json()["sensor"]["tier"] == "B"60    assert (await client.post(f"{V}/sensors/{sid}/set_interval", headers=ADMIN, json={})).status_code == 42261    assert (await client.post(f"{V}/sensors/{sid}/set_connector", headers=ADMIN, json={"connector_id": "nope-v9"})).status_code == 42262    assert (await client.post(f"{V}/sensors/{sid}/set_connector", headers=ADMIN, json={"connector_id": fixture_data["connector"]})).status_code == 20063    r = await client.post(f"{V}/sensors/{sid}/run_now", headers=ADMIN)64    assert r.status_code == 20065    r = await client.post(f"{V}/sensors/{sid}/rediscover", headers=ADMIN, json={"reason": "test"})66    assert r.status_code == 200 and r.json()["queued"]["key"].startswith(f"discover:{fixture_data['beta']}:")67    async with transaction() as conn:68        job = await fetch_one(conn, "select kind, status, payload from queue_jobs where id = :id", id=r.json()["queued"]["id"])69    assert job["kind"] == "discover" and job["status"] == "pending"70    r = await client.post(f"{V}/sensors/{sid}/retire", headers=ADMIN)71    assert r.json()["sensor"]["status"] == "retired" and r.json()["sensor"]["retired_at"]72    assert (await client.post(f"{V}/sensors/{sid}/resume", headers=ADMIN)).json()["sensor"]["status"] == "active"73    assert (await client.post(f"{V}/sensors/{sid}/explode", headers=ADMIN)).status_code == 40474    assert (await client.post(f"{V}/sensors/sen_nope/pause", headers=ADMIN)).status_code == 404757677async def test_companies_list_create_rediscover(client, fixture_data):  # type: ignore[no-untyped-def]78    r = await client.get(f"{V}/companies", headers=ADMIN, params={"onboarding_status": "active", "country": "ZZ"})79    assert r.status_code == 200 and r.json()["total"] == 2 and "onboarding_error" in r.json()["items"][0]80    suffix = fixture_data["suffix"]81    r = await client.post(f"{V}/companies", headers=ADMIN, json={"website": f"https://www.ztest-created-{suffix}.example/home?utm_source=x", "display_name": f"Ztest Created {suffix}",82                                                                  "country": "ZZ", "industries": [fixture_data["industry"]]})83    assert r.status_code == 201, r.text84    body = r.json()85    card = body["company"]86    assert card["canonical_domain"] == f"ztest-created-{suffix}.example" and card["website"] == f"https://www.ztest-created-{suffix}.example/home"87    assert card["slug"] == f"ztest-created-{suffix}" and card["onboarding_status"] == "pending" and body["queued"]["kind"] == "discover"88    async with transaction() as conn:89        assert await fetch_val(conn, "select status from queue_jobs where key = :k", k=f"discover:{card['id']}") == "pending"90    dup = await client.post(f"{V}/companies", headers=ADMIN, json={"website": f"http://ztest-created-{suffix}.example"})91    assert dup.status_code == 40992    assert (await client.post(f"{V}/companies", headers=ADMIN, json={"website": "ftp://ztest.example"})).status_code == 42293    assert (await client.post(f"{V}/companies", headers=ADMIN, json={"website": "https://localhost"})).status_code == 42294    assert (await client.post(f"{V}/companies", headers=ADMIN, json={"website": f"https://ztest-created2-{suffix}.example", "country": "QQ"})).status_code == 42295    assert (await client.post(f"{V}/companies", headers=ADMIN, json={"website": f"https://ztest-created2-{suffix}.example", "industries": ["nope-ind"]})).status_code == 42296    r = await client.post(f"{V}/companies/{card['slug']}/rediscover", headers=ADMIN)97    assert r.status_code == 200 and r.json()["queued"]["key"].startswith(f"discover:{card['id']}:")98    assert (await client.post(f"{V}/companies/nope/rediscover", headers=ADMIN)).status_code == 40499    async with transaction() as conn:      # remove the created company so the ZZ counts used by the other modules stay stable100        await execute(conn, "delete from queue_jobs where key like :k", k=f"discover:{card['id']}%")101        await execute(conn, "delete from companies where id = :id", id=card["id"])102    cache.clear()103104105async def test_failures_queue_llm_reviews(client, fixture_data):  # type: ignore[no-untyped-def]106    r = await client.get(f"{V}/failures", headers=ADMIN, params={"class": "http_5xx", "company": fixture_data["beta_slug"]})107    assert r.status_code == 200 and r.json()["total"] == 1 and r.json()["by_class"] == {"HTTP_5XX": 1} and "HTTP_5XX" in r.json()["classes"]108    q = (await client.get(f"{V}/queue", headers=ADMIN, params={"status": "dead"})).json()109    assert any(i["id"] == fixture_data["queue_dead"] for i in q["items"]) and any(c["status"] == "dead" for c in q["counts"])110    r = await client.post(f"{V}/queue/requeue-dead", headers=ADMIN, json={"kind": "run_sensor"})111    assert r.status_code == 200 and r.json()["requeued"] >= 1112    async with transaction() as conn:113        assert await fetch_val(conn, "select status from queue_jobs where id = :id", id=fixture_data["queue_dead"]) == "pending"114    llm = (await client.get(f"{V}/llm", headers=ADMIN)).json()115    assert {"items", "stats"} <= set(llm) and "budget" in llm["stats"]116    reviews = (await client.get(f"{V}/reviews", headers=ADMIN, params={"kind": "major_event"})).json()117    assert any(x["id"] == fixture_data["review"] for x in reviews["items"]) and reviews["open_by_kind"].get("major_event", 0) >= 1118    r = await client.post(f"{V}/reviews/{fixture_data['review']}", headers=ADMIN, json={"resolution": "accepted", "label": "correct", "note": "looks right"})119    assert r.status_code == 200 and r.json()["review"]["status"] == "accepted" and r.json()["review"]["payload"]["resolution"]["label"] == "correct"120    assert (await client.post(f"{V}/reviews/{fixture_data['review']}", headers=ADMIN, json={"resolution": "rejected"})).status_code == 409121    assert (await client.post(f"{V}/reviews/rev_nope", headers=ADMIN, json={"resolution": "rejected"})).status_code == 404122    assert (await client.get(f"{V}/quality", headers=ADMIN)).json()["calibration"]["correct"] >= 1123124125async def test_event_retract_restore_and_cache_clear(client, fixture_data):  # type: ignore[no-untyped-def]126    eid = fixture_data["ev_product"]127    r = await client.post(f"{V}/events/{eid}/retract", headers=ADMIN, json={"reason": "synthetic correction"})128    assert r.status_code == 200 and r.json()["event"]["status"] == "retracted" and r.json()["event"]["retracted_reason"] == "synthetic correction"129    assert (await client.get(f"/api/v1/events/{eid}")).json()["status"] == "retracted"130    assert all(e["id"] != eid for e in (await client.get("/api/v1/live?limit=50")).json()["items"])131    assert (await client.post(f"{V}/events/{eid}/retract", headers=ADMIN, json={"reason": "x"})).status_code == 422132    r = await client.post(f"{V}/events/{eid}/restore", headers=ADMIN)133    assert r.status_code == 200 and r.json()["event"]["status"] == "active"134    audit = r.json()["event"]["payload"]["_audit"]135    assert [a["action"] for a in audit] == ["retract", "restore"] and audit[0]["reason"] == "synthetic correction"136    assert (await client.post(f"{V}/events/evt_nope/restore", headers=ADMIN)).status_code == 404137    r = await client.post(f"{V}/cache/clear", headers=ADMIN)138    assert r.status_code == 200 and r.json()["ok"] is True139    assert (await client.post(f"{V}/cache/clear?prefix=pulse", headers=ADMIN)).json()["cleared"] == "pulse"140