generated from coulomb/repo-seed
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
from railiance_fabric.cli import main as cli_main
|
|
from railiance_fabric.registry import RegistryStore
|
|
from railiance_fabric.server import RegistryHandler
|
|
|
|
|
|
def test_registry_scan_manifest_dry_run_keeps_repo_failures_isolated(tmp_path: Path, capsys) -> None:
|
|
repo = _minimal_repo(tmp_path, "fixture-repo")
|
|
manifest = _manifest(
|
|
tmp_path,
|
|
[
|
|
{"slug": "fixture-repo", "name": "Fixture Repo", "path": str(repo)},
|
|
{"slug": "missing-repo", "name": "Missing Repo", "path": str(tmp_path / "missing-repo")},
|
|
],
|
|
)
|
|
output_dir = tmp_path / "snapshots"
|
|
|
|
assert cli_main(
|
|
[
|
|
"registry",
|
|
"scan-manifest",
|
|
str(manifest),
|
|
"--dry-run",
|
|
"--output-dir",
|
|
str(output_dir),
|
|
"--json",
|
|
]
|
|
) == 0
|
|
|
|
summary = json.loads(capsys.readouterr().out)
|
|
assert summary["counts"]["total"] == 2
|
|
assert summary["counts"]["scanned"] == 1
|
|
assert summary["counts"]["errors"] == 1
|
|
assert summary["counts"]["llm_skipped"] == 2
|
|
assert summary["repositories"][0]["status"] == "scanned"
|
|
assert summary["repositories"][1]["status"] == "error"
|
|
assert (output_dir / "fixture-repo-deterministic.discovery.json").is_file()
|
|
|
|
|
|
def test_registry_scan_manifest_ingests_and_accepts_snapshots(tmp_path: Path, capsys) -> None:
|
|
repo = _minimal_repo(tmp_path, "fixture-repo")
|
|
manifest = _manifest(
|
|
tmp_path,
|
|
[{"slug": "fixture-repo", "name": "Fixture Repo", "domain": "testing", "path": str(repo)}],
|
|
)
|
|
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}"
|
|
assert cli_main(
|
|
[
|
|
"registry",
|
|
"scan-manifest",
|
|
str(manifest),
|
|
"--registry-url",
|
|
base_url,
|
|
"--repo-slug",
|
|
"fixture-repo",
|
|
"--ingest",
|
|
"--accept",
|
|
"--json",
|
|
]
|
|
) == 0
|
|
|
|
summary = json.loads(capsys.readouterr().out)
|
|
assert summary["counts"]["total"] == 1
|
|
assert summary["counts"]["scanned"] == 1
|
|
assert summary["counts"]["ingested"] == 1
|
|
assert summary["counts"]["accepted"] == 1
|
|
assert summary["counts"]["errors"] == 0
|
|
assert summary["repositories"][0]["discovery_snapshot_id"] == 1
|
|
assert store.list_discovery_snapshots("fixture-repo")[0]["id"] == 1
|
|
assert store.latest_snapshots()[0]["commit"].startswith("discovery:")
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def _minimal_repo(tmp_path: Path, slug: str) -> Path:
|
|
repo = tmp_path / slug
|
|
repo.mkdir()
|
|
(repo / "README.md").write_text(f"# {slug}\n", encoding="utf-8")
|
|
return repo
|
|
|
|
|
|
def _manifest(tmp_path: Path, repositories: list[dict[str, object]]) -> Path:
|
|
manifest = tmp_path / "local-repos.yaml"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"apiVersion": "railiance.fabric/v1alpha1",
|
|
"kind": "RegistryOnboardingManifest",
|
|
"registry_url": "http://127.0.0.1:8765",
|
|
"repositories": repositories,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return manifest
|