hub-core/tests/test_compatibility.py

from __future__ import annotations

import asyncio

from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import create_async_engine

from hub_core.runtime.app import create_app
from hub_core.runtime.config import RuntimeSettings
from hub_core.runtime.postgres_store import PostgresPortStore
from hub_core.runtime.tables import runtime_metadata

GROUPS = frozenset({"system", "registry", "credentials", "interaction", "deferred", "operator"})
HEADERS = {"Authorization": "Bearer operator-token"}


def compatibility_client(tmp_path, *, write_groups: frozenset[str] = GROUPS) -> tuple[TestClient, PostgresPortStore]:
    database_url = f"sqlite+aiosqlite:///{tmp_path / 'compat.db'}"

    async def create_schema() -> None:
        engine = create_async_engine(database_url)
        async with engine.begin() as connection:
            await connection.run_sync(runtime_metadata.create_all)
        await engine.dispose()

    asyncio.run(create_schema())
    store = PostgresPortStore.from_url(database_url)
    settings = RuntimeSettings(
        environment="production",
        backend="postgresql",
        allow_ephemeral=False,
        database_url=database_url,
        api_token="operator-token",
        v2_groups=GROUPS,
        v2_write_groups=write_groups,
        legacy_health=True,
    )
    return TestClient(create_app(settings=settings, port_store=store)), store


def test_route_group_configuration_rejects_two_writers() -> None:
    try:
        RuntimeSettings(
            v2_groups=frozenset({"registry"}),
            v2_write_groups=frozenset({"registry"}),
            legacy_write_groups=frozenset({"registry"}),
        )
    except ValueError as exc:
        assert "write groups overlap" in str(exc)
    else:
        raise AssertionError("overlapping writers were accepted")


def test_system_compatibility_and_protected_auth(tmp_path) -> None:
    client, store = compatibility_client(tmp_path)

    assert client.get("/healthz").json()["service"] == "core-hub"
    assert client.get("/api/v2/widget-types").status_code == 200
    denied = client.get("/api/v2/hubs")
    assert denied.status_code == 401
    assert denied.json()["detail"]["code"] == "unauthorized"
    document = client.get("/api/v2/openapi.json").json()
    assert "/api/v2/hubs" in document["paths"]
    assert "/hubs" in document["paths"]
    assert client.get("/widget-types").status_code == 200
    assert client.get("/hubs").status_code == 401
    assert client.get("/hubs", headers=HEADERS).status_code == 200
    assert client.get("/console", headers=HEADERS).status_code == 200
    asyncio.run(store.aclose())


def test_ops_hub_bootstrap_and_dynamic_key(tmp_path) -> None:
    client, store = compatibility_client(tmp_path)

    hub = client.post(
        "/api/v2/hubs",
        headers=HEADERS,
        json={"slug": "ops-hub", "name": "Ops Hub", "domain": "ops.coulomb.social"},
    ).json()
    manifest = client.post(
        "/api/v2/hub-capability-manifests",
        headers=HEADERS,
        json={"hubId": hub["id"], "manifestVersion": "1.0"},
    ).json()
    activated = client.post(
        f"/api/v2/hub-capability-manifests/{manifest['id']}/activate", headers=HEADERS
    )
    assert activated.json()["status"] == "active"
    consumer = client.post(
        "/api/v2/api-consumers",
        headers=HEADERS,
        json={"name": "ops-hub", "hubCapabilityManifestId": manifest["id"]},
    ).json()
    issued = client.post(
        f"/api/v2/api-consumers/{consumer['id']}/api-keys",
        headers=HEADERS,
        json={"scopes": "framework:read hub:ops-hub:write"},
    ).json()
    runtime_headers = {"Authorization": f"Bearer {issued['fullKey']}"}
    widget = client.post(
        "/api/v2/widgets",
        headers=runtime_headers,
        json={"hubId": hub["id"], "name": "Readiness", "widgetType": "ops-readiness-gate"},
    ).json()
    event = client.post(
        "/api/v2/interaction-events",
        headers=runtime_headers,
        json={
            "widgetId": widget["id"],
            "eventType": "ops-endpoint-verified",
            "metadata": {"expectedStatus": 401},
        },
    )

    assert event.status_code == 201
    assert client.get("/api/v2/hub-registry", headers=runtime_headers).status_code == 200
    assert client.get("/api/v2/interaction-events", headers=runtime_headers).json()["count"] == 1
    assert "fullKey" not in str(client.get("/api/v2/api-consumers", headers=HEADERS).json())
    asyncio.run(store.aclose())


def test_read_only_group_rejects_write(tmp_path) -> None:
    client, store = compatibility_client(tmp_path, write_groups=frozenset())
    response = client.post(
        "/api/v2/hubs", headers=HEADERS, json={"slug": "ops-hub", "name": "Ops Hub"}
    )

    assert response.status_code == 503
    assert response.json()["detail"] == "compatibility group is read-only"
    deferred = client.post("/annotations", headers=HEADERS)
    assert deferred.status_code == 503
    assert deferred.json()["detail"] == "compatibility group is read-only"
    asyncio.run(store.aclose())