repo-manager/tests/test_sbom.py

from __future__ import annotations

import json
import subprocess
from pathlib import Path

import pytest

from repo_manager.cli import main
from repo_manager.sbom_client import (
    SBOMContractError,
    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"