133 lines
5 KiB
Python
133 lines
5 KiB
Python
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())
|