diff --git a/app/contracts/model.py b/app/contracts/model.py new file mode 100644 index 0000000..ee6ba24 --- /dev/null +++ b/app/contracts/model.py @@ -0,0 +1,134 @@ +"""Immutable normalized representation of Proxmox API contracts.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from datetime import datetime +from typing import Any, Self + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +type JsonValue = None | bool | int | float | str | list[JsonValue] | dict[str, JsonValue] + + +def canonical_json(value: BaseModel | Mapping[str, Any] | Sequence[Any]) -> bytes: + """Serialize a JSON-compatible value deterministically as UTF-8.""" + + data: Any + if isinstance(value, BaseModel): + data = value.model_dump(mode="json", exclude_none=True) + else: + data = value + return json.dumps( + data, + default=_json_default, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode() + + +def _json_default(value: object) -> Any: + if isinstance(value, BaseModel): + return value.model_dump(mode="json", exclude_none=True) + if isinstance(value, datetime): + return value.isoformat() + raise TypeError(f"{type(value).__name__} is not JSON serializable") + + +def sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +class FrozenModel(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + +class Schema(FrozenModel): + """Proxmox's JSON-Schema-like dialect with retained extensions.""" + + type: str | None = None + description: str | None = None + properties: dict[str, Schema] = Field(default_factory=dict) + items: Schema | None = None + enum: tuple[JsonValue, ...] = () + optional: bool | None = None + default: JsonValue = None + minimum: int | float | None = None + maximum: int | float | None = None + min_length: int | None = None + max_length: int | None = None + pattern: str | None = None + format: str | dict[str, JsonValue] | None = None + extra: dict[str, JsonValue] = Field(default_factory=dict) + + +class Parameter(FrozenModel): + name: str + definition: Schema + + +class Permissions(FrozenModel): + user: str | None = None + description: str | None = None + expression: dict[str, JsonValue] = Field(default_factory=dict) + extra: dict[str, JsonValue] = Field(default_factory=dict) + + +class Method(FrozenModel): + verb: str + name: str + description: str | None = None + parameters: tuple[Parameter, ...] = () + returns: Schema = Field(default_factory=Schema) + permissions: Permissions | None = None + protected: bool = False + allow_token: bool | None = None + extra: dict[str, JsonValue] = Field(default_factory=dict) + checksum: str + + +class PathContract(FrozenModel): + path: str + methods: tuple[Method, ...] + extra: dict[str, JsonValue] = Field(default_factory=dict) + + +class Snapshot(FrozenModel): + format_version: int = 1 + source_version: str + retrieved_at: datetime + raw_sha256: str + paths: tuple[PathContract, ...] + path_count: int + method_count: int + extra: dict[str, JsonValue] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_counts_and_uniqueness(self) -> Self: + if self.path_count != len(self.paths): + raise ValueError("path_count does not match paths") + methods = sum(len(path.methods) for path in self.paths) + if self.method_count != methods: + raise ValueError("method_count does not match methods") + keys = [(path.path, method.verb) for path in self.paths for method in path.methods] + if len(keys) != len(set(keys)): + raise ValueError("duplicate path and method") + return self + + def canonical_bytes(self) -> bytes: + return canonical_json(self) + + def checksum(self) -> str: + return sha256(self.canonical_bytes()) + + +class Manifest(FrozenModel): + source_version: str + raw_sha256: str + snapshot_sha256: str + path_count: int + method_count: int diff --git a/app/contracts/normalize.py b/app/contracts/normalize.py new file mode 100644 index 0000000..9bed0ee --- /dev/null +++ b/app/contracts/normalize.py @@ -0,0 +1,168 @@ +"""Normalize parsed API Viewer trees into stable contract models.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime +from typing import Any, cast + +from app.contracts.model import ( + JsonValue, + Manifest, + Method, + Parameter, + PathContract, + Permissions, + Schema, + Snapshot, + canonical_json, + sha256, +) +from app.contracts.source import ParsedSource + +SCHEMA_FIELDS = { + "type", + "description", + "properties", + "items", + "enum", + "optional", + "default", + "minimum", + "maximum", + "minLength", + "maxLength", + "pattern", + "format", +} +METHOD_FIELDS = { + "allowtoken", + "description", + "method", + "name", + "parameters", + "permissions", + "protected", + "returns", +} + + +def _json(value: Any) -> JsonValue: + return cast(JsonValue, value) + + +def normalize_schema(raw: Mapping[str, Any] | None) -> Schema: + source = raw or {} + properties = source.get("properties", {}) + normalized_properties = { + str(name): normalize_schema(cast(Mapping[str, Any], schema)) + for name, schema in cast(Mapping[str, Any], properties).items() + } + items = source.get("items") + extra = {key: _json(value) for key, value in source.items() if key not in SCHEMA_FIELDS} + return Schema( + type=source.get("type"), + description=source.get("description"), + properties=normalized_properties, + items=normalize_schema(cast(Mapping[str, Any], items)) + if isinstance(items, Mapping) + else None, + enum=tuple(_json(value) for value in source.get("enum", ())), + optional=bool(source["optional"]) if "optional" in source else None, + default=_json(source.get("default")), + minimum=source.get("minimum"), + maximum=source.get("maximum"), + min_length=source.get("minLength"), + max_length=source.get("maxLength"), + pattern=source.get("pattern"), + format=_json(source.get("format")), + extra=extra, + ) + + +def normalize_permissions(raw: Mapping[str, Any] | None) -> Permissions | None: + if raw is None: + return None + known = {"user", "description"} + expression_keys = {"and", "or", "check", "userParam"} + return Permissions( + user=raw.get("user"), + description=raw.get("description"), + expression={key: _json(raw[key]) for key in expression_keys if key in raw}, + extra={ + key: _json(value) for key, value in raw.items() if key not in known | expression_keys + }, + ) + + +def normalize_method(verb: str, raw: Mapping[str, Any]) -> Method: + parameters_raw = cast(Mapping[str, Any], raw.get("parameters", {})).get("properties", {}) + parameters = tuple( + Parameter(name=str(name), definition=normalize_schema(cast(Mapping[str, Any], schema))) + for name, schema in sorted(cast(Mapping[str, Any], parameters_raw).items()) + ) + values: dict[str, Any] = { + "verb": verb.upper(), + "name": str(raw.get("name", verb.lower())), + "description": raw.get("description"), + "parameters": parameters, + "returns": normalize_schema(cast(Mapping[str, Any] | None, raw.get("returns"))), + "permissions": normalize_permissions( + cast(Mapping[str, Any] | None, raw.get("permissions")) + ), + "protected": bool(raw.get("protected", False)), + "allow_token": bool(raw["allowtoken"]) if "allowtoken" in raw else None, + "extra": {key: _json(value) for key, value in raw.items() if key not in METHOD_FIELDS}, + } + checksum = sha256(canonical_json(values)) + return Method(**values, checksum=checksum) + + +def _walk(nodes: tuple[dict[str, Any], ...]) -> list[PathContract]: + paths: list[PathContract] = [] + + def visit(node: Mapping[str, Any]) -> None: + info = node.get("info") + path = node.get("path") + if isinstance(info, Mapping) and isinstance(path, str): + methods = tuple( + normalize_method(str(verb), cast(Mapping[str, Any], method)) + for verb, method in sorted(info.items()) + if isinstance(method, Mapping) + ) + extra = { + key: _json(value) + for key, value in node.items() + if key not in {"children", "info", "leaf", "path", "text"} + } + paths.append(PathContract(path=path, methods=methods, extra=extra)) + for child in node.get("children", ()): + if isinstance(child, Mapping): + visit(child) + + for root in nodes: + visit(root) + return sorted(paths, key=lambda item: item.path) + + +def normalize_snapshot( + parsed: ParsedSource, *, raw: bytes, source_version: str, retrieved_at: datetime +) -> tuple[Snapshot, Manifest]: + paths = tuple(_walk(parsed.nodes)) + snapshot = Snapshot( + source_version=source_version, + retrieved_at=retrieved_at, + raw_sha256=sha256(raw), + paths=paths, + path_count=len(paths), + method_count=sum(len(path.methods) for path in paths), + extra={"warning_count": len(parsed.warnings)}, + ) + manifest = Manifest( + source_version=source_version, + raw_sha256=snapshot.raw_sha256, + snapshot_sha256=snapshot.checksum(), + path_count=snapshot.path_count, + method_count=snapshot.method_count, + ) + return snapshot, manifest diff --git a/pyproject.toml b/pyproject.toml index c5fffb3..b8b5887 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ dependencies = [ [project.optional-dependencies] dev = [ + "hypothesis>=6.135,<7", "httpx>=0.28,<0.29", "mypy>=1.17,<1.18", "pytest>=8.4,<9", @@ -62,4 +63,3 @@ source = ["app"] [tool.coverage.report] fail_under = 80 show_missing = true - diff --git a/tests/unit/test_contract_model.py b/tests/unit/test_contract_model.py new file mode 100644 index 0000000..64c2969 --- /dev/null +++ b/tests/unit/test_contract_model.py @@ -0,0 +1,75 @@ +"""Determinism and validation checks for normalized contracts.""" + +import json +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from hypothesis import given +from hypothesis import strategies as st +from pydantic import ValidationError + +from app.contracts.model import Snapshot, canonical_json +from app.contracts.normalize import normalize_snapshot +from app.contracts.source import ApiViewerParser + +FIXTURE = Path(__file__).parents[1] / "fixtures" / "api-viewer" / "pve-9.2.3-version.json" +RETRIEVED_AT = datetime(2026, 7, 12, 20, 8, 59, tzinfo=UTC) + + +def make_snapshot() -> Snapshot: + raw = FIXTURE.read_bytes() + parsed = ApiViewerParser().parse(raw) + snapshot, _ = normalize_snapshot( + parsed, raw=raw, source_version="9.2.3", retrieved_at=RETRIEVED_AT + ) + return snapshot + + +def test_normalization_is_deterministic_and_round_trips() -> None: + first = make_snapshot() + second = make_snapshot() + + assert first.canonical_bytes() == second.canonical_bytes() + assert first.checksum() == second.checksum() + assert Snapshot.model_validate_json(first.canonical_bytes()) == first + assert first.paths[0].methods[0].checksum == second.paths[0].methods[0].checksum + + +def test_snapshot_validates_declared_counts() -> None: + data = make_snapshot().model_dump(mode="json") + data["method_count"] = 99 + + with pytest.raises(ValidationError, match="method_count"): + Snapshot.model_validate(data) + + +def test_unknown_schema_fields_are_retained() -> None: + raw = json.dumps( + [ + { + "path": "/future", + "info": { + "GET": { + "name": "future", + "returns": {"type": "string", "futureKeyword": {"x": 1}}, + } + }, + } + ] + ).encode() + snapshot, _ = normalize_snapshot( + ApiViewerParser().parse(raw), + raw=raw, + source_version="test", + retrieved_at=RETRIEVED_AT, + ) + + assert snapshot.paths[0].methods[0].returns.extra["futureKeyword"] == {"x": 1} + + +@given(st.dictionaries(st.text(min_size=1), st.integers(), max_size=10)) +def test_canonical_json_is_independent_of_mapping_order(values: dict[str, int]) -> None: + reversed_values = dict(reversed(tuple(values.items()))) + + assert canonical_json(values) == canonical_json(reversed_values)