coulomb/repo-manager
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: publish classifications from Forgejo

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c
This commit is contained in:
tegwick 2026-09-01 01:39:39 +02:00
parent 738d407bb7
commit fd624021ec
12 changed files with 2000 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

144
tests/test_repository_publisher.py
View file

@ -1,5 +1,6 @@
from __future__ import annotations
import base64
import subprocess
from datetime import UTC, datetime
from pathlib import Path
@ -10,6 +11,7 @@ from fastapi.testclient import TestClient
from repo_manager.repository_publisher import (
ClassificationPublisher,
ForgejoClassificationClient,
ProjectionCursorError,
RepositoryRegistration,
build_classification_snapshot,
@ -87,6 +89,25 @@ def test_registry_rejects_duplicate_stable_identity(tmp_path: Path) -> None:
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")
@ -197,6 +218,94 @@ def test_uncommitted_classification_cannot_be_published_as_head(tmp_path: Path)
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(
@ -273,3 +382,38 @@ def test_registry_bootstrap_uses_primary_direct_identity_reads(tmp_path: Path) -
"/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",
}
]