inecs
f8d3cbdd59
Add the FastAPI app, PostgreSQL migrations, Docker/Helm packaging, API contracts, docs, client examples, and the unit/integration/compatibility test suite for local client and tooling labs without a real vCenter.
126 lines
4.9 KiB
Python
126 lines
4.9 KiB
Python
"""Catalog-scoped compatibility payload tests (vSphere plane)."""
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import cast
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.compatibility import CompatibilityDimension, build_report
|
|
from app.config import Settings
|
|
from app.contracts.model import Method, PathContract, Schema, Snapshot
|
|
from app.main import create_app
|
|
from app.vsphere.contracts.matrix import VERSIONS
|
|
from app.web.compatibility_catalog import compatibility_payload
|
|
from tests.unit.test_health import FakeDatabase
|
|
|
|
|
|
def _snapshot(source_version: str, path: str) -> Snapshot:
|
|
method = Method(
|
|
verb="GET",
|
|
name="index",
|
|
returns=Schema(type="object"),
|
|
checksum="1" * 64,
|
|
)
|
|
return Snapshot(
|
|
source_version=source_version,
|
|
retrieved_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
raw_sha256="0" * 64,
|
|
paths=(PathContract(path=path, methods=(method,)),),
|
|
path_count=1,
|
|
method_count=1,
|
|
)
|
|
|
|
|
|
def test_catalog_compatibility_uses_selected_snapshot_version() -> None:
|
|
runtime_snapshot = _snapshot("8.0.2", "/api/vcenter/vm")
|
|
catalog_snapshot = _snapshot("7.0.3", "/api/vcenter/host")
|
|
runtime_report = build_report(
|
|
runtime_snapshot,
|
|
implemented=frozenset({("/api/vcenter/vm", "GET")}),
|
|
dimensions={CompatibilityDimension.ROUTE_METHOD: frozenset({("/api/vcenter/vm", "GET")})},
|
|
)
|
|
payload = compatibility_payload(
|
|
catalog_snapshot,
|
|
7,
|
|
implemented_methods=frozenset({("/api/vcenter/host", "GET"), ("/api/vcenter/vm", "GET")}),
|
|
runtime_report=runtime_report,
|
|
runtime_version="8.0.2",
|
|
settings=None,
|
|
)
|
|
assert payload["catalog_version"] == "7.0.3"
|
|
assert payload["runtime_version"] == "8.0.2"
|
|
assert payload["evidence_scope"] == "catalog"
|
|
assert payload["total_declared"] == 1
|
|
levels = cast(dict[str, dict[str, object]], payload["levels"])
|
|
assert levels["implemented"]["count"] == 1
|
|
|
|
|
|
def test_catalog_compatibility_reuses_runtime_report_for_matching_version() -> None:
|
|
runtime_snapshot = _snapshot("8.0.2", "/api/vcenter/vm")
|
|
runtime_report = build_report(
|
|
runtime_snapshot,
|
|
implemented=frozenset({("/api/vcenter/vm", "GET")}),
|
|
dimensions={CompatibilityDimension.ROUTE_METHOD: frozenset({("/api/vcenter/vm", "GET")})},
|
|
)
|
|
payload = compatibility_payload(
|
|
runtime_snapshot,
|
|
9,
|
|
implemented_methods=frozenset({("/api/vcenter/vm", "GET")}),
|
|
runtime_report=runtime_report,
|
|
runtime_version="8.0.2",
|
|
settings=None,
|
|
)
|
|
assert payload["catalog_version"] == "8.0.2"
|
|
assert payload["evidence_scope"] == "full"
|
|
|
|
|
|
async def test_ui_compatibility_endpoint_follows_selected_major() -> None:
|
|
settings = Settings()
|
|
app = create_app(
|
|
settings=settings,
|
|
database_factory=lambda _settings: FakeDatabase(True),
|
|
worker_factories=(),
|
|
)
|
|
async with app.router.lifespan_context(app):
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
major7 = await client.get("/ui/api/compatibility", params={"major": 7})
|
|
major9 = await client.get("/ui/api/compatibility", params={"major": 9})
|
|
assert major7.status_code == 200
|
|
assert major9.status_code == 200
|
|
body7 = major7.json()
|
|
body9 = major9.json()
|
|
assert body7["plane"] == "vsphere-rest"
|
|
assert body9["plane"] == "vsphere-rest"
|
|
assert body7["catalog_version"] == VERSIONS[7]["version"]
|
|
assert body9["catalog_version"] == VERSIONS[9]["version"]
|
|
assert body7["major"] == 7
|
|
assert body9["major"] == 9
|
|
assert body7["total_declared"] > 0
|
|
assert body9["total_declared"] > 0
|
|
# Catalog floor: older majors report a subset; major 9 covers the full registry.
|
|
assert body7["levels"]["implemented"]["count"] < body7["total_declared"]
|
|
assert body9["levels"]["implemented"]["count"] == body9["total_declared"]
|
|
|
|
|
|
async def test_ui_compatibility_covers_all_bundled_majors() -> None:
|
|
settings = Settings()
|
|
app = create_app(
|
|
settings=settings,
|
|
database_factory=lambda _settings: FakeDatabase(True),
|
|
worker_factories=(),
|
|
)
|
|
async with app.router.lifespan_context(app):
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
for major in (6, 7, 8, 9):
|
|
response = await client.get("/ui/api/compatibility", params={"major": major})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["plane"] == "vsphere-rest"
|
|
assert body["catalog_version"] == VERSIONS[major]["version"]
|
|
implemented = body["levels"]["implemented"]["count"]
|
|
declared = body["total_declared"]
|
|
assert implemented > 0
|
|
assert implemented <= declared
|
|
if major == 9:
|
|
assert implemented == declared
|