184 lines
6.9 KiB
Python
184 lines
6.9 KiB
Python
"""Tests for the sovereignty metrics store and API routes.
|
|
|
|
Refs: #981
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from infrastructure.sovereignty_metrics import (
|
|
GRADUATION_TARGETS,
|
|
SovereigntyMetric,
|
|
SovereigntyMetricsStore,
|
|
emit_sovereignty_metric,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
"""Create a fresh sovereignty metrics store with a temp DB."""
|
|
return SovereigntyMetricsStore(db_path=tmp_path / "test_sov.db")
|
|
|
|
|
|
class TestSovereigntyMetricsStore:
|
|
def test_record_and_get_latest(self, store):
|
|
metric = SovereigntyMetric(metric_type="cache_hit_rate", value=0.42)
|
|
store.record(metric)
|
|
|
|
results = store.get_latest("cache_hit_rate", limit=10)
|
|
assert len(results) == 1
|
|
assert results[0]["value"] == 0.42
|
|
|
|
def test_get_latest_returns_most_recent_first(self, store):
|
|
for val in [0.1, 0.2, 0.3]:
|
|
store.record(SovereigntyMetric(metric_type="cache_hit_rate", value=val))
|
|
|
|
results = store.get_latest("cache_hit_rate", limit=10)
|
|
assert len(results) == 3
|
|
assert results[0]["value"] == 0.3 # most recent first
|
|
|
|
def test_get_latest_respects_limit(self, store):
|
|
for i in range(10):
|
|
store.record(SovereigntyMetric(metric_type="api_cost", value=float(i)))
|
|
|
|
results = store.get_latest("api_cost", limit=3)
|
|
assert len(results) == 3
|
|
|
|
def test_get_latest_filters_by_type(self, store):
|
|
store.record(SovereigntyMetric(metric_type="cache_hit_rate", value=0.5))
|
|
store.record(SovereigntyMetric(metric_type="api_cost", value=1.20))
|
|
|
|
results = store.get_latest("cache_hit_rate")
|
|
assert len(results) == 1
|
|
assert results[0]["value"] == 0.5
|
|
|
|
def test_get_summary_empty(self, store):
|
|
summary = store.get_summary()
|
|
assert "cache_hit_rate" in summary
|
|
assert summary["cache_hit_rate"]["current"] is None
|
|
assert summary["cache_hit_rate"]["phase"] == "pre-start"
|
|
|
|
def test_get_summary_with_data(self, store):
|
|
store.record(SovereigntyMetric(metric_type="cache_hit_rate", value=0.85))
|
|
store.record(SovereigntyMetric(metric_type="api_cost", value=0.08))
|
|
|
|
summary = store.get_summary()
|
|
assert summary["cache_hit_rate"]["current"] == 0.85
|
|
assert summary["cache_hit_rate"]["phase"] == "month3"
|
|
assert summary["api_cost"]["current"] == 0.08
|
|
assert summary["api_cost"]["phase"] == "month3"
|
|
|
|
def test_get_summary_graduation(self, store):
|
|
store.record(SovereigntyMetric(metric_type="cache_hit_rate", value=0.95))
|
|
summary = store.get_summary()
|
|
assert summary["cache_hit_rate"]["phase"] == "graduated"
|
|
|
|
def test_alert_on_high_api_cost(self, store):
|
|
"""API cost above threshold triggers an alert."""
|
|
with patch("infrastructure.sovereignty_metrics.settings") as mock_settings:
|
|
mock_settings.sovereignty_api_cost_alert_threshold = 1.00
|
|
mock_settings.db_busy_timeout_ms = 5000
|
|
store.record(SovereigntyMetric(metric_type="api_cost", value=2.50))
|
|
|
|
alerts = store.get_alerts(unacknowledged_only=True)
|
|
assert len(alerts) == 1
|
|
assert alerts[0]["alert_type"] == "api_cost_exceeded"
|
|
assert alerts[0]["value"] == 2.50
|
|
|
|
def test_no_alert_below_threshold(self, store):
|
|
"""API cost below threshold does not trigger an alert."""
|
|
with patch("infrastructure.sovereignty_metrics.settings") as mock_settings:
|
|
mock_settings.sovereignty_api_cost_alert_threshold = 1.00
|
|
mock_settings.db_busy_timeout_ms = 5000
|
|
store.record(SovereigntyMetric(metric_type="api_cost", value=0.50))
|
|
|
|
alerts = store.get_alerts(unacknowledged_only=True)
|
|
assert len(alerts) == 0
|
|
|
|
def test_acknowledge_alert(self, store):
|
|
with patch("infrastructure.sovereignty_metrics.settings") as mock_settings:
|
|
mock_settings.sovereignty_api_cost_alert_threshold = 0.50
|
|
mock_settings.db_busy_timeout_ms = 5000
|
|
store.record(SovereigntyMetric(metric_type="api_cost", value=1.00))
|
|
|
|
alerts = store.get_alerts(unacknowledged_only=True)
|
|
assert len(alerts) == 1
|
|
|
|
store.acknowledge_alert(alerts[0]["id"])
|
|
assert len(store.get_alerts(unacknowledged_only=True)) == 0
|
|
assert len(store.get_alerts(unacknowledged_only=False)) == 1
|
|
|
|
def test_metadata_preserved(self, store):
|
|
store.record(
|
|
SovereigntyMetric(
|
|
metric_type="cache_hit_rate",
|
|
value=0.5,
|
|
metadata={"source": "research_orchestrator"},
|
|
)
|
|
)
|
|
results = store.get_latest("cache_hit_rate")
|
|
assert results[0]["metadata"]["source"] == "research_orchestrator"
|
|
|
|
def test_summary_trend_data(self, store):
|
|
for v in [0.1, 0.2, 0.3]:
|
|
store.record(SovereigntyMetric(metric_type="cache_hit_rate", value=v))
|
|
|
|
summary = store.get_summary()
|
|
trend = summary["cache_hit_rate"]["trend"]
|
|
assert len(trend) == 3
|
|
assert trend[0]["v"] == 0.1 # oldest first (reversed)
|
|
assert trend[-1]["v"] == 0.3
|
|
|
|
def test_graduation_targets_complete(self):
|
|
"""All expected metric types have graduation targets."""
|
|
expected = {
|
|
"cache_hit_rate",
|
|
"api_cost",
|
|
"time_to_report",
|
|
"human_involvement",
|
|
"local_artifacts",
|
|
}
|
|
assert set(GRADUATION_TARGETS.keys()) == expected
|
|
|
|
|
|
class TestEmitSovereigntyMetric:
|
|
@pytest.mark.asyncio
|
|
async def test_emit_records_and_publishes(self, tmp_path):
|
|
"""emit_sovereignty_metric records to store and publishes event."""
|
|
with (
|
|
patch("infrastructure.sovereignty_metrics._store", None),
|
|
patch(
|
|
"infrastructure.sovereignty_metrics.DB_PATH",
|
|
tmp_path / "emit_test.db",
|
|
),
|
|
patch("infrastructure.events.bus.emit", new_callable=AsyncMock) as mock_emit,
|
|
):
|
|
await emit_sovereignty_metric("cache_hit_rate", 0.75, {"source": "test"})
|
|
|
|
mock_emit.assert_called_once()
|
|
call_args = mock_emit.call_args
|
|
assert call_args[0][0] == "sovereignty.metric.cache_hit_rate"
|
|
|
|
|
|
class TestSovereigntyMetricsRoutes:
|
|
def test_metrics_api_returns_200(self, client):
|
|
response = client.get("/sovereignty/metrics")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "metrics" in data
|
|
assert "alerts" in data
|
|
assert "targets" in data
|
|
|
|
def test_metrics_panel_returns_html(self, client):
|
|
response = client.get("/sovereignty/metrics/panel")
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
def test_alerts_api_returns_200(self, client):
|
|
response = client.get("/sovereignty/alerts")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "alerts" in data
|
|
assert "unacknowledged" in data
|