"""Unit and integration tests for Model Usage & Performance Analytics (#651).""" from __future__ import annotations import os import tempfile import unittest from starlette.testclient import TestClient import control_plane_db from webui.analytics_loader import ( ANALYTICS_SCHEMA_VERSION, compute_percentile, load_analytics, record_usage, ) from webui.app import create_app from webui import console_redaction class AnalyticsLoaderTest(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.temp_dir.name, "test_control_plane.sqlite3") os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path self.db = control_plane_db.ControlPlaneDB(db_path=self.db_path) def tearDown(self) -> None: self.temp_dir.cleanup() def test_compute_percentile(self) -> None: self.assertIsNone(compute_percentile([], 50.0)) self.assertEqual(compute_percentile([100], 50.0), 100.0) # 2 elements: [100, 200] self.assertEqual(compute_percentile([100, 200], 50.0), 150.0) # 100 elements: 1..100 vals = list(range(1, 101)) self.assertEqual(compute_percentile(vals, 50.0), 50.5) self.assertAlmostEqual(compute_percentile(vals, 90.0), 90.1) def test_record_and_aggregate_usage(self) -> None: # Record event 1 (complete data) u1 = record_usage( db_path=self.db_path, remote="dadeschools", org="Scaled-Tech-Consulting", repo="Gitea-Tools", role="author", model="gemini-3.6-flash", issue_number=651, stage="implementation", input_tokens=1000, output_tokens=500, estimated_cost_usd=0.0015, latency_ms=200, duration_ms=3000, metadata={"secret_key": "secret123", "note": "token=secret123"}, ) self.assertGreater(u1, 0) # Record event 2 (missing tokens and cost -> unknown) u2 = record_usage( db_path=self.db_path, remote="dadeschools", org="Scaled-Tech-Consulting", repo="Gitea-Tools", role="reviewer", model="claude-3-5-sonnet", pr_number=846, stage="review", latency_ms=500, duration_ms=6000, ) self.assertGreater(u2, u1) snapshot = load_analytics( db_path=self.db_path, remote="dadeschools", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(snapshot.ok) self.assertEqual(snapshot.schema_version, ANALYTICS_SCHEMA_VERSION) self.assertEqual(snapshot.total_events, 2) # Verify overall summary summary = snapshot.overall_summary self.assertEqual(summary.total_events, 2) self.assertEqual(summary.events_with_tokens, 1) self.assertEqual(summary.total_tokens, 1500) self.assertEqual(summary.events_with_cost, 1) self.assertEqual(summary.estimated_cost_usd, 0.0015) self.assertEqual(summary.events_with_latency, 2) self.assertEqual(summary.latency_p50_ms, 350.0) # Verify missing data handling (AC 3: not zero-fabricated) reviewer_model = snapshot.by_model.get("claude-3-5-sonnet") self.assertIsNotNone(reviewer_model) self.assertEqual(reviewer_model.total_events, 1) self.assertEqual(reviewer_model.events_with_tokens, 0) self.assertIsNone(reviewer_model.total_tokens) self.assertEqual(reviewer_model.display_tokens, "Unknown") self.assertEqual(reviewer_model.events_with_cost, 0) self.assertIsNone(reviewer_model.estimated_cost_usd) self.assertEqual(reviewer_model.display_cost, "Unknown") # Verify redaction (AC 4) e1 = [e for e in snapshot.events if e.usage_id == u1][0] self.assertIsNotNone(e1.metadata) self.assertNotIn("secret123", e1.metadata) self.assertIn("[REDACTED]", e1.metadata) def test_missing_db_fail_soft(self) -> None: invalid_path = "/nonexistent_path_dir/db.sqlite3" snapshot = load_analytics(db_path=invalid_path) self.assertFalse(snapshot.ok) self.assertIn("control_plane_db_unavailable", snapshot.reason) self.assertEqual(snapshot.overall_summary.display_tokens, "Unknown") class AnalyticsWebUITest(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.temp_dir.name, "test_webui.sqlite3") os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path self.app = create_app() self.client = TestClient(self.app) record_usage( db_path=self.db_path, remote="dadeschools", org="Scaled-Tech-Consulting", repo="Gitea-Tools", role="author", model="gemini-3.6-flash", issue_number=651, stage="implementation", input_tokens=2000, output_tokens=1000, estimated_cost_usd=0.003, latency_ms=150, duration_ms=2500, ) def tearDown(self) -> None: self.temp_dir.cleanup() def test_analytics_html_route(self) -> None: response = self.client.get("/analytics") self.assertEqual(response.status_code, 200) self.assertIn("Model Usage & Performance Analytics", response.text) self.assertIn("gemini-3.6-flash", response.text) self.assertIn("3,000", response.text) def test_analytics_api_route(self) -> None: response = self.client.get("/api/v1/analytics") self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["ok"]) self.assertEqual(data["total_events"], 1) self.assertIn("gemini-3.6-flash", data["by_model"]) def test_analytics_ingest_endpoint(self) -> None: payload = { "remote": "dadeschools", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "role": "reviewer", "model": "claude-3-5-sonnet", "pr_number": 846, "stage": "review", "input_tokens": 500, "output_tokens": 100, "latency_ms": 400, "metadata": "Review note token=secret456", } response = self.client.post("/api/v1/analytics/usage", json=payload) self.assertEqual(response.status_code, 201) res_json = response.json() self.assertTrue(res_json["ok"]) self.assertGreater(res_json["usage_id"], 0) # Check that it appears in GET /api/v1/analytics res2 = self.client.get("/api/v1/analytics") self.assertEqual(res2.status_code, 200) data2 = res2.json() self.assertEqual(data2["total_events"], 2) if __name__ == "__main__": unittest.main()