repo-manager/tests/test_sbom.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 (
    SBOMContractError,
    SBOMNexusClient,
    SBOMNexusConfig,
    SBOMServiceError,
    licence_report_from_snapshot,
    scan_repository_via_nexus,
    validate_snapshot_contract,
)


def _snapshot() -> dict:
    return {
        "schema": "sbom-nexus.snapshot.v1",
        "ok": True,
        "repo_slug": "example",
        "source_revision": "abc123",
        "generated_at": "2026-08-22T18:00:00Z",
        "entry_count": 1,
        "entries": [
            {
                "package_name": "pyyaml",
                "package_version": "6.0.3",
                "ecosystem": "python",
                "license_spdx": "MIT",
                "is_direct": True,
                "is_dev": False,
                "source_path": "uv.lock",
            }
        ],
        "sources": [{"path": "uv.lock", "entry_count": 1, "sha256": "abc"}],
        "licence_report": {
            "groups": [
                {
                    "license_spdx": "MIT",
                    "count": 1,
                    "is_copyleft": False,
                }
            ],
            "copyleft_direct_prod": [],
            "copyleft_direct_count": 0,
        },
        "errors": [],
    }


def test_scan_delegates_to_sbom_nexus_without_shell(monkeypatch, tmp_path: Path) -> None:
    observed = {}

    def fake_run(command, **kwargs):
        observed["command"] = command
        observed["kwargs"] = kwargs
        return subprocess.CompletedProcess(command, 0, json.dumps(_snapshot()), "")

    monkeypatch.setenv("SBOM_NEXUS_CLI", "/opt/sbom-nexus/bin/sbom-nexus")
    monkeypatch.setattr(subprocess, "run", fake_run)

    result = scan_repository_via_nexus(tmp_path, slug="example")

    assert observed["command"] == [
        "/opt/sbom-nexus/bin/sbom-nexus",
        "scan",
        str(tmp_path),
        "--slug",
        "example",
    ]
    assert observed["kwargs"] == {
        "capture_output": True,
        "text": True,
        "check": False,
    }
    assert result["schema"] == "sbom-nexus.snapshot.v1"
    assert result["product_owner"] == "sbom-nexus"
    assert result["delegated_by"] == "repo-manager"
    assert result["repo_manager_context"] == {
        "mode": "local-preview",
        "authoritative": False,
        "persisted": False,
        "advances_last_attempt_at": False,
        "advances_last_success_at": False,
        "creates_snapshot_history": False,
    }


def test_missing_nexus_cli_returns_actionable_error(monkeypatch, tmp_path: Path) -> None:
    monkeypatch.delenv("SBOM_NEXUS_CLI", raising=False)
    monkeypatch.setattr("repo_manager.sbom_client.shutil.which", lambda _name: None)

    result = scan_repository_via_nexus(tmp_path)

    assert result["ok"] is False
    assert result["schema"] == "sbom-nexus.snapshot.v1"
    assert "SBOM_NEXUS_CLI" in result["errors"][0]["detail"]


def test_licence_report_alias_preserves_shape() -> None:
    result = licence_report_from_snapshot(_snapshot())

    assert set(result) == {
        "ok",
        "repo_slug",
        "source_revision",
        "generated_at",
        "entry_count",
        "licence_report",
        "errors",
        "repo_manager_context",
        "delegated_by",
        "product_owner",
    }
    assert result["licence_report"]["copyleft_direct_count"] == 0
    assert result["repo_manager_context"]["authoritative"] is False


def test_snapshot_contract_allows_additive_fields_and_rejects_unknown_schema() -> None:
    validate_snapshot_contract({**_snapshot(), "future_addition": {"accepted": True}})

    with pytest.raises(SBOMContractError, match="unsupported SBOM Nexus schema"):
        validate_snapshot_contract({**_snapshot(), "schema": "sbom-nexus.snapshot.v2"})


def test_scan_turns_unknown_schema_into_deterministic_contract_error(
    monkeypatch, tmp_path: Path
) -> None:
    monkeypatch.setenv("SBOM_NEXUS_CLI", "/opt/sbom-nexus/bin/sbom-nexus")
    monkeypatch.setattr(
        subprocess,
        "run",
        lambda command, **kwargs: subprocess.CompletedProcess(
            command,
            0,
            json.dumps({**_snapshot(), "schema": "future.snapshot.v9"}),
            "",
        ),
    )

    result = scan_repository_via_nexus(tmp_path)

    assert result["ok"] is False
    assert result["errors"][0]["reason"] == "sbom-nexus-contract"
    assert result["repo_manager_context"]["persisted"] is False


def test_cli_scan_preserves_output_file_behavior(monkeypatch, tmp_path: Path, capsys) -> None:
    monkeypatch.setattr(
        "repo_manager.sbom_client.scan_repository_via_nexus",
        lambda path, slug=None: {
            **_snapshot(),
            "delegated_by": "repo-manager",
            "product_owner": "sbom-nexus",
        },
    )
    output = tmp_path / "snapshot.json"

    exit_code = main(
        [
            "sbom",
            "scan",
            "--path",
            str(tmp_path),
            "--slug",
            "example",
            "--output",
            str(output),
        ]
    )

    assert exit_code == 0
    assert json.loads(output.read_text())["schema"] == "sbom-nexus.snapshot.v1"
    assert json.loads(capsys.readouterr().out)["product_owner"] == "sbom-nexus"


def test_service_config_is_explicit_bounded_and_redacts_token(monkeypatch) -> None:
    monkeypatch.setenv("SBOM_NEXUS_URL", "https://nexus.example.test/")
    monkeypatch.setenv("SBOM_NEXUS_TIMEOUT_SECONDS", "12.5")
    monkeypatch.setenv("SBOM_NEXUS_TOKEN", "do-not-print-this")

    config = SBOMNexusConfig.from_environment()

    assert config.base_url == "https://nexus.example.test"
    assert config.timeout_seconds == 12.5
    assert "do-not-print-this" not in repr(config)
    with pytest.raises(ValueError, match="must not contain credentials"):
        SBOMNexusConfig("https://user:secret@nexus.example.test")
    with pytest.raises(ValueError, match="between 0.1 and 300"):
        SBOMNexusConfig("https://nexus.example.test", timeout_seconds=301)


def test_authoritative_client_calls_pinned_routes_and_marks_service_context() -> None:
    calls = []

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        if request.url.path.startswith("/repositories/"):
            return httpx.Response(200, json={"slug": "demo", "active": True})
        if request.method == "POST":
            return httpx.Response(
                200,
                json={
                    "repo_slug": "demo",
                    "snapshot_id": "snapshot-1",
                    "status": "ingested",
                    "entry_count": 2,
                    "snapshot_at": "2026-08-22T20:00:00Z",
                    "source_revision": "abc123",
                },
            )
        if request.url.path == "/sbom/report/licences/":
            return httpx.Response(
                200,
                json={
                    "groups": [],
                    "copyleft_direct_prod": [],
                    "copyleft_direct_count": 0,
                },
            )
        return httpx.Response(
            200,
            json={
                "repo_slug": "demo",
                "last_attempt_at": "2026-08-22T20:00:00Z",
                "last_success_at": "2026-08-22T20:00:00Z",
                "last_status": "ingested",
                "entry_count": 2,
                "snapshot_id": "snapshot-1",
                "entries": [],
            },
        )

    client = SBOMNexusClient(
        SBOMNexusConfig("https://nexus.example.test", bearer_token="runtime-secret"),
        transport=httpx.MockTransport(handler),
    )

    projected = client.upsert_repository(
        "demo",
        nexus_checkout_path="/srv/controlled/demo/abc123",
    )
    ingested = client.ingest_repository(
        "demo",
        expected_source_revision="abc123",
        operation_id="operation-1",
    )
    latest = client.latest_snapshot("demo")
    report = client.licence_report()

    assert [request.url.path for request in calls] == [
        "/repositories/demo",
        "/sbom/demo/ingest",
        "/sbom/demo",
        "/sbom/report/licences/",
    ]
    assert all(request.headers["Authorization"] == "Bearer runtime-secret" for request in calls)
    assert calls[1].headers["Idempotency-Key"] == "operation-1"
    assert projected["repo_manager_context"]["writes_state"] is True
    assert ingested["repo_manager_context"]["mode"] == "authoritative-service"
    assert latest["repo_manager_context"]["authoritative"] is True
    assert report["repo_manager_context"]["operation"] == "licence-report"


def test_authoritative_client_fails_closed_on_source_revision_mismatch() -> None:
    transport = httpx.MockTransport(
        lambda request: httpx.Response(
            200,
            json={
                "repo_slug": "demo",
                "snapshot_id": "snapshot-1",
                "status": "ingested",
                "entry_count": 1,
                "snapshot_at": "2026-08-22T20:00:00Z",
                "source_revision": "different",
            },
        )
    )
    client = SBOMNexusClient(SBOMNexusConfig("https://nexus.example.test"), transport=transport)

    with pytest.raises(SBOMServiceError) as raised:
        client.ingest_repository("demo", expected_source_revision="expected")

    assert raised.value.code == "source_revision_mismatch"
    assert raised.value.mutation_may_have_committed is True


def test_authoritative_client_has_sanitized_timeout_and_http_failures() -> None:
    secret = "never-echo-this-token"

    def timeout_handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("upstream timed out", request=request)

    timeout_client = SBOMNexusClient(
        SBOMNexusConfig("https://nexus.example.test", bearer_token=secret),
        transport=httpx.MockTransport(timeout_handler),
    )
    with pytest.raises(SBOMServiceError) as timeout:
        timeout_client.latest_snapshot("demo")
    assert timeout.value.to_dict()["code"] == "timeout"
    assert secret not in str(timeout.value)
    assert secret not in json.dumps(timeout.value.to_dict())

    http_client = SBOMNexusClient(
        SBOMNexusConfig("https://nexus.example.test", bearer_token=secret),
        transport=httpx.MockTransport(lambda request: httpx.Response(503, json={"detail": secret})),
    )
    with pytest.raises(SBOMServiceError) as failed:
        http_client.ingest_repository("demo")
    assert failed.value.status_code == 503
    assert failed.value.mutation_may_have_committed is True
    assert secret not in str(failed.value)
    assert secret not in json.dumps(failed.value.to_dict())


def test_authoritative_client_rejects_malformed_success_without_exposing_body() -> None:
    client = SBOMNexusClient(
        SBOMNexusConfig("https://nexus.example.test"),
        transport=httpx.MockTransport(
            lambda request: httpx.Response(200, json={"unexpected": "sensitive body"})
        ),
    )

    with pytest.raises(SBOMServiceError) as failed:
        client.latest_snapshot("demo")

    assert failed.value.code == "contract_error"
    assert "sensitive body" not in str(failed.value)