repo-manager/tests/test_repository_publisher.py

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",
        }
    ]