tegwick
ab936a1e98
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0230c-b06c-7641-808a-e191b6d1da49
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
from hub_core.contracts import (
|
|
WORKLOAD_PROJECTION_CONTRACT_ID,
|
|
WORKLOAD_PROJECTION_CONTRACT_VERSION,
|
|
workload_projection_contract_root,
|
|
)
|
|
|
|
|
|
ROOT = workload_projection_contract_root()
|
|
SCHEMAS = ROOT.joinpath("schemas")
|
|
FIXTURE = ROOT.joinpath("fixtures", "repo-manager-nine-workloads.json")
|
|
OPENAPI = ROOT.joinpath("openapi", "workload-projection.openapi.json")
|
|
COMPATIBILITY = ROOT.joinpath("compatibility-matrix.json")
|
|
|
|
|
|
def load(resource: Any) -> Any:
|
|
return json.loads(resource.read_text(encoding="utf-8"))
|
|
|
|
|
|
def validate(instance: Any, name: str) -> None:
|
|
schema = load(SCHEMAS.joinpath(name))
|
|
Draft202012Validator.check_schema(schema)
|
|
Draft202012Validator(schema, format_checker=FormatChecker()).validate(instance)
|
|
|
|
|
|
def test_packaged_workload_contract_identity_and_artifacts() -> None:
|
|
assert WORKLOAD_PROJECTION_CONTRACT_ID == "helixforge.workload-projection"
|
|
assert WORKLOAD_PROJECTION_CONTRACT_VERSION == "1.0.0"
|
|
for artifact in (ROOT.joinpath("README.md"), FIXTURE, OPENAPI, COMPATIBILITY):
|
|
assert artifact.is_file()
|
|
for name in (
|
|
"workload-projection-page.schema.json",
|
|
"workload-projection.schema.json",
|
|
"workload-resolution.schema.json",
|
|
):
|
|
Draft202012Validator.check_schema(load(SCHEMAS.joinpath(name)))
|
|
|
|
|
|
def test_nine_declaration_fixture_is_valid_exact_and_canonical_utc() -> None:
|
|
fixture = load(FIXTURE)
|
|
validate(fixture, "workload-projection-page.schema.json")
|
|
workloads = fixture["workloads"]
|
|
identities = [(record["rapp_id"], record["name"]) for record in workloads]
|
|
|
|
assert len(workloads) == fixture["snapshot"]["total_workload_count"] == 9
|
|
assert identities == sorted(identities)
|
|
assert len(identities) == len(set(identities))
|
|
assert all(record["declaration_repo"] == record["rapp_id"] for record in workloads)
|
|
assert all(
|
|
record["declaration_path"]
|
|
== f"{record['declaration_repo']}/declarations/rapp.yaml"
|
|
for record in workloads
|
|
)
|
|
assert fixture["snapshot"]["generated_at"].endswith("Z")
|
|
assert all(record["observed_at"].endswith("Z") for record in workloads)
|
|
|
|
|
|
def test_compatibility_matrix_preserves_repo_manager_authority_and_forbids_inference() -> None:
|
|
matrix = load(COMPATIBILITY)
|
|
|
|
assert matrix["contract_id"] == WORKLOAD_PROJECTION_CONTRACT_ID
|
|
assert matrix["accepted_inputs"] == [
|
|
{
|
|
"contract_id": "helixforge.workload-projection-source",
|
|
"version": "1.0.0",
|
|
"source_contract": "helixforge.workload-reference/v1",
|
|
"status": "current",
|
|
"coercion": False,
|
|
}
|
|
]
|
|
assert matrix["authority"]["index_and_resolution"] == "repo-manager"
|
|
assert set(matrix["forbidden_fallbacks"]) == {
|
|
"repository_name",
|
|
"owner_repo",
|
|
"path_segment",
|
|
"alias",
|
|
}
|
|
|
|
|
|
def test_workload_openapi_is_projection_query_read_only() -> None:
|
|
document = load(OPENAPI)
|
|
operations = [
|
|
(method, operation)
|
|
for path_item in document["paths"].values()
|
|
for method, operation in path_item.items()
|
|
if method in {"get", "post", "put", "patch", "delete"}
|
|
]
|
|
|
|
assert document["openapi"] == "3.1.0"
|
|
assert operations and all(method == "get" for method, _ in operations)
|
|
assert all(operation["x-port-id"] == "port.projection.query" for _, operation in operations)
|