sbom-nexus/tests/test_api.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from pathlib import Path

from fastapi.testclient import TestClient

from sbom_nexus.api import create_app


def client_for(tmp_path: Path) -> TestClient:
    return TestClient(create_app(tmp_path / "nexus.db"))


def register(
    client: TestClient,
    slug: str,
    *,
    checkout_path: str | None = None,
    last_sbom_at: datetime | None = None,
    active: bool = True,
) -> None:
    response = client.put(
        f"/repositories/{slug}",
        json={
            "checkout_path": checkout_path,
            "last_sbom_at": last_sbom_at.isoformat() if last_sbom_at else None,
            "active": active,
        },
    )
    assert response.status_code == 200


def test_health_and_legacy_ingest_query_and_licence_report(tmp_path: Path) -> None:
    client = client_for(tmp_path)
    assert client.get("/state/health").json() == {
        "status": "ok",
        "store": "connected",
        "dialect": "sqlite",
    }
    register(client, "demo")

    response = client.post(
        "/sbom/ingest/",
        json={
            "repo_slug": "demo",
            "entries": [
                {
                    "package_name": "safe",
                    "package_version": "1.0",
                    "ecosystem": "python",
                    "license_spdx": "MIT",
                    "is_direct": True,
                    "is_dev": False,
                },
                {
                    "package_name": "copyleft",
                    "package_version": "2.0",
                    "ecosystem": "python",
                    "license_spdx": "GPL-3.0-only",
                    "is_direct": True,
                    "is_dev": False,
                },
            ],
        },
    )

    assert response.status_code == 200
    assert response.json()["ingested"] == 2
    assert len(client.get("/sbom/snapshots/?repo_slug=demo").json()) == 1
    assert client.get("/sbom/demo").json()["entry_count"] == 2
    assert len(client.get("/sbom/?license_spdx=MIT").json()) == 1
    report = client.get("/sbom/report/licences/").json()
    assert report["copyleft_direct_count"] == 1
    assert report["groups"][0]["repos"] == ["demo"]


def test_catch_up_is_bounded_null_first_then_oldest(tmp_path: Path) -> None:
    client = client_for(tmp_path)
    now = datetime.now(UTC)
    register(client, "never-b")
    register(client, "never-a")
    register(client, "oldest", last_sbom_at=now - timedelta(days=90))
    register(client, "old", last_sbom_at=now - timedelta(days=60))
    register(client, "fresh", last_sbom_at=now - timedelta(days=2))
    register(client, "inactive", active=False)

    result = client.get("/sbom/catch-up?limit=3").json()

    assert [repo["repo_slug"] for repo in result["repos"]] == [
        "never-a",
        "never-b",
        "oldest",
    ]
    assert result["selected_count"] == 3
    assert result["stale_count"] == 4
    assert result["never_count"] == 2
    assert result["total_count"] == 5


def test_recorded_skip_advances_queue_without_claiming_success(tmp_path: Path) -> None:
    client = client_for(tmp_path)
    register(client, "a-no-checkout")
    register(client, "b-never")

    before = client.get("/sbom/catch-up?limit=1").json()
    assert before["repos"][0]["repo_slug"] == "a-no-checkout"

    outcome = client.post("/sbom/a-no-checkout/ingest").json()
    assert outcome["status"] == "skipped"
    assert outcome["reason"] == "no-checkout"

    after = client.get("/sbom/catch-up?limit=1").json()
    assert after["repos"][0]["repo_slug"] == "b-never"
    repo = client.get("/sbom/a-no-checkout").json()
    assert repo["last_attempt_at"] is not None
    assert repo["last_success_at"] is None


def test_checkout_scan_records_provenance_and_success(tmp_path: Path) -> None:
    repo = tmp_path / "checkout"
    repo.mkdir()
    (repo / "requirements.txt").write_text("fastapi==0.136.1\n", encoding="utf-8")
    client = client_for(tmp_path)
    register(client, "checkout", checkout_path=str(repo))

    outcome = client.post("/sbom/checkout/ingest")

    assert outcome.status_code == 200
    assert outcome.json()["status"] == "ingested"
    detail = client.get(f"/sbom/snapshots/{outcome.json()['snapshot_id']}").json()
    assert detail["sources"][0]["path"] == "requirements.txt"
    assert detail["sources"][0]["sha256"]
    assert detail["entries"][0]["package_name"] == "fastapi"
    assert detail["entries"][0]["snapshot_at"] == detail["snapshot_at"]


def test_historical_import_is_idempotent(tmp_path: Path) -> None:
    client = client_for(tmp_path)
    payload = {
        "repo_slug": "legacy",
        "legacy_id": "017f8f04-84c2-4e58-8be1-33a4cb43c43f",
        "snapshot_at": "2026-01-02T03:04:05Z",
        "source": "state-hub:manual",
        "entries": [
            {
                "package_name": "legacy-package",
                "package_version": "1.0",
                "ecosystem": "other",
            }
        ],
    }

    first = client.post("/sbom/import/", json=payload)
    second = client.post("/sbom/import/", json=payload)

    assert first.status_code == 200
    assert first.json()["imported"] is True
    assert second.json()["imported"] is False
    assert len(client.get("/sbom/snapshots/?repo_slug=legacy").json()) == 1