from __future__ import annotations import json from datetime import datetime, timezone from uuid import uuid4 from fastapi.testclient import TestClient from hub_core.contracts import extension_contract_root from hub_core.runtime.app import create_app from hub_core.runtime.cli import _sync_database_url, build_parser from hub_core.runtime.config import RuntimeSettings from hub_core.runtime.store import InMemoryPortStore def client(*, allow_ephemeral: bool = True, environment: str = "test") -> TestClient: settings = RuntimeSettings( environment=environment, backend="memory", allow_ephemeral=allow_ephemeral, ) return TestClient(create_app(settings=settings, port_store=InMemoryPortStore())) def ops_hub_package() -> dict: fixture = extension_contract_root().joinpath("fixtures", "ops-hub.extension.json") return json.loads(fixture.read_text(encoding="utf-8")) def event_body(event_type: str) -> dict: return { "schema_version": "0.1.0", "correlation_id": str(uuid4()), "event_type": event_type, "occurred_at": datetime.now(timezone.utc).isoformat(), "subject_refs": {"hub": "ops-hub"}, "payload": {"result": "ok"}, } def test_health_and_ephemeral_readiness() -> None: runtime = client() health = runtime.get("/healthz") ready = runtime.get("/readyz") assert health.status_code == 200 assert health.json()["service"] == "hub-core" assert ready.status_code == 200 assert ready.json()["status"] == "ok" assert ready.json()["checks"]["active_backend"] == "memory" def test_production_readiness_fails_closed_for_ephemeral_backend() -> None: response = client(allow_ephemeral=False, environment="production").get("/readyz") assert response.status_code == 503 assert response.json()["status"] == "degraded" assert response.json()["checks"]["ephemeral_backend"] == "not_allowed" def test_protected_compatibility_readiness_requires_authorization_dependency() -> None: settings = RuntimeSettings( environment="test", backend="memory", allow_ephemeral=True, v2_groups=frozenset({"registry"}), ) response = TestClient( create_app(settings=settings, port_store=InMemoryPortStore()) ).get("/readyz") assert response.status_code == 503 assert response.json()["checks"]["authorization"] == "unavailable" def test_registry_validates_and_registers_idempotently() -> None: runtime = client() package = ops_hub_package() correlation_id = str(uuid4()) first = runtime.post( "/ports/registry/registrations", headers={"X-Correlation-ID": correlation_id}, json=package, ) second = runtime.post( "/ports/registry/registrations", headers={"X-Correlation-ID": correlation_id}, json=package, ) projection = runtime.get("/ports/projections/hub_registry") assert first.status_code == 202 assert first.json()["status"] == "accepted" assert second.status_code == 202 assert second.json()["status"] == "duplicate" assert projection.status_code == 200 assert projection.json()["data"]["items"][0]["descriptor"]["hub_slug"] == "ops-hub" def test_registry_rejects_contract_mismatch() -> None: runtime = client() package = ops_hub_package() package["manifest"]["reuse_surface_id"] = "capability.operations.other-hub" response = runtime.post( "/ports/registry/registrations", headers={"X-Correlation-ID": str(uuid4())}, json=package, ) assert response.status_code == 422 assert "reuse_surface_id must match" in response.json()["detail"] def test_messaging_port_writes_and_reads_conversation() -> None: runtime = client() conversation_id = uuid4() body = { "schema_version": "0.1.0", "correlation_id": str(uuid4()), "conversation_id": str(conversation_id), "from_address": "agent:codex", "to_addresses": ["hub:ops-hub", "agent:operator"], "body": "Non-secret runtime smoke.", "subject_refs": {"hub": "ops-hub"}, } sent = runtime.post("/ports/messaging/messages", json=body) listed = runtime.get( "/ports/messaging/messages", params={"address": "hub:ops-hub", "conversation_id": str(conversation_id)}, ) assert sent.status_code == 202 assert listed.status_code == 200 assert len(listed.json()["items"]) == 1 assert listed.json()["items"][0]["data"]["body"] == body["body"] def test_progress_and_interaction_events_stay_separate() -> None: runtime = client() progress = runtime.post("/ports/events/progress", json=event_body("hub.progress.recorded")) interaction = runtime.post( "/ports/events/interaction", json=event_body("hub.interaction.recorded"), ) progress_projection = runtime.get("/ports/projections/progress_events").json() interaction_projection = runtime.get("/ports/projections/interaction_events").json() assert progress.status_code == 202 assert interaction.status_code == 202 assert [item["family"] for item in progress_projection["data"]["items"]] == ["progress"] assert [item["family"] for item in interaction_projection["data"]["items"]] == [ "interaction" ] def test_event_ports_reject_wrong_or_uncataloged_families() -> None: runtime = client() wrong_family = runtime.post( "/ports/events/progress", json=event_body("hub.interaction.recorded"), ) unknown = runtime.post( "/ports/events/interaction", json=event_body("hub.interaction.unknown"), ) assert wrong_family.status_code == 422 assert "belongs to 'interaction'" in wrong_family.json()["detail"] assert unknown.status_code == 422 assert "is not cataloged" in unknown.json()["detail"] def test_unknown_projection_is_not_found() -> None: response = client().get("/ports/projections/not-a-projection") assert response.status_code == 404 def test_runtime_openapi_marks_the_five_minimal_ports() -> None: document = client().get("/openapi.json").json() port_ids = { operation["x-port-id"] for path_item in document["paths"].values() for method, operation in path_item.items() if method in {"get", "post", "put", "patch", "delete"} and "x-port-id" in operation } assert port_ids == { "port.registry", "port.messaging", "port.events.progress", "port.events.interaction", "port.projection.query", } def test_cli_exposes_api_mcp_and_migration_processes() -> None: parser = build_parser(RuntimeSettings()) assert parser.parse_args(["api"]).command == "api" assert parser.parse_args(["mcp"]).command == "mcp" assert parser.parse_args(["migrate", "head", "--database-url", "postgresql://db"]).command == ( "migrate" ) assert _sync_database_url("postgresql+asyncpg://db") == "postgresql+psycopg2://db"