generated from coulomb/repo-seed
Add multi-repo registry onboarding
This commit is contained in:
@@ -231,6 +231,63 @@ def test_registry_http_service_serves_queries(tmp_path: Path) -> None:
|
||||
thread.join(timeout=5)
|
||||
|
||||
|
||||
def test_registry_sync_manifest_registers_multiple_repos(tmp_path: Path) -> None:
|
||||
store = RegistryStore(tmp_path / "registry.sqlite3")
|
||||
store.init_schema()
|
||||
|
||||
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}"
|
||||
sbom_path = tmp_path / "bom.json"
|
||||
sbom_path.write_text(json.dumps(_cyclonedx_bom()), encoding="utf-8")
|
||||
manifest_path = tmp_path / "manifest.yaml"
|
||||
manifest_path.write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"apiVersion: railiance.fabric/v1alpha1",
|
||||
"kind: RegistryOnboardingManifest",
|
||||
"repositories:",
|
||||
" - slug: railiance-fabric",
|
||||
" name: Railiance Fabric",
|
||||
f" path: {Path('.').resolve()}",
|
||||
" commit: manifest-commit",
|
||||
f" sbom: {sbom_path.name}",
|
||||
" - slug: missing-repo",
|
||||
" name: Missing Repo",
|
||||
f" path: {tmp_path / 'missing-repo'}",
|
||||
"",
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
assert cli_main(
|
||||
[
|
||||
"registry",
|
||||
"sync-manifest",
|
||||
str(manifest_path),
|
||||
"--registry-url",
|
||||
base_url,
|
||||
]
|
||||
) == 0
|
||||
|
||||
repositories = {repo["slug"]: repo for repo in store.list_repositories()}
|
||||
assert set(repositories) == {"missing-repo", "railiance-fabric"}
|
||||
assert store.latest_snapshot("railiance-fabric")["commit"] == "manifest-commit"
|
||||
assert store.list_snapshots("missing-repo") == []
|
||||
assert store.list_libraries(repo_slug="railiance-fabric")[0]["name"] == "jsonschema"
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=5)
|
||||
|
||||
|
||||
def _post_json(url: str, payload: dict) -> dict:
|
||||
request = urllib.request.Request(
|
||||
url,
|
||||
|
||||
Reference in New Issue
Block a user