354 lines
14 KiB
Python
354 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Literal, Protocol
|
|
from uuid import uuid4
|
|
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
from hub_core.contracts import CONTRACT_VERSION, extension_contract_root
|
|
|
|
|
|
class ResponseLike(Protocol):
|
|
status_code: int
|
|
|
|
def json(self) -> Any: ...
|
|
|
|
|
|
class ConformanceTarget(Protocol):
|
|
"""Small HTTP surface shared by httpx.Client and FastAPI TestClient."""
|
|
|
|
def get(self, url: str, **kwargs: Any) -> ResponseLike: ...
|
|
|
|
def post(self, url: str, **kwargs: Any) -> ResponseLike: ...
|
|
|
|
|
|
CheckStatus = Literal["pass", "fail"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckResult:
|
|
check_id: str
|
|
tier: int
|
|
status: CheckStatus
|
|
summary: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConformanceReport:
|
|
contract_version: str
|
|
checks: tuple[CheckResult, ...]
|
|
|
|
@property
|
|
def passed(self) -> bool:
|
|
return all(check.status == "pass" for check in self.checks)
|
|
|
|
@property
|
|
def passed_count(self) -> int:
|
|
return sum(check.status == "pass" for check in self.checks)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"contract_version": self.contract_version,
|
|
"passed": self.passed,
|
|
"summary": {"passed": self.passed_count, "total": len(self.checks)},
|
|
"checks": [asdict(check) for check in self.checks],
|
|
}
|
|
|
|
|
|
class ConformanceHarness:
|
|
"""Run the implemented Tier 2/3 profile against an isolated HTTP target.
|
|
|
|
The target should be disposable or use a dedicated test namespace. The
|
|
harness writes the packaged ops-hub fixture plus one message and one event
|
|
from each implemented framework event family.
|
|
"""
|
|
|
|
def __init__(self, target: ConformanceTarget) -> None:
|
|
self.target = target
|
|
root = extension_contract_root()
|
|
self.package = _load_json(root.joinpath("fixtures", "ops-hub.extension.json"))
|
|
self.scenario = _load_json(root.joinpath("fixtures", "projection-rebuild.json"))
|
|
self.catalog = _load_json(root.joinpath("catalogs", "event-types.json"))
|
|
self.schema_root = root.joinpath("schemas")
|
|
|
|
def run(self) -> ConformanceReport:
|
|
results = [
|
|
self._check("C1", 2, "descriptor and manifest validate", self._schema_validate),
|
|
self._check("C6", 2, "fixtures contain no secret material", self._no_secrets),
|
|
self._check("C3", 2, "health endpoint passes", self._health_probe),
|
|
]
|
|
|
|
correlation_id = str(uuid4())
|
|
registration = self.target.post(
|
|
"/ports/registry/registrations",
|
|
headers={"X-Correlation-ID": correlation_id},
|
|
json=self.package,
|
|
)
|
|
duplicate = self.target.post(
|
|
"/ports/registry/registrations",
|
|
headers={"X-Correlation-ID": correlation_id},
|
|
json=self.package,
|
|
)
|
|
results.append(
|
|
self._check(
|
|
"C4",
|
|
2,
|
|
"manifest activation is idempotent",
|
|
lambda: _assert_registration_idempotent(registration, duplicate),
|
|
)
|
|
)
|
|
results.append(
|
|
self._check(
|
|
"C8",
|
|
2,
|
|
"registry write propagates correlation id",
|
|
lambda: _assert_correlation(registration, correlation_id),
|
|
)
|
|
)
|
|
|
|
progress = _materialize_event(self.scenario["authority"]["progress_events"][0])
|
|
interaction = _materialize_event(self.scenario["authority"]["interaction_events"][0])
|
|
progress_response = self.target.post("/ports/events/progress", json=progress)
|
|
interaction_response = self.target.post("/ports/events/interaction", json=interaction)
|
|
wrong_family = self.target.post("/ports/events/progress", json=interaction)
|
|
unknown_event = {**interaction, "event_type": "hub.interaction.uncataloged"}
|
|
unknown_response = self.target.post("/ports/events/interaction", json=unknown_event)
|
|
results.append(
|
|
self._check(
|
|
"C5",
|
|
2,
|
|
"cataloged events are accepted and invalid types rejected",
|
|
lambda: _assert_event_validation(
|
|
progress_response,
|
|
interaction_response,
|
|
wrong_family,
|
|
unknown_response,
|
|
),
|
|
)
|
|
)
|
|
|
|
message = _materialize_message(self.scenario["authority"]["messages"][0])
|
|
message_response = self.target.post("/ports/messaging/messages", json=message)
|
|
projections = {
|
|
projection_id: self.target.get(f"/ports/projections/{projection_id}")
|
|
for projection_id in self.scenario["expected_projections"]
|
|
}
|
|
results.append(
|
|
self._check(
|
|
"F2",
|
|
3,
|
|
"progress and interaction event families remain separate",
|
|
lambda: _assert_family_separation(
|
|
projections,
|
|
progress["correlation_id"],
|
|
interaction["correlation_id"],
|
|
),
|
|
)
|
|
)
|
|
results.append(
|
|
self._check(
|
|
"F3",
|
|
3,
|
|
"projections rebuild from authority fixture with provenance",
|
|
lambda: _assert_projection_rebuild(
|
|
projections,
|
|
self.scenario["expected_projections"],
|
|
message_response,
|
|
message["correlation_id"],
|
|
),
|
|
)
|
|
)
|
|
return ConformanceReport(contract_version=CONTRACT_VERSION, checks=tuple(results))
|
|
|
|
def _schema_validate(self) -> None:
|
|
_validator(self.schema_root.joinpath("hub-descriptor.schema.json")).validate(
|
|
self.package["descriptor"]
|
|
)
|
|
_validator(self.schema_root.joinpath("hub-manifest.schema.json")).validate(
|
|
self.package["manifest"]
|
|
)
|
|
_validator(self.schema_root.joinpath("event-type-catalog.schema.json")).validate(
|
|
self.catalog
|
|
)
|
|
if self.package["descriptor"]["reuse_surface_id"] != self.package["manifest"][
|
|
"reuse_surface_id"
|
|
]:
|
|
raise AssertionError("descriptor and manifest reuse_surface_id differ")
|
|
|
|
def _no_secrets(self) -> None:
|
|
violations = find_secret_violations(
|
|
{"package": self.package, "scenario": self.scenario, "catalog": self.catalog}
|
|
)
|
|
if violations:
|
|
raise AssertionError("secret-like material: " + ", ".join(violations))
|
|
|
|
def _health_probe(self) -> None:
|
|
response = self.target.get("/healthz")
|
|
_expect_status(response, 200, "health probe")
|
|
if response.json().get("status") != "ok":
|
|
raise AssertionError("health response status is not ok")
|
|
|
|
@staticmethod
|
|
def _check(
|
|
check_id: str,
|
|
tier: int,
|
|
summary: str,
|
|
operation: Any,
|
|
) -> CheckResult:
|
|
try:
|
|
operation()
|
|
except Exception as exc: # each check must produce a complete report
|
|
return CheckResult(check_id, tier, "fail", f"{summary}: {exc}")
|
|
return CheckResult(check_id, tier, "pass", summary)
|
|
|
|
|
|
SECRET_KEY = re.compile(
|
|
r"(?:^|_)(?:api_?(?:key|token)|access_?token|auth_?token|client_?secret|credential|passwd|password|private_?key|secret)(?:$|_)",
|
|
re.IGNORECASE,
|
|
)
|
|
SECRET_VALUE = re.compile(
|
|
r"(?:postgres(?:ql)?|mysql|mariadb|mongodb(?:\+srv)?|redis)://[^\s/:]+:[^\s/@]+@|-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def find_secret_violations(value: Any, path: str = "$") -> list[str]:
|
|
"""Return JSON paths containing credential-shaped keys or values."""
|
|
|
|
violations: list[str] = []
|
|
if isinstance(value, Mapping):
|
|
for key, child in value.items():
|
|
child_path = f"{path}.{key}"
|
|
if SECRET_KEY.search(str(key)):
|
|
violations.append(child_path)
|
|
violations.extend(find_secret_violations(child, child_path))
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
|
|
for index, child in enumerate(value):
|
|
violations.extend(find_secret_violations(child, f"{path}[{index}]"))
|
|
elif isinstance(value, str) and SECRET_VALUE.search(value):
|
|
violations.append(path)
|
|
return violations
|
|
|
|
|
|
def _assert_registration_idempotent(
|
|
registration: ResponseLike,
|
|
duplicate: ResponseLike,
|
|
) -> None:
|
|
_expect_status(registration, 202, "initial activation")
|
|
_expect_status(duplicate, 202, "duplicate activation")
|
|
if registration.json().get("status") not in {"accepted", "duplicate"}:
|
|
raise AssertionError("initial activation did not return an activation status")
|
|
if duplicate.json().get("status") != "duplicate":
|
|
raise AssertionError("second activation was not reported as duplicate")
|
|
|
|
|
|
def _assert_correlation(response: ResponseLike, expected: str) -> None:
|
|
_expect_status(response, 202, "correlated registry write")
|
|
if response.json().get("correlation_id") != expected:
|
|
raise AssertionError("response correlation_id does not match request")
|
|
|
|
|
|
def _assert_event_validation(
|
|
progress: ResponseLike,
|
|
interaction: ResponseLike,
|
|
wrong_family: ResponseLike,
|
|
unknown: ResponseLike,
|
|
) -> None:
|
|
_expect_status(progress, 202, "progress event")
|
|
_expect_status(interaction, 202, "interaction event")
|
|
_expect_status(wrong_family, 422, "wrong-family event")
|
|
_expect_status(unknown, 422, "uncataloged event")
|
|
|
|
|
|
def _assert_family_separation(
|
|
projections: Mapping[str, ResponseLike],
|
|
progress_correlation: str,
|
|
interaction_correlation: str,
|
|
) -> None:
|
|
progress_items = _projection_items(projections["progress_events"], "progress_events")
|
|
interaction_items = _projection_items(
|
|
projections["interaction_events"], "interaction_events"
|
|
)
|
|
progress_correlations = {item.get("correlation_id") for item in progress_items}
|
|
interaction_correlations = {item.get("correlation_id") for item in interaction_items}
|
|
if progress_correlation not in progress_correlations:
|
|
raise AssertionError("progress fixture missing from progress projection")
|
|
if interaction_correlation not in interaction_correlations:
|
|
raise AssertionError("interaction fixture missing from interaction projection")
|
|
if progress_correlation in interaction_correlations or interaction_correlation in progress_correlations:
|
|
raise AssertionError("event correlation crossed family projection boundary")
|
|
if any(item.get("family") != "progress" for item in progress_items):
|
|
raise AssertionError("progress projection contains another event family")
|
|
if any(item.get("family") != "interaction" for item in interaction_items):
|
|
raise AssertionError("interaction projection contains another event family")
|
|
|
|
|
|
def _assert_projection_rebuild(
|
|
projections: Mapping[str, ResponseLike],
|
|
expectations: Mapping[str, Any],
|
|
message_response: ResponseLike,
|
|
message_correlation: str,
|
|
) -> None:
|
|
_expect_status(message_response, 202, "fixture message")
|
|
for projection_id, expected in expectations.items():
|
|
response = projections[projection_id]
|
|
_expect_status(response, 200, f"{projection_id} projection")
|
|
body = response.json()
|
|
data = body.get("data", {})
|
|
if data.get("projection_id") != projection_id:
|
|
raise AssertionError(f"{projection_id} identity was not preserved")
|
|
if data.get("rebuild_from") != expected["rebuild_from"]:
|
|
raise AssertionError(f"{projection_id} rebuild sources differ")
|
|
provenance = body.get("provenance", {})
|
|
if not provenance.get("source_ref") or not provenance.get("schema_version"):
|
|
raise AssertionError(f"{projection_id} lacks provenance")
|
|
if not provenance.get("content_hash"):
|
|
raise AssertionError(f"{projection_id} lacks content hash")
|
|
|
|
registry_items = _projection_items(projections["hub_registry"], "hub_registry")
|
|
if not any(item.get("descriptor", {}).get("hub_slug") == "ops-hub" for item in registry_items):
|
|
raise AssertionError("ops-hub authority fixture missing from registry projection")
|
|
message_items = _projection_items(projections["messages"], "messages")
|
|
if not any(item.get("correlation_id") == message_correlation for item in message_items):
|
|
raise AssertionError("message authority fixture missing from message projection")
|
|
|
|
|
|
def _projection_items(response: ResponseLike, projection_id: str) -> list[dict[str, Any]]:
|
|
_expect_status(response, 200, f"{projection_id} projection")
|
|
items = response.json().get("data", {}).get("items")
|
|
if not isinstance(items, list):
|
|
raise AssertionError(f"{projection_id} items are not a list")
|
|
return items
|
|
|
|
|
|
def _materialize_event(template: Mapping[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
**template,
|
|
"correlation_id": str(uuid4()),
|
|
"occurred_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def _materialize_message(template: Mapping[str, Any]) -> dict[str, Any]:
|
|
return {**template, "correlation_id": str(uuid4()), "conversation_id": str(uuid4())}
|
|
|
|
|
|
def _expect_status(response: ResponseLike, expected: int, purpose: str) -> None:
|
|
if response.status_code != expected:
|
|
detail = json.dumps(response.json(), sort_keys=True)
|
|
raise AssertionError(f"{purpose} returned {response.status_code}, expected {expected}: {detail}")
|
|
|
|
|
|
def _validator(resource: Any) -> Draft202012Validator:
|
|
schema = _load_json(resource)
|
|
Draft202012Validator.check_schema(schema)
|
|
return Draft202012Validator(schema, format_checker=FormatChecker())
|
|
|
|
|
|
def _load_json(resource: Any) -> Any:
|
|
return json.loads(resource.read_text(encoding="utf-8"))
|