coulomb/repo-manager
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
repo-manager/tests/test_repository_publisher.py
tegwick fd624021ec feat: publish classifications from Forgejo 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c
2026-09-01 01:39:39 +02:00

419 lines
14 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import base64
import subprocess
from datetime import UTC, datetime
from pathlib import Path
import httpx
import yaml
from fastapi.testclient import TestClient
from repo_manager.repository_publisher import (
ClassificationPublisher,
ForgejoClassificationClient,
ProjectionCursorError,
RepositoryRegistration,
build_classification_snapshot,
import_state_hub_registry,
load_repository_registry,
)
from repo_manager.runtime import PublisherSettings, create_app
NOW = datetime(2026, 9, 1, 8, 30, tzinfo=UTC)
SECRET = "projection-test-secret-is-at-least-32-bytes"
def _repo(root: Path, name: str, *, domain: str = "infotech") -> Path:
repo = root / name
repo.mkdir()
subprocess.run(["git", "init", "-q", "-b", "main"], cwd=repo, check=True)
subprocess.run(["git", "config", "user.name", "Test"], cwd=repo, check=True)
subprocess.run(["git", "config", "user.email", "test@example.invalid"], cwd=repo, check=True)
(repo / ".repo-classification.yaml").write_text(
yaml.safe_dump(
{
"repo_classification": {
"category": "tooling",
"domain": domain,
"secondary_domains": ["agents"],
"capability_tags": ["repository-governance"],
"business_stake": ["technology"],
"business_mechanics": ["control"],
}
},
sort_keys=False,
),
encoding="utf-8",
)
subprocess.run(["git", "add", ".repo-classification.yaml"], cwd=repo, check=True)
subprocess.run(["git", "commit", "-qm", "classify"], cwd=repo, check=True)
return repo
def _registry(path: Path, registrations: list[dict]) -> Path:
path.write_text(
yaml.safe_dump(
{"schema": "repo-manager.repository-registry.v1", "repositories": registrations},
sort_keys=False,
),
encoding="utf-8",
)
return path
def test_registry_rejects_duplicate_stable_identity(tmp_path: Path) -> None:
repo = _repo(tmp_path, "one")
registry = _registry(
tmp_path / "registry.yaml",
[
{
"repository_id": "11111111-1111-4111-8111-111111111111",
"slug": "one",
"lifecycle": "active",
"path": str(repo),
},
{
"repository_id": "11111111-1111-4111-8111-111111111111",
"slug": "two",
"lifecycle": "active",
"path": str(repo),
},
],
)
try:
load_repository_registry(registry)
except ValueError as exc:
assert "duplicate repository_id" in str(exc)
else:
raise AssertionError("duplicate stable identity was accepted")
def test_registry_accepts_exact_forgejo_identity(tmp_path: Path) -> None:
registry = _registry(
tmp_path / "registry.yaml",
[
{
"repository_id": "11111111-1111-4111-8111-111111111111",
"slug": "repo-manager",
"lifecycle": "active",
"forgejo_repository": "coulomb/repo-manager",
}
],
)
registrations = load_repository_registry(registry)
assert registrations[0].path is None
assert registrations[0].forgejo_repository == "coulomb/repo-manager"
def test_snapshot_is_uuid_ordered_and_carries_source_provenance(tmp_path: Path) -> None:
first = _repo(tmp_path, "first")
second = _repo(tmp_path, "second")
registrations = (
RepositoryRegistration(
"22222222-2222-4222-8222-222222222222", "second", "active", second
),
RepositoryRegistration(
"11111111-1111-4111-8111-111111111111", "first", "archived", first
),
)
snapshot = build_classification_snapshot(registrations, observed_at=NOW)
assert [row["repository_id"] for row in snapshot.repositories] == [
"11111111-1111-4111-8111-111111111111",
"22222222-2222-4222-8222-222222222222",
]
assert snapshot.repositories[0]["lifecycle"] == "archived"
assert snapshot.repositories[0]["revision"]["observed_at"] == "2026-09-01T08:30:00Z"
assert len(snapshot.repositories[0]["revision"]["head_sha"]) == 40
assert len(snapshot.repositories[0]["revision"]["source_fingerprint"]) == 64
assert snapshot.diagnostics == ()
def test_paging_is_bound_to_an_immutable_snapshot_and_signed(tmp_path: Path) -> None:
first = _repo(tmp_path, "first")
second = _repo(tmp_path, "second")
registry = _registry(
tmp_path / "registry.yaml",
[
{
"repository_id": "22222222-2222-4222-8222-222222222222",
"slug": "second",
"lifecycle": "active",
"path": str(second),
},
{
"repository_id": "11111111-1111-4111-8111-111111111111",
"slug": "first",
"lifecycle": "active",
"path": str(first),
},
],
)
publisher = ClassificationPublisher(
registry, cursor_secret=SECRET, page_size=1, now=lambda: NOW
)
page_one = publisher.fetch_page()
cursor = page_one["snapshot"]["next_cursor"]
page_two = publisher.fetch_page(cursor)
assert page_one["contract_id"] == "helixforge.repository-classification-projection"
assert page_one["contract_version"] == "1.0.0"
assert page_one["snapshot"]["page_cursor"] is None
assert page_one["snapshot"]["final_page"] is False
assert page_two["snapshot"]["page_cursor"] == cursor
assert page_two["snapshot"]["final_page"] is True
assert page_two["snapshot"]["next_cursor"] is None
assert page_two["snapshot"]["snapshot_id"] == page_one["snapshot"]["snapshot_id"]
assert page_one["repositories"][0]["repository_id"] < page_two["repositories"][0][
"repository_id"
]
tampered = ("A" if cursor[0] != "A" else "B") + cursor[1:]
try:
publisher.fetch_page(tampered)
except ProjectionCursorError as exc:
assert "signature" in str(exc) or "malformed" in str(exc)
else:
raise AssertionError("tampered projection cursor was accepted")
def test_invalid_source_is_an_error_diagnostic_not_a_partial_upsert(tmp_path: Path) -> None:
repo = _repo(tmp_path, "broken")
(repo / ".repo-classification.yaml").write_text(
"repo_classification:\n category: unknown\n domain: infotech\n",
encoding="utf-8",
)
snapshot = build_classification_snapshot(
(
RepositoryRegistration(
"11111111-1111-4111-8111-111111111111", "broken", "active", repo
),
),
observed_at=NOW,
)
assert snapshot.repositories == ()
assert snapshot.diagnostics[0]["severity"] == "error"
assert snapshot.diagnostics[0]["code"] == "repo_projection.source_invalid"
def test_uncommitted_classification_cannot_be_published_as_head(tmp_path: Path) -> None:
repo = _repo(tmp_path, "dirty")
(repo / ".repo-classification.yaml").write_text(
"repo_classification:\n category: product\n domain: infotech\n",
encoding="utf-8",
)
snapshot = build_classification_snapshot(
(
RepositoryRegistration(
"11111111-1111-4111-8111-111111111111", "dirty", "active", repo
),
),
observed_at=NOW,
)
assert snapshot.repositories == ()
assert "uncommitted changes" in snapshot.diagnostics[0]["message"]
def test_forgejo_snapshot_is_pinned_to_the_observed_default_branch_head() -> None:
head = "a" * 40
classification = yaml.safe_dump(
{
"repo_classification": {
"category": "tooling",
"domain": "infotech",
"secondary_domains": ["agents"],
"capability_tags": ["repository-governance"],
"business_stake": ["technology"],
"business_mechanics": ["control"],
}
}
).encode()
seen_refs: list[str | None] = []
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/api/v1/repos/coulomb/repo-manager":
return httpx.Response(
200,
json={"full_name": "coulomb/repo-manager", "default_branch": "main"},
)
if request.url.path.endswith("/branches/main"):
return httpx.Response(200, json={"commit": {"id": head}})
if request.url.path.endswith("/contents/.repo-classification.yaml"):
seen_refs.append(request.url.params.get("ref"))
return httpx.Response(
200,
json={
"type": "file",
"encoding": "base64",
"content": base64.b64encode(classification).decode(),
},
)
return httpx.Response(404)
forgejo = ForgejoClassificationClient(
"https://forgejo.invalid", transport=httpx.MockTransport(handler)
)
snapshot = build_classification_snapshot(
(
RepositoryRegistration(
"11111111-1111-4111-8111-111111111111",
"repo-manager",
"active",
forgejo_repository="coulomb/repo-manager",
),
),
observed_at=NOW,
forgejo_client=forgejo,
)
forgejo.close()
assert seen_refs == [head]
assert snapshot.diagnostics == ()
assert snapshot.repositories[0]["revision"]["head_sha"] == head
assert snapshot.repositories[0]["classification"]["domain"] == "infotech"
def test_forgejo_identity_mismatch_is_a_bounded_diagnostic() -> None:
def handler(_request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={"full_name": "coulomb/wrong", "default_branch": "main"},
)
forgejo = ForgejoClassificationClient(
"https://forgejo.invalid", transport=httpx.MockTransport(handler)
)
snapshot = build_classification_snapshot(
(
RepositoryRegistration(
"11111111-1111-4111-8111-111111111111",
"repo-manager",
"active",
forgejo_repository="coulomb/repo-manager",
),
),
observed_at=NOW,
forgejo_client=forgejo,
)
forgejo.close()
assert snapshot.repositories == ()
assert snapshot.diagnostics[0]["code"] == "repo_projection.source_invalid"
assert "identity" in snapshot.diagnostics[0]["message"]
def test_http_port_reports_readiness_and_serves_the_frozen_envelope(tmp_path: Path) -> None:
repo = _repo(tmp_path, "repo-manager")
registry = _registry(
tmp_path / "registry.yaml",
[
{
"repository_id": "11111111-1111-4111-8111-111111111111",
"slug": "repo-manager",
"lifecycle": "active",
"path": str(repo),
}
],
)
app = create_app(
settings=PublisherSettings(
registry_path=registry,
cursor_secret=SECRET,
api_token="publisher-token",
page_size=100,
)
)
with TestClient(app) as client:
assert client.get("/readyz").json() == {"status": "ok", "repository_count": 1}
assert client.get("/ports/repositories/classifications").status_code == 401
response = client.get(
"/ports/repositories/classifications",
headers={"Authorization": "Bearer publisher-token"},
)
assert response.status_code == 200
assert response.json()["snapshot"]["total_repository_count"] == 1
methods = {
method
for method in client.get("/openapi.json").json()["paths"][
"/ports/repositories/classifications"
]
}
assert methods == {"get"}
def test_registry_bootstrap_uses_primary_direct_identity_reads(tmp_path: Path) -> None:
_repo(tmp_path, "first")
_repo(tmp_path, "second")
seen: list[str] = []
def handler(request: httpx.Request) -> httpx.Response:
seen.append(request.url.path)
if request.url.path == "/state/health":
return httpx.Response(
200,
json={"status": "ok", "instance_role": "primary", "instance_label": "railiance01"},
)
slug = request.url.path.removeprefix("/repos/")
identifiers = {
"first": "11111111-1111-4111-8111-111111111111",
"second": "22222222-2222-4222-8222-222222222222",
}
return httpx.Response(
200,
json={"id": identifiers[slug], "slug": slug, "status": "active"},
)
result = import_state_hub_registry(
tmp_path,
api_base="http://state-hub.invalid",
workers=2,
transport=httpx.MockTransport(handler),
)
assert result["ok"] is True
assert len(result["repositories"]) == 2
assert "/repos/" not in seen
assert {path for path in seen if path.startswith("/repos/")} == {
"/repos/first",
"/repos/second",
}
assert all(Path(row["path"]).is_absolute() for row in result["repositories"])
def test_registry_bootstrap_can_emit_forgejo_sources(tmp_path: Path) -> None:
_repo(tmp_path, "repo-manager")
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/state/health":
return httpx.Response(
200,
json={"status": "ok", "instance_role": "primary", "instance_label": "primary"},
)
return httpx.Response(
200,
json={
"id": "11111111-1111-4111-8111-111111111111",
"slug": "repo-manager",
"status": "active",
},
)
result = import_state_hub_registry(
tmp_path,
api_base="http://state-hub.invalid",
source="forgejo",
transport=httpx.MockTransport(handler),
)
assert result["repositories"] == [
{
"repository_id": "11111111-1111-4111-8111-111111111111",
"slug": "repo-manager",
"lifecycle": "active",
"forgejo_repository": "coulomb/repo-manager",
}
]
Reference in a new issue View git blame Copy permalink