import sys from pathlib import Path import pytest from fastapi.testclient import TestClient sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import app as analytics # noqa: E402 def test_build_query_forces_tenant_and_dimension_whitelist(): sql, params = analytics.build_query( tenant="379", dataset="aud0.events", metrics=["event_count"], dimensions=["event_type"], filters={"app_id": "portal"}, from_time="2026-01-01T00:00:00Z", to_time=None, limit=50, ) assert "tenant_id = {tenant:String}" in sql assert "dataset_id = {dataset:String}" in sql assert "dim_values[indexOf(dim_names, 'event_type')]" in sql assert params["tenant"] == "379" assert params["dataset"] == "aud0.events" assert params["metric0"] == "event_count" assert params["filter0"] == "portal" with pytest.raises(analytics.AnalyticsError): analytics.build_query( tenant="379", dataset=None, metrics=["event_count"], dimensions=["drop table"], filters=None, from_time=None, to_time=None, limit=10, ) def test_data_routes_require_a_verified_identity(): client = TestClient(analytics.app) assert client.get("/api/0pt/analytics/datasets").status_code == 401 assert client.get("/api/0pt/analytics/metrics").status_code == 401 def test_health_is_always_a_liveness_answer(): client = TestClient(analytics.app) response = client.get("/health") assert response.status_code == 200 assert "providers" in response.json()