feat: report compatibility evidence

This commit is contained in:
Sergey Antropoff
2026-07-12 23:45:55 +03:00
parent a64174910c
commit 355bb23d5e
4 changed files with 167 additions and 1 deletions
+3
View File
@@ -34,6 +34,9 @@ class HandlerRegistry:
def get(self, path: str, verb: str) -> Handler | None:
return self._handlers.get((path, verb.upper()))
def keys(self) -> frozenset[tuple[str, str]]:
return frozenset(self._handlers)
def register_contract_routes(
app: FastAPI,
+99
View File
@@ -0,0 +1,99 @@
"""Evidence-based compatibility accounting and reporting."""
from __future__ import annotations
from dataclasses import dataclass
from app.contracts.model import Snapshot
MethodKey = tuple[str, str]
@dataclass(frozen=True, slots=True)
class CompatibilityReport:
source_version: str
declared: frozenset[MethodKey]
schema_only: frozenset[MethodKey]
implemented: frozenset[MethodKey]
observed: frozenset[MethodKey]
verified: frozenset[MethodKey]
def as_json(self) -> dict[str, object]:
levels = {
"declared": self.declared,
"schema_only": self.schema_only,
"implemented": self.implemented,
"observed": self.observed,
"verified": self.verified,
}
total = len(self.declared)
return {
"source_version": self.source_version,
"total_declared": total,
"levels": {
name: {
"count": len(methods),
"score": len(methods) / total if total else 1.0,
"methods": [f"{verb} {path}" for path, verb in sorted(methods)],
}
for name, methods in levels.items()
},
"groups": self._groups(),
}
def _groups(self) -> dict[str, dict[str, int]]:
groups: dict[str, dict[str, int]] = {}
for path, verb in self.declared:
group = path.strip("/").split("/", 1)[0] or "root"
counters = groups.setdefault(group, {"declared": 0, "implemented": 0, "verified": 0})
counters["declared"] += 1
counters["implemented"] += int((path, verb) in self.implemented)
counters["verified"] += int((path, verb) in self.verified)
return dict(sorted(groups.items()))
def as_markdown(self) -> str:
levels = {
"declared": self.declared,
"schema_only": self.schema_only,
"implemented": self.implemented,
"observed": self.observed,
"verified": self.verified,
}
total = len(self.declared)
lines = [
"# Compatibility report",
"",
"| Level | Count | Score |",
"|---|---:|---:|",
]
for name, methods in levels.items():
score = len(methods) / total if total else 1.0
lines.append(f"| {name} | {len(methods)} | {score:.2%} |")
return "\n".join(lines)
def build_report(
snapshot: Snapshot,
*,
implemented: frozenset[MethodKey] = frozenset(),
observed: frozenset[MethodKey] = frozenset(),
verified: frozenset[MethodKey] = frozenset(),
) -> CompatibilityReport:
declared = frozenset(
(path.path, method.verb) for path in snapshot.paths for method in path.methods
)
for name, evidence in {
"implemented": implemented,
"observed": observed,
"verified": verified,
}.items():
if not evidence <= declared:
raise ValueError(f"{name} evidence references undeclared methods")
return CompatibilityReport(
source_version=snapshot.source_version,
declared=declared,
schema_only=declared - implemented,
implemented=implemented,
observed=observed,
verified=verified,
)
+9 -1
View File
@@ -7,6 +7,7 @@ from fastapi import FastAPI
from app.api.errors import ApiError, api_error_handler, unhandled_exception_handler
from app.api.middleware import RequestContextMiddleware
from app.api.registry import HandlerRegistry, register_contract_routes
from app.compatibility import build_report
from app.config import Settings, get_settings
from app.contracts.model import Snapshot
from app.lifespan import DatabaseFactory, create_lifespan, default_database_factory
@@ -34,12 +35,19 @@ def create_app(
app.include_router(health_router)
if resolved.contract_snapshot is not None:
snapshot = Snapshot.model_validate_json(resolved.contract_snapshot.read_bytes())
resolved_handlers = handlers or HandlerRegistry()
register_contract_routes(
app,
snapshot,
handlers or HandlerRegistry(),
resolved_handlers,
resolved.contract_fallback,
)
report = build_report(snapshot, implemented=resolved_handlers.keys())
@app.get("/admin/compatibility", include_in_schema=False)
async def compatibility_report() -> dict[str, object]:
return report.as_json()
return app
+56
View File
@@ -0,0 +1,56 @@
"""Compatibility accounting tests."""
from datetime import UTC, datetime
import pytest
from app.compatibility import build_report
from app.contracts.model import Method, PathContract, Schema, Snapshot
def snapshot() -> Snapshot:
methods = (
Method(
verb="GET",
name="version",
returns=Schema(type="object"),
checksum="1" * 64,
),
Method(
verb="POST",
name="update",
returns=Schema(type="null"),
checksum="2" * 64,
),
)
return Snapshot(
source_version="9.2.3",
retrieved_at=datetime(2026, 1, 1, tzinfo=UTC),
raw_sha256="0" * 64,
paths=(PathContract(path="/nodes/{node}", methods=methods),),
path_count=1,
method_count=2,
)
def test_report_scores_levels_and_groups_independently() -> None:
report = build_report(
snapshot(),
implemented=frozenset({("/nodes/{node}", "GET")}),
observed=frozenset({("/nodes/{node}", "GET"), ("/nodes/{node}", "POST")}),
verified=frozenset({("/nodes/{node}", "GET")}),
)
data = report.as_json()
assert data["total_declared"] == 2
levels = data["levels"]
assert isinstance(levels, dict)
assert levels["implemented"]["score"] == 0.5
assert levels["observed"]["score"] == 1.0
assert data["groups"] == {"nodes": {"declared": 2, "implemented": 1, "verified": 1}}
assert "| implemented | 1 | 50.00% |" in report.as_markdown()
def test_report_rejects_unbound_evidence() -> None:
with pytest.raises(ValueError, match="undeclared"):
build_report(snapshot(), verified=frozenset({("/missing", "GET")}))