repo-manager/tests/test_source_ref.py

from __future__ import annotations

import json
import subprocess
from pathlib import Path

import httpx
import pytest

from repo_manager.cli import main
from repo_manager.sbom_client import SBOMNexusClient, SBOMNexusConfig, SBOMServiceError
from repo_manager.source_ref import ForgejoSourceResolver, canonical_coulomb_repository


def _git(repo: Path, *args: str) -> None:
    subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True)


def _repo(tmp_path: Path, remote: str) -> Path:
    repo = tmp_path / "demo"
    repo.mkdir()
    _git(repo, "init")
    _git(repo, "remote", "add", "origin", remote)
    return repo


@pytest.mark.parametrize(
    "remote",
    [
        "forgejo-remote:coulomb/demo.git",
        "git@forgejo.coulomb.social:coulomb/demo.git",
        "ssh://git@forgejo.coulomb.social/coulomb/demo.git",
        "https://forgejo.coulomb.social/coulomb/demo.git",
    ],
)
def test_normalizes_only_canonical_coulomb_repository(remote: str) -> None:
    assert canonical_coulomb_repository(remote, "demo") == "coulomb/demo"
    assert canonical_coulomb_repository(remote, "other") is None


def test_rejects_non_coulomb_and_untrusted_remote_hosts() -> None:
    assert (
        canonical_coulomb_repository("https://forgejo.coulomb.social/other/demo.git", "demo")
        is None
    )
    assert canonical_coulomb_repository("https://github.com/coulomb/demo.git", "demo") is None


def test_resolves_public_default_branch_from_forgejo_not_local_head(tmp_path: Path) -> None:
    repo = _repo(tmp_path, "forgejo-remote:coulomb/demo.git")
    revision = "a" * 40

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/api/v1/repos/coulomb/demo":
            return httpx.Response(
                200,
                json={
                    "full_name": "coulomb/demo",
                    "private": False,
                    "default_branch": "main",
                },
            )
        return httpx.Response(
            200,
            json=[
                {
                    "ref": "refs/heads/main",
                    "object": {"type": "commit", "sha": revision},
                }
            ],
        )

    result = ForgejoSourceResolver(
        transport=httpx.MockTransport(handler),
        now=lambda: "2026-08-22T22:00:00Z",
    ).resolve(repo)

    assert result == {
        "ok": True,
        "supported": True,
        "repo_slug": "demo",
        "source_ref": {
            "kind": "forgejo-archive-v1",
            "repository": "coulomb/demo",
            "revision": revision,
            "observed_ref": "refs/heads/main",
            "observed_at": "2026-08-22T22:00:00Z",
        },
    }


def test_private_or_unresolvable_repository_has_no_v1_source_ref(tmp_path: Path) -> None:
    repo = _repo(tmp_path, "https://forgejo.coulomb.social/coulomb/demo.git")
    private = ForgejoSourceResolver(
        transport=httpx.MockTransport(
            lambda request: httpx.Response(
                200,
                json={
                    "full_name": "coulomb/demo",
                    "private": True,
                    "default_branch": "main",
                },
            )
        )
    ).resolve(repo)
    assert private["supported"] is False
    assert private["reason"] == "private-repository"
    assert private["source_ref"] is None

    unavailable = ForgejoSourceResolver(
        transport=httpx.MockTransport(lambda request: httpx.Response(404))
    ).resolve(repo)
    assert unavailable["supported"] is False
    assert unavailable["reason"] == "source-unresolvable"


def test_projection_requires_nexus_to_echo_exact_source_ref() -> None:
    source_ref = {
        "kind": "forgejo-archive-v1",
        "repository": "coulomb/demo",
        "revision": "a" * 40,
        "observed_ref": "refs/heads/main",
        "observed_at": "2026-08-22T22:00:00Z",
    }
    bodies = []

    def echo(request: httpx.Request) -> httpx.Response:
        body = json.loads(request.content)
        bodies.append(body)
        return httpx.Response(
            200,
            json={"slug": "demo", "active": True, "source_ref": body["source_ref"]},
        )

    client = SBOMNexusClient(
        SBOMNexusConfig("https://nexus.example.test"),
        transport=httpx.MockTransport(echo),
    )
    result = client.upsert_repository(
        "demo",
        nexus_checkout_path=None,
        source_ref=source_ref,
    )

    assert bodies == [{"checkout_path": None, "active": True, "source_ref": source_ref}]
    assert result["source_ref"] == source_ref

    dropping_client = SBOMNexusClient(
        SBOMNexusConfig("https://nexus.example.test"),
        transport=httpx.MockTransport(
            lambda request: httpx.Response(200, json={"slug": "demo", "active": True})
        ),
    )
    with pytest.raises(SBOMServiceError) as failed:
        dropping_client.upsert_repository(
            "demo",
            nexus_checkout_path=None,
            source_ref=source_ref,
        )
    assert failed.value.code == "contract_error"
    assert failed.value.mutation_may_have_committed is True


def test_cli_source_ref_is_read_only_until_explicit_projection(
    monkeypatch, tmp_path: Path, capsys
) -> None:
    source_ref = {
        "kind": "forgejo-archive-v1",
        "repository": "coulomb/demo",
        "revision": "a" * 40,
        "observed_ref": "refs/heads/main",
        "observed_at": "2026-08-22T22:00:00Z",
    }
    resolution = {
        "ok": True,
        "supported": True,
        "repo_slug": "demo",
        "source_ref": source_ref,
    }
    monkeypatch.setattr(
        "repo_manager.source_ref.ForgejoSourceResolver.resolve",
        lambda self, path, repo_slug=None, remote_name="origin": resolution,
    )

    assert main(["sbom", "source-ref", "--path", str(tmp_path)]) == 0
    assert json.loads(capsys.readouterr().out) == resolution

    assert main(["sbom", "source-ref", "--path", str(tmp_path), "--project"]) == 2
    rejected = json.loads(capsys.readouterr().out)
    assert rejected["error"] == "--project requires --confirm-authoritative"


def test_cli_projects_source_ref_through_configured_nexus(
    monkeypatch, tmp_path: Path, capsys
) -> None:
    source_ref = {
        "kind": "forgejo-archive-v1",
        "repository": "coulomb/demo",
        "revision": "a" * 40,
        "observed_ref": "refs/heads/main",
        "observed_at": "2026-08-22T22:00:00Z",
    }
    monkeypatch.setattr(
        "repo_manager.source_ref.ForgejoSourceResolver.resolve",
        lambda self, path, repo_slug=None, remote_name="origin": {
            "ok": True,
            "supported": True,
            "repo_slug": "demo",
            "source_ref": source_ref,
        },
    )
    monkeypatch.setenv("SBOM_NEXUS_URL", "https://nexus.example.test")
    monkeypatch.setattr(
        "repo_manager.sbom_client.SBOMNexusClient.upsert_repository",
        lambda self, repo_slug, nexus_checkout_path, active=True, source_ref=None: {
            "slug": repo_slug,
            "source_ref": source_ref,
        },
    )

    exit_code = main(
        [
            "sbom",
            "source-ref",
            "--path",
            str(tmp_path),
            "--project",
            "--confirm-authoritative",
        ]
    )

    assert exit_code == 0
    result = json.loads(capsys.readouterr().out)
    assert result["projection"]["source_ref"] == source_ref