From 355bb23d5e431068116ddeedfff19ed20952337d Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Sun, 12 Jul 2026 23:45:55 +0300 Subject: [PATCH] feat: report compatibility evidence --- app/api/registry.py | 3 + app/compatibility.py | 99 ++++++++++++++++++++++++++++++++ app/main.py | 10 +++- tests/unit/test_compatibility.py | 56 ++++++++++++++++++ 4 files changed, 167 insertions(+), 1 deletion(-) create mode 100644 app/compatibility.py create mode 100644 tests/unit/test_compatibility.py diff --git a/app/api/registry.py b/app/api/registry.py index 5dafe6b..5cb0390 100644 --- a/app/api/registry.py +++ b/app/api/registry.py @@ -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, diff --git a/app/compatibility.py b/app/compatibility.py new file mode 100644 index 0000000..b0d24e3 --- /dev/null +++ b/app/compatibility.py @@ -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, + ) diff --git a/app/main.py b/app/main.py index d8f7402..616242b 100644 --- a/app/main.py +++ b/app/main.py @@ -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 diff --git a/tests/unit/test_compatibility.py b/tests/unit/test_compatibility.py new file mode 100644 index 0000000..abf6150 --- /dev/null +++ b/tests/unit/test_compatibility.py @@ -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")}))