tegwick
9751927d38
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 38s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118
256 lines
8.5 KiB
Python
256 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from sbom_nexus import api
|
|
from sbom_nexus.api import create_app
|
|
from sbom_nexus.source_fetch import ControlledSource
|
|
|
|
|
|
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,
|
|
source_ref: dict | None = None,
|
|
) -> 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,
|
|
"source_ref": source_ref,
|
|
},
|
|
)
|
|
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",
|
|
}
|
|
assert client.get("/state/live").json() == {"status": "ok"}
|
|
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_ingest_operation_key_replays_without_a_second_snapshot(tmp_path: Path) -> None:
|
|
repo = tmp_path / "idempotent"
|
|
repo.mkdir()
|
|
(repo / "requirements.txt").write_text("fastapi==0.136.1\n", encoding="utf-8")
|
|
client = client_for(tmp_path)
|
|
register(client, "idempotent", checkout_path=str(repo))
|
|
headers = {
|
|
"Idempotency-Key": "activity-operation-1",
|
|
"X-Activity-Core-Operation-ID": "activity-operation-1",
|
|
}
|
|
|
|
first = client.post("/sbom/idempotent/ingest", headers=headers)
|
|
second = client.post("/sbom/idempotent/ingest", headers=headers)
|
|
|
|
assert first.status_code == 200
|
|
assert second.json() == first.json()
|
|
assert len(client.get("/sbom/snapshots/?repo_slug=idempotent").json()) == 1
|
|
conflict = client.post(
|
|
"/sbom/idempotent/skip",
|
|
headers=headers,
|
|
json={"reason": "no-checkout"},
|
|
)
|
|
assert conflict.status_code == 409
|
|
|
|
|
|
def test_skip_operation_key_replays_original_outcome(tmp_path: Path) -> None:
|
|
client = client_for(tmp_path)
|
|
register(client, "skip-replay")
|
|
headers = {"Idempotency-Key": "skip-operation-1"}
|
|
|
|
first = client.post(
|
|
"/sbom/skip-replay/skip",
|
|
headers=headers,
|
|
json={"reason": "source-unavailable", "detail": "not projected"},
|
|
)
|
|
second = client.post(
|
|
"/sbom/skip-replay/skip",
|
|
headers=headers,
|
|
json={"reason": "source-unavailable", "detail": "not projected"},
|
|
)
|
|
|
|
assert first.status_code == 200
|
|
assert second.json() == first.json()
|
|
assert len(client.get("/sbom/snapshots/?repo_slug=skip-replay").json()) == 1
|
|
|
|
|
|
def test_controlled_source_records_explicit_revision_and_archive_provenance(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
extracted = tmp_path / "extracted"
|
|
extracted.mkdir()
|
|
(extracted / "requirements.txt").write_text(
|
|
"fastapi==0.136.1\n", encoding="utf-8"
|
|
)
|
|
revision = "a" * 40
|
|
source_ref = {
|
|
"kind": "forgejo-archive-v1",
|
|
"repository": "coulomb/controlled",
|
|
"revision": revision,
|
|
"observed_ref": "refs/heads/main",
|
|
"observed_at": "2026-08-22T20:00:00Z",
|
|
}
|
|
|
|
@contextmanager
|
|
def fake_fetch(repo_slug: str, selected: dict):
|
|
assert repo_slug == "controlled"
|
|
assert selected["revision"] == revision
|
|
yield ControlledSource(
|
|
root=extracted,
|
|
provenance={**selected, "archive_sha256": "b" * 64, "archive_bytes": 42},
|
|
)
|
|
|
|
monkeypatch.setattr(api, "fetch_controlled_source", fake_fetch)
|
|
client = client_for(tmp_path)
|
|
register(client, "controlled", source_ref=source_ref)
|
|
|
|
result = client.post(
|
|
"/sbom/controlled/ingest",
|
|
json={"source_ref": source_ref},
|
|
headers={"Idempotency-Key": "controlled-operation-1"},
|
|
)
|
|
|
|
assert result.status_code == 200
|
|
assert result.json()["source_revision"] == revision
|
|
detail = client.get(f"/sbom/snapshots/{result.json()['snapshot_id']}").json()
|
|
assert detail["source"] == "forgejo-archive-v1"
|
|
assert detail["source_provenance"]["archive_sha256"] == "b" * 64
|
|
assert detail["sources"][0]["path"] == "requirements.txt"
|
|
|
|
|
|
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
|