152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
import json
|
|
from collections.abc import Mapping, Sequence
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
from hub_core.contracts import CONTRACT_ID, CONTRACT_VERSION, extension_contract_root
|
|
|
|
|
|
ROOT = extension_contract_root()
|
|
SCHEMAS = ROOT.joinpath("schemas")
|
|
FIXTURE = ROOT.joinpath("fixtures", "ops-hub.extension.json")
|
|
REBUILD_FIXTURE = ROOT.joinpath("fixtures", "projection-rebuild.json")
|
|
CATALOG = ROOT.joinpath("catalogs", "event-types.json")
|
|
OPENAPI = ROOT.joinpath("openapi", "ports.openapi.json")
|
|
COMPATIBILITY = ROOT.joinpath("compatibility-matrix.json")
|
|
|
|
PORT_IDS = {
|
|
"port.registry",
|
|
"port.addressing",
|
|
"port.messaging",
|
|
"port.events.progress",
|
|
"port.events.interaction",
|
|
"port.projection.query",
|
|
"port.repo",
|
|
"port.work",
|
|
"port.policy",
|
|
"port.telemetry",
|
|
"port.schedule",
|
|
}
|
|
|
|
SECRET_KEYS = {
|
|
"api_key",
|
|
"credential",
|
|
"password",
|
|
"passwd",
|
|
"private_key",
|
|
"secret",
|
|
"token",
|
|
}
|
|
|
|
|
|
def load_json(resource: Any) -> Any:
|
|
return json.loads(resource.read_text(encoding="utf-8"))
|
|
|
|
|
|
def validate(instance: Any, schema_name: str) -> None:
|
|
schema = load_json(SCHEMAS.joinpath(schema_name))
|
|
Draft202012Validator.check_schema(schema)
|
|
Draft202012Validator(schema, format_checker=FormatChecker()).validate(instance)
|
|
|
|
|
|
def iter_keys(value: Any) -> set[str]:
|
|
if isinstance(value, Mapping):
|
|
keys = {str(key).lower() for key in value}
|
|
for child in value.values():
|
|
keys.update(iter_keys(child))
|
|
return keys
|
|
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
|
|
keys: set[str] = set()
|
|
for child in value:
|
|
keys.update(iter_keys(child))
|
|
return keys
|
|
return set()
|
|
|
|
|
|
def test_packaged_contract_identity_and_artifacts() -> None:
|
|
assert CONTRACT_ID == "helixforge.hub-extension"
|
|
assert CONTRACT_VERSION == "0.1.0"
|
|
assert ROOT.joinpath("README.md").is_file()
|
|
assert FIXTURE.is_file()
|
|
assert REBUILD_FIXTURE.is_file()
|
|
assert CATALOG.is_file()
|
|
assert OPENAPI.is_file()
|
|
assert COMPATIBILITY.is_file()
|
|
|
|
|
|
def test_ops_hub_fixture_validates_against_descriptor_and_manifest_schemas() -> None:
|
|
package = load_json(FIXTURE)
|
|
|
|
validate(package["descriptor"], "hub-descriptor.schema.json")
|
|
validate(package["manifest"], "hub-manifest.schema.json")
|
|
|
|
assert package["descriptor"]["reuse_surface_id"] == package["manifest"]["reuse_surface_id"]
|
|
|
|
|
|
def test_event_catalog_validates_and_keeps_event_families_distinct() -> None:
|
|
catalog = load_json(CATALOG)
|
|
validate(catalog, "event-type-catalog.schema.json")
|
|
|
|
event_types = catalog["event_types"]
|
|
names = [entry["type"] for entry in event_types]
|
|
families = {entry["family"] for entry in event_types}
|
|
assert len(names) == len(set(names))
|
|
assert {"progress", "interaction"} <= families
|
|
|
|
|
|
def test_fixture_events_resolve_in_catalog() -> None:
|
|
package = load_json(FIXTURE)
|
|
catalog = load_json(CATALOG)
|
|
known = {entry["type"] for entry in catalog["event_types"]}
|
|
declared = set(package["manifest"]["events_emitted"])
|
|
declared.update(package["manifest"]["events_consumed"])
|
|
assert declared <= known
|
|
|
|
|
|
def test_contract_examples_contain_no_secret_fields() -> None:
|
|
assert not (iter_keys(load_json(FIXTURE)) & SECRET_KEYS)
|
|
assert not (iter_keys(load_json(REBUILD_FIXTURE)) & SECRET_KEYS)
|
|
assert not (iter_keys(load_json(CATALOG)) & SECRET_KEYS)
|
|
|
|
|
|
def test_openapi_declares_every_named_port() -> None:
|
|
document = load_json(OPENAPI)
|
|
assert document["openapi"] == "3.1.0"
|
|
assert document["info"]["version"] == CONTRACT_VERSION
|
|
|
|
operations = [
|
|
operation
|
|
for path_item in document["paths"].values()
|
|
for method, operation in path_item.items()
|
|
if method in {"get", "post", "put", "patch", "delete"}
|
|
]
|
|
assert {operation["x-port-id"] for operation in operations} == PORT_IDS
|
|
assert all(operation["x-direction"] in {"in", "out"} for operation in operations)
|
|
assert all(operation.get("operationId") for operation in operations)
|
|
assert all(operation.get("responses") for operation in operations)
|
|
|
|
|
|
def test_manifest_port_enum_matches_openapi_ports() -> None:
|
|
manifest_schema = load_json(SCHEMAS.joinpath("hub-manifest.schema.json"))
|
|
declared = set(manifest_schema["$defs"]["portId"]["enum"])
|
|
openapi = load_json(OPENAPI)
|
|
implemented = {
|
|
operation["x-port-id"]
|
|
for path_item in openapi["paths"].values()
|
|
for method, operation in path_item.items()
|
|
if method in {"get", "post", "put", "patch", "delete"}
|
|
}
|
|
assert declared == implemented == PORT_IDS
|
|
|
|
|
|
def test_compatibility_matrix_covers_current_and_core_hub_adapter() -> None:
|
|
matrix = load_json(COMPATIBILITY)
|
|
assert matrix["contract_id"] == CONTRACT_ID
|
|
assert matrix["current_version"] == CONTRACT_VERSION
|
|
assert any(entry["version"] == CONTRACT_VERSION for entry in matrix["versions"])
|
|
assert any(
|
|
adapter["source"] == "core-hub.hub-manifest"
|
|
and adapter["target_version"] == CONTRACT_VERSION
|
|
for adapter in matrix["migration_adapters"]
|
|
)
|