167 lines
6.0 KiB
Python
167 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import jsonschema
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
CONTRACT = ROOT / "contracts" / "repository" / "v1"
|
|
FIXTURES = CONTRACT / "fixtures"
|
|
|
|
|
|
def load(path: Path) -> Any:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
raise AssertionError(f"invalid contract JSON: {path}") from error
|
|
|
|
|
|
def validator(name: str) -> jsonschema.Draft202012Validator:
|
|
schema = load(CONTRACT / f"{name}.schema.json")
|
|
jsonschema.Draft202012Validator.check_schema(schema)
|
|
return jsonschema.Draft202012Validator(
|
|
schema,
|
|
format_checker=jsonschema.FormatChecker(),
|
|
)
|
|
|
|
|
|
def test_repository_and_manifest_golden_files_validate() -> None:
|
|
for name in ("repository", "manifest"):
|
|
validator(name).validate(load(FIXTURES / f"valid-{name}.json"))
|
|
|
|
|
|
def test_invalid_golden_files_are_rejected() -> None:
|
|
for name in ("repository", "manifest"):
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator(name).validate(load(FIXTURES / f"invalid-{name}.json"))
|
|
|
|
|
|
def test_valid_golden_files_are_canonical_json() -> None:
|
|
for path in sorted(FIXTURES.glob("valid-*.json")):
|
|
data = load(path)
|
|
canonical = json.dumps(data, sort_keys=True, separators=(",", ":")) + "\n"
|
|
assert path.read_text(encoding="utf-8") == canonical
|
|
|
|
|
|
def test_protocol_contracts_are_versioned_and_closed() -> None:
|
|
repository_schema = load(CONTRACT / "repository.schema.json")
|
|
manifest_schema = load(CONTRACT / "manifest.schema.json")
|
|
|
|
assert repository_schema["properties"]["format_version"]["const"] == 1
|
|
assert manifest_schema["properties"]["format_version"]["const"] == 1
|
|
assert not repository_schema["additionalProperties"]
|
|
assert not manifest_schema["additionalProperties"]
|
|
|
|
|
|
def test_uuidv7_and_rfc3339_formats_are_enforced() -> None:
|
|
repository = load(FIXTURES / "valid-repository.json")
|
|
manifest = load(FIXTURES / "valid-manifest.json")
|
|
|
|
repository["repository_id"] = "123e4567-e89b-42d3-a456-426614174000"
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("repository").validate(repository)
|
|
|
|
manifest["created_at"] = "2026-07-27 12:00:00"
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("manifest").validate(manifest)
|
|
|
|
|
|
def test_manifest_signature_and_source_consistency_are_required() -> None:
|
|
manifest = load(FIXTURES / "valid-manifest.json")
|
|
assert manifest["manifest_signature"]["algorithm"] == "ed25519"
|
|
|
|
missing_signature = deepcopy(manifest)
|
|
del missing_signature["manifest_signature"]
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("manifest").validate(missing_signature)
|
|
|
|
incomplete_consistency = deepcopy(manifest)
|
|
incomplete_consistency["source_consistency"] = {"adapter": "local"}
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("manifest").validate(incomplete_consistency)
|
|
|
|
|
|
def test_entry_types_bind_blobs_and_safe_link_targets() -> None:
|
|
manifest = load(FIXTURES / "valid-manifest.json")
|
|
entries = manifest["entries"]
|
|
|
|
file_without_blob = deepcopy(manifest)
|
|
file_without_blob["entries"][1]["blob_digest"] = None
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("manifest").validate(file_without_blob)
|
|
|
|
directory_with_blob = deepcopy(manifest)
|
|
directory_with_blob["entries"][0]["blob_digest"] = "c" * 64
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("manifest").validate(directory_with_blob)
|
|
|
|
for unsafe_target in ("/etc/passwd", "../escape", "dir\\escape"):
|
|
unsafe_symlink = deepcopy(manifest)
|
|
unsafe_symlink["entries"] = [
|
|
{
|
|
"blob_digest": None,
|
|
"link_target": unsafe_target,
|
|
"metadata_support": [],
|
|
"mode": None,
|
|
"mtime_ns": None,
|
|
"path": "link",
|
|
"size": 0,
|
|
"type": "symlink",
|
|
}
|
|
]
|
|
with pytest.raises(jsonschema.ValidationError):
|
|
validator("manifest").validate(unsafe_symlink)
|
|
|
|
assert entries[1]["type"] == "file"
|
|
|
|
|
|
def test_normalized_path_vectors_are_relative_posix_paths() -> None:
|
|
vectors = load(CONTRACT / "normalized-paths.json")
|
|
valid = [case for case in vectors if case["valid"]]
|
|
invalid = [case for case in vectors if not case["valid"]]
|
|
|
|
assert {case["raw"] for case in invalid} >= {"/absolute", "../escape", "a\\b"}
|
|
for case in valid:
|
|
normalized = case["normalized"]
|
|
assert normalized and not normalized.startswith("/")
|
|
assert "\\" not in normalized
|
|
assert ".." not in normalized.split("/")
|
|
|
|
|
|
def test_state_errors_capabilities_and_fault_points_are_frozen() -> None:
|
|
transitions = load(CONTRACT / "execution-transitions.json")
|
|
assert transitions["queued"] == ["cancelled", "preparing"]
|
|
assert transitions["verifying"] == ["committed", "failed"]
|
|
assert transitions["committed"] == []
|
|
|
|
errors = load(CONTRACT / "error-codes.json")
|
|
assert len(errors) == len(set(errors))
|
|
assert all(code == code.lower() and " " not in code for code in errors)
|
|
|
|
capabilities = load(CONTRACT / "capabilities-v2.0.json")
|
|
assert capabilities["sources"] == ["local", "ssh"]
|
|
|
|
manifest_schema = load(CONTRACT / "manifest.schema.json")
|
|
adapter_kinds = manifest_schema["properties"]["source_consistency"]["properties"]["adapter"][
|
|
"enum"
|
|
]
|
|
assert adapter_kinds == ["local", "postgresql", "mysql"]
|
|
assert "ssh" not in adapter_kinds
|
|
|
|
assert not capabilities["features"]["tar_download"]
|
|
assert not capabilities["features"]["postgresql"]
|
|
assert not capabilities["features"]["mysql"]
|
|
|
|
fault_points = load(CONTRACT / "fault-points.json")
|
|
assert {
|
|
"blob.before_write",
|
|
"blob.after_fsync",
|
|
"manifest.before_publish",
|
|
"metadata.before_commit",
|
|
"restore.before_replace",
|
|
}.issubset(fault_points)
|