generated from coulomb/repo-seed
137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import urllib.request
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
from railiance_fabric.graph import build_graph
|
|
from railiance_fabric.registry import (
|
|
RegistryStore,
|
|
backstage_projection,
|
|
blast_radius,
|
|
consumers,
|
|
providers,
|
|
unresolved_dependencies,
|
|
xregistry_projection,
|
|
)
|
|
from railiance_fabric.server import RegistryHandler
|
|
|
|
|
|
def test_registry_accepts_snapshot_and_queries_graph(tmp_path: Path) -> None:
|
|
store = RegistryStore(tmp_path / "registry.sqlite3")
|
|
store.init_schema()
|
|
store.upsert_repository(
|
|
{
|
|
"slug": "railiance-fabric",
|
|
"name": "Railiance Fabric",
|
|
"remote_url": "https://example.invalid/railiance-fabric.git",
|
|
}
|
|
)
|
|
|
|
graph = build_graph([Path(".")])
|
|
snapshot = store.add_snapshot(
|
|
"railiance-fabric",
|
|
{
|
|
"commit": "test-commit",
|
|
"generated_at": "2026-05-17T00:00:00Z",
|
|
"graph": graph.to_export(),
|
|
},
|
|
)
|
|
combined = store.combined_graph()
|
|
artifact = store.add_artifact(
|
|
{
|
|
"repo_slug": "railiance-fabric",
|
|
"target_id": "flex-auth.api.http-api",
|
|
"target_kind": "InterfaceDeclaration",
|
|
"artifact_type": "openapi",
|
|
"name": "flex-auth OpenAPI",
|
|
"uri": "docs/contracts/flex-auth.openapi.yaml",
|
|
"media_type": "application/vnd.oai.openapi+yaml",
|
|
"version": "0.1.0",
|
|
"metadata": {"source": "test"},
|
|
}
|
|
)
|
|
|
|
assert snapshot["repo_slug"] == "railiance-fabric"
|
|
assert artifact["artifact_type"] == "openapi"
|
|
assert store.graph_node_detail("flex-auth.api.http-api")["artifacts"][0]["name"] == "flex-auth OpenAPI"
|
|
assert providers(combined, "runtime-secrets")[0]["provider_id"] == "railiance-platform.openbao.runtime-secrets"
|
|
assert {match["status"] for match in consumers(combined, "railiance-platform.openbao.kv-v2")} >= {"exact"}
|
|
assert unresolved_dependencies(combined) == []
|
|
assert blast_radius(combined, "openbao-kv-v2-mount")
|
|
assert any(item["kind"] == "Component" for item in backstage_projection(combined)["items"])
|
|
assert "services" in xregistry_projection(combined)["groups"]
|
|
|
|
|
|
def test_registry_http_service_serves_queries(tmp_path: Path) -> None:
|
|
store = RegistryStore(tmp_path / "registry.sqlite3")
|
|
store.init_schema()
|
|
store.upsert_repository({"slug": "railiance-fabric", "name": "Railiance Fabric"})
|
|
store.add_snapshot(
|
|
"railiance-fabric",
|
|
{
|
|
"commit": "test-commit",
|
|
"generated_at": "2026-05-17T00:00:00Z",
|
|
"graph": build_graph([Path(".")]).to_export(),
|
|
},
|
|
)
|
|
|
|
class Handler(RegistryHandler):
|
|
pass
|
|
|
|
Handler.store = store
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
base_url = f"http://127.0.0.1:{server.server_port}"
|
|
with urllib.request.urlopen(f"{base_url}/health", timeout=5) as response:
|
|
assert json.loads(response.read())["status"] == "ok"
|
|
with urllib.request.urlopen(
|
|
f"{base_url}/graph/providers?capability_type=runtime-secrets",
|
|
timeout=5,
|
|
) as response:
|
|
providers_payload = json.loads(response.read())
|
|
artifact_payload = _post_json(
|
|
f"{base_url}/artifacts",
|
|
{
|
|
"repo_slug": "railiance-fabric",
|
|
"target_id": "railiance-platform.openbao.kv-v2",
|
|
"target_kind": "InterfaceDeclaration",
|
|
"artifact_type": "openapi",
|
|
"name": "OpenBao KV API",
|
|
"uri": "https://example.invalid/openbao.yaml",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(
|
|
f"{base_url}/artifacts?target_id=railiance-platform.openbao.kv-v2",
|
|
timeout=5,
|
|
) as response:
|
|
artifacts_payload = json.loads(response.read())
|
|
with urllib.request.urlopen(f"{base_url}/exports/backstage", timeout=5) as response:
|
|
backstage_payload = json.loads(response.read())
|
|
with urllib.request.urlopen(f"{base_url}/exports/xregistry", timeout=5) as response:
|
|
xregistry_payload = json.loads(response.read())
|
|
assert providers_payload[0]["provider_id"] == "railiance-platform.openbao.runtime-secrets"
|
|
assert artifact_payload["name"] == "OpenBao KV API"
|
|
assert artifacts_payload[0]["artifact_type"] == "openapi"
|
|
assert backstage_payload["kind"] == "BackstageCatalogProjection"
|
|
assert "interfaces" in xregistry_payload["groups"]
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def _post_json(url: str, payload: dict) -> dict:
|
|
request = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return json.loads(response.read())
|