wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 4 Packages Projects Releases Wiki Activity

feat(sre): add 24h automation SLO scorecard

Browse Source
This commit is contained in:
Your Name
2026-07-19 01:47:35 +08:00
parent fa51ab9022
commit 8cd01cb278
5 changed files with 744 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
apps/api/tests/test_automation_slo_scorecard.py Normal file
View File

@@ -0,0 +1,275 @@
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1 import agents
from src.services import automation_slo_scorecard as scorecard_module
from src.services.automation_slo_scorecard import (
build_automation_slo_scorecard,
read_automation_slo_scorecard,
)
def _sample(
start: datetime,
*,
lane: str,
action_seconds: int,
resolve_seconds: int,
positive: bool,
) -> dict:
return {
"lane": lane,
"received_at": start,
"first_action_at": start + timedelta(seconds=action_seconds),
"resolved_at": start + timedelta(seconds=resolve_seconds),
"false_positive_observed": True,
"false_positive": positive,
"recurrence_observed": True,
"recurrence": positive,
"human_intervention": positive,
"execution_observed": True,
"verifier_observed": True,
"verifier_passed": not positive,
"rollback_performed": positive,
"freshness_observed": True,
"freshness_passed": not positive,
"asset_identity_resolved": not positive,
}
def test_scorecard_computes_all_required_metrics_and_provider_cost() -> None:
end = datetime(2026, 7, 19, 4, 0, tzinfo=UTC)
start = end - timedelta(hours=24)
scorecard = build_automation_slo_scorecard(
project_id="awoooi",
incident_samples=[
_sample(
start + timedelta(hours=1),
lane="kubernetes_workload",
action_seconds=10,
resolve_seconds=100,
positive=False,
),
_sample(
start + timedelta(hours=2),
lane="windows_vmware",
action_seconds=20,
resolve_seconds=200,
positive=True,
),
],
run_summary={
"total_runs": 5,
"non_shadow_runs": 3,
"controlled_runs_with_steps": 2,
"completed_runs": 4,
"failed_runs": 1,
},
provider_cost_rows=[
{
"provider": "ollama_gcp_a",
"model": "qwen3",
"call_count": 3,
"prompt_tokens": 1200,
"completion_tokens": 300,
"cost_usd": "0",
},
{
"provider": "gemini",
"model": "gemini-critic",
"call_count": 1,
"prompt_tokens": 100,
"completion_tokens": 50,
"cost_usd": "0.0123",
},
],
window_start=start,
window_end=end,
)
assert scorecard["status"] == (
"runtime_aggregate_ready_same_run_verification_pending"
)
assert scorecard["runtime_closed"] is False
assert scorecard["metrics"] == {
"mtta_seconds": 15.0,
"mttr_seconds": 150.0,
"false_positive_rate": 0.5,
"recurrence_rate": 0.5,
"human_intervention_rate": 0.5,
"verifier_pass_rate": 0.5,
"rollback_rate": 0.5,
"signal_freshness_slo_pass_rate": 0.5,
"asset_coverage_percent": 50.0,
}
assert scorecard["missing_metrics"] == []
assert scorecard["coverage_blockers"] == []
assert len(scorecard["lane_scorecards"]) == 2
assert scorecard["runs"]["controlled_runs_with_steps"] == 2
assert scorecard["provider_usage"] == {
"rows": [
{
"provider": "ollama_gcp_a",
"model": "qwen3",
"call_count": 3,
"prompt_tokens": 1200,
"completion_tokens": 300,
"cost_usd": "0.0000",
},
{
"provider": "gemini",
"model": "gemini-critic",
"call_count": 1,
"prompt_tokens": 100,
"completion_tokens": 50,
"cost_usd": "0.0123",
},
],
"call_count": 4,
"prompt_tokens": 1300,
"completion_tokens": 350,
"cost_usd": "0.0123",
}
assert scorecard["evidence"] == {
"source": "postgresql_read_only_aggregate",
"same_run_receipt_verification": "pending",
"no_false_green": True,
}
def test_scorecard_does_not_false_green_missing_runtime_samples() -> None:
end = datetime(2026, 7, 19, 4, 0, tzinfo=UTC)
scorecard = build_automation_slo_scorecard(
project_id="awoooi",
incident_samples=[],
run_summary={},
provider_cost_rows=[],
window_start=end - timedelta(hours=24),
window_end=end,
)
assert scorecard["status"] == "no_runtime_sample_fail_closed"
assert scorecard["runtime_closed"] is False
assert set(scorecard["missing_metrics"]) == {
"asset_coverage_percent",
"false_positive_rate",
"human_intervention_rate",
"mtta_seconds",
"mttr_seconds",
"recurrence_rate",
"rollback_rate",
"signal_freshness_slo_pass_rate",
"verifier_pass_rate",
}
assert scorecard["provider_usage"]["cost_usd"] == "0.0000"
def test_scorecard_endpoint_is_project_scoped_and_read_only(monkeypatch) -> None:
readback = AsyncMock(
return_value={
"schema_version": "sre_automation_slo_scorecard_v1",
"project_id": "awoooi",
"status": "partial_missing_receipt_coverage",
"runtime_closed": False,
}
)
monkeypatch.setattr(agents, "read_automation_slo_scorecard", readback)
app = FastAPI()
app.include_router(agents.router, prefix="/api/v1")
response = TestClient(app).get(
"/api/v1/agents/sre-automation-slo-scorecard-24h",
headers={"X-Project-ID": "awoooi"},
)
assert response.status_code == 200
assert response.json()["runtime_closed"] is False
readback.assert_awaited_once_with(project_id="awoooi")
@pytest.mark.asyncio
async def test_runtime_reader_executes_only_three_bounded_read_queries(
monkeypatch,
) -> None:
end = datetime(2026, 7, 19, 4, 0, tzinfo=UTC)
sample = _sample(
end - timedelta(hours=1),
lane="kubernetes_workload",
action_seconds=10,
resolve_seconds=100,
positive=False,
)
class Result:
def __init__(self, rows):
self.rows = rows
def mappings(self):
return self
def all(self):
return self.rows
def first(self):
return self.rows[0] if self.rows else None
class DB:
def __init__(self):
self.statements = []
self.results = [
Result([sample]),
Result(
[
{
"total_runs": 1,
"non_shadow_runs": 1,
"controlled_runs_with_steps": 1,
"completed_runs": 1,
"failed_runs": 0,
}
]
),
Result([]),
]
async def execute(self, statement, _parameters):
self.statements.append(str(statement))
return self.results[len(self.statements) - 1]
class Context:
def __init__(self, db):
self.db = db
async def __aenter__(self):
return self.db
async def __aexit__(self, *_args):
return False
db = DB()
monkeypatch.setattr(
scorecard_module,
"get_db_context",
lambda project_id: Context(db),
)
scorecard = await read_automation_slo_scorecard(
project_id="awoooi",
window_end=end,
)
assert len(db.statements) == 3
assert all(
statement.lstrip().startswith(("WITH", "SELECT")) for statement in db.statements
)
assert all(
keyword not in " ".join(statement.upper().split())
for statement in db.statements
for keyword in (" INSERT ", " UPDATE ", " DELETE ", " ALTER ", " DROP ")
)
assert scorecard["metrics"]["mtta_seconds"] == 10.0
assert scorecard["runtime_closed"] is False

14
apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py
View File

@@ -116,10 +116,10 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() -
"total_items": 18,
"by_priority": {"P0": 16, "P1": 2},
"by_status": {
"source_implemented_runtime_pending": 15,
"planned": 3,
"source_implemented_runtime_pending": 16,
"planned": 2,
},
"source_implemented_items": 15,
"source_implemented_items": 16,
"runtime_closed_items": 0,
"program_completion_percent": 0,
"asset_coverage_status": "partial",
@@ -158,6 +158,12 @@ def test_ledger_links_confirmed_runtime_gaps_without_false_closure() -> None:
items["AIA-SRE-014"]["confirmed_truth"]
)
assert "host99 Agent99" in " ".join(items["AIA-SRE-017"]["runtime_gaps"])
assert items["AIA-SRE-016"]["status"] == (
"source_implemented_runtime_pending"
)
assert "three project-scoped aggregate queries" in " ".join(
items["AIA-SRE-016"]["confirmed_truth"]
)
assert items["AIA-SRE-009"]["status"] == (
"source_implemented_runtime_pending"
)
@@ -200,7 +206,7 @@ def test_projection_keeps_program_asset_and_runtime_truth_separate() -> None:
payload = load_sre_k3s_controlled_automation_work_items()
projection = build_sre_k3s_program_projection(payload)
assert projection["rollups"]["source_implemented_items"] == 15
assert projection["rollups"]["source_implemented_items"] == 16
assert projection["rollups"]["runtime_closed_items"] == 0
assert projection["rollups"]["program_completion_percent"] == 0
assert projection["domain_count"] == 8