generated from coulomb/repo-seed
83 lines
2.7 KiB
Python
83 lines
2.7 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,
|
|
blast_radius,
|
|
consumers,
|
|
providers,
|
|
unresolved_dependencies,
|
|
)
|
|
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()
|
|
|
|
assert snapshot["repo_slug"] == "railiance-fabric"
|
|
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")
|
|
|
|
|
|
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())
|
|
assert providers_payload[0]["provider_id"] == "railiance-platform.openbao.runtime-secrets"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|