197 lines
6.9 KiB
Python
197 lines
6.9 KiB
Python
"""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()
|