sysadminandClaude Opus 4.8
b179610e7f
Address review blockers and medium/low findings: - F1: HTML-escape all dynamic analytics fields (html.escape quote=True) - F2: Gate POST /api/v1/analytics/usage through console_authz record_analytics_usage (fail closed for unauthenticated / Phase 1) - F3: Enforce usage_events retention (max rows + max age) - F4: Coerce None remote/org/repo to empty strings in load_analytics Add tests for XSS escaping, unauthorized ingest deny, retention, and None scope coercion. Document authz + retention in analytics guide. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
253 lines
9.3 KiB
Python
253 lines
9.3 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_unauthorized_denied(self) -> None:
|
|
"""F2: unauthenticated POST must not write the control-plane DB."""
|
|
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, 403)
|
|
res_json = response.json()
|
|
self.assertFalse(res_json.get("ok", True))
|
|
self.assertEqual(res_json.get("error"), "unauthorized")
|
|
authorization = res_json.get("authorization") or {}
|
|
self.assertFalse(authorization.get("allowed"))
|
|
self.assertFalse(authorization.get("execution_enabled"))
|
|
|
|
# No new row written
|
|
res2 = self.client.get("/api/v1/analytics")
|
|
self.assertEqual(res2.status_code, 200)
|
|
self.assertEqual(res2.json()["total_events"], 1)
|
|
|
|
def test_html_escapes_script_bearing_model_role_stage(self) -> None:
|
|
"""F1: stored XSS — dynamic model/role/stage must render escaped."""
|
|
xss = '<script>alert(1)</script>'
|
|
record_usage(
|
|
db_path=self.db_path,
|
|
remote="dadeschools",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
role=xss,
|
|
model=xss,
|
|
stage=xss,
|
|
issue_number=999,
|
|
status="success",
|
|
)
|
|
response = self.client.get("/analytics")
|
|
self.assertEqual(response.status_code, 200)
|
|
# Raw tag must not appear; escaped form must.
|
|
self.assertNotIn("<script>alert(1)</script>", response.text)
|
|
self.assertIn("<script>alert(1)</script>", response.text)
|
|
|
|
def test_load_analytics_coerces_none_scope(self) -> None:
|
|
"""F4: None remote/org/repo become empty strings, never None."""
|
|
snapshot = load_analytics(db_path=self.db_path, remote=None, org=None, repo=None)
|
|
self.assertIsInstance(snapshot.remote, str)
|
|
self.assertIsInstance(snapshot.org, str)
|
|
self.assertIsInstance(snapshot.repo, str)
|
|
self.assertEqual(snapshot.remote, "")
|
|
self.assertEqual(snapshot.org, "")
|
|
self.assertEqual(snapshot.repo, "")
|
|
|
|
def test_usage_events_retention_max_rows(self) -> None:
|
|
"""F3: record_usage_event enforces USAGE_EVENTS_MAX_ROWS."""
|
|
db = control_plane_db.ControlPlaneDB(db_path=self.db_path)
|
|
original_max = db.USAGE_EVENTS_MAX_ROWS
|
|
try:
|
|
db.USAGE_EVENTS_MAX_ROWS = 3
|
|
for i in range(5):
|
|
db.record_usage_event(
|
|
remote="dadeschools",
|
|
org="org",
|
|
repo="repo",
|
|
role="author",
|
|
model=f"model-{i}",
|
|
stage="test",
|
|
)
|
|
rows = db.query_usage_events(limit=100)
|
|
self.assertLessEqual(len(rows), 3)
|
|
# Newest three retained
|
|
models = {r["model"] for r in rows}
|
|
self.assertEqual(models, {"model-2", "model-3", "model-4"})
|
|
finally:
|
|
db.USAGE_EVENTS_MAX_ROWS = original_max
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|