tegwick
76e6eda086
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
280 lines
9.7 KiB
Python
280 lines
9.7 KiB
Python
"""Compatibility coverage for the reversible SBOM Nexus read facade."""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from api.config import settings
|
|
from api.services.sbom_nexus import SBOMNexusError
|
|
|
|
|
|
async def _create_repo(client) -> dict:
|
|
domain = await client.post(
|
|
"/domains/", json={"slug": "sbom-test", "name": "SBOM Test"}
|
|
)
|
|
assert domain.status_code == 201
|
|
repo = await client.post(
|
|
"/repos/",
|
|
json={
|
|
"slug": "testrepo",
|
|
"name": "Test Repo",
|
|
"domain_slug": "sbom-test",
|
|
"local_path": "/tmp/testrepo",
|
|
},
|
|
)
|
|
assert repo.status_code == 201
|
|
return repo.json()
|
|
|
|
|
|
def _entry(snapshot_id: str, nexus_repo_id: str) -> dict:
|
|
return {
|
|
"id": str(uuid.uuid4()),
|
|
"repo_id": nexus_repo_id,
|
|
"repo_slug": "testrepo",
|
|
"snapshot_id": snapshot_id,
|
|
"package_name": "fastapi",
|
|
"package_version": "0.115.0",
|
|
"ecosystem": "python",
|
|
"license_spdx": "MIT",
|
|
"is_direct": True,
|
|
"is_dev": False,
|
|
"snapshot_at": "2026-08-22T12:00:00+00:00",
|
|
"created_at": "2026-08-22T12:00:01+00:00",
|
|
"source_path": "uv.lock",
|
|
}
|
|
|
|
|
|
async def test_nexus_read_mode_preserves_state_hub_response_contract(client, monkeypatch):
|
|
repo = await _create_repo(client)
|
|
snapshot_id = str(uuid.uuid4())
|
|
nexus_repo_id = str(uuid.uuid4())
|
|
entry = _entry(snapshot_id, nexus_repo_id)
|
|
snapshot = {
|
|
"id": snapshot_id,
|
|
"repo_id": nexus_repo_id,
|
|
"repo_slug": "testrepo",
|
|
"snapshot_at": "2026-08-22T12:00:00+00:00",
|
|
"source": "state-hub:manual",
|
|
"legacy_id": snapshot_id,
|
|
"entry_count": 1,
|
|
"created_at": "2026-08-22T12:00:01+00:00",
|
|
"status": "imported",
|
|
}
|
|
|
|
async def fake_get(path: str, *, params=None):
|
|
if path == "/sbom/snapshots/":
|
|
assert params == {"repo_slug": "testrepo"}
|
|
return [snapshot]
|
|
if path == f"/sbom/snapshots/{snapshot_id}":
|
|
return {**snapshot, "entries": [entry]}
|
|
if path == "/sbom/":
|
|
assert params == {
|
|
"repo_slug": "testrepo",
|
|
"ecosystem": "python",
|
|
"is_direct": True,
|
|
}
|
|
return [entry]
|
|
if path == "/sbom/report/licences/":
|
|
return {
|
|
"groups": [
|
|
{
|
|
"license_spdx": "MIT",
|
|
"count": 1,
|
|
"repos": ["testrepo"],
|
|
"is_copyleft": False,
|
|
}
|
|
],
|
|
"copyleft_direct_count": 0,
|
|
"signal_qualification": "extra Nexus field",
|
|
}
|
|
if path == "/sbom/testrepo":
|
|
return {
|
|
"repo_slug": "testrepo",
|
|
"last_sbom_at": "2026-08-22T12:00:00+00:00",
|
|
"last_attempt_at": "2026-08-22T12:00:00+00:00",
|
|
"entry_count": 1,
|
|
"entries": [entry],
|
|
}
|
|
raise AssertionError(f"unexpected Nexus path {path}")
|
|
|
|
monkeypatch.setattr(settings, "sbom_nexus_read_mode", "nexus")
|
|
monkeypatch.setattr("api.routers.sbom.get_json", fake_get)
|
|
|
|
snapshots = await client.get("/sbom/snapshots/?repo_slug=testrepo")
|
|
detail = await client.get(f"/sbom/snapshots/{snapshot_id}")
|
|
entries = await client.get(
|
|
"/sbom/?repo_slug=testrepo&ecosystem=python&is_direct=true"
|
|
)
|
|
licences = await client.get("/sbom/report/licences/")
|
|
repo_view = await client.get("/sbom/testrepo")
|
|
|
|
assert snapshots.status_code == 200
|
|
assert detail.status_code == 200
|
|
assert entries.status_code == 200
|
|
assert licences.status_code == 200
|
|
assert repo_view.status_code == 200
|
|
assert snapshots.json()[0]["id"] == snapshot_id
|
|
assert snapshots.json()[0]["repo_id"] == repo["id"]
|
|
assert snapshots.json()[0]["source"] == "manual"
|
|
assert detail.json()["source"] == "manual"
|
|
assert detail.json()["entries"][0]["repo_id"] == repo["id"]
|
|
assert entries.json()[0]["repo_id"] == repo["id"]
|
|
assert "source_path" not in entries.json()[0]
|
|
assert set(licences.json()) == {"groups", "copyleft_direct_count"}
|
|
assert repo_view.json()["entries"][0]["repo_id"] == repo["id"]
|
|
assert set(repo_view.json()) == {
|
|
"repo_slug",
|
|
"last_sbom_at",
|
|
"entry_count",
|
|
"entries",
|
|
}
|
|
|
|
|
|
async def test_nexus_read_mode_does_not_move_ingest_write_authority(client, monkeypatch):
|
|
await _create_repo(client)
|
|
|
|
async def unexpected_get(*args, **kwargs):
|
|
raise AssertionError("POST /sbom/ingest/ must not call SBOM Nexus")
|
|
|
|
monkeypatch.setattr(settings, "sbom_nexus_read_mode", "nexus")
|
|
monkeypatch.setattr("api.routers.sbom.get_json", unexpected_get)
|
|
response = await client.post(
|
|
"/sbom/ingest/",
|
|
json={
|
|
"repo_slug": "testrepo",
|
|
"entries": [
|
|
{
|
|
"package_name": "fastapi",
|
|
"package_version": "0.115.0",
|
|
"ecosystem": "python",
|
|
"license_spdx": "MIT",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["ingested"] == 1
|
|
|
|
|
|
async def test_nexus_write_mode_moves_authority_and_updates_projection(client, monkeypatch):
|
|
repo = await _create_repo(client)
|
|
snapshot_id = str(uuid.uuid4())
|
|
observed = {}
|
|
|
|
async def fake_post(path: str, *, body):
|
|
observed["path"] = path
|
|
observed["body"] = body
|
|
return {
|
|
"repo_slug": "testrepo",
|
|
"snapshot_id": snapshot_id,
|
|
"ingested": 1,
|
|
"snapshot_at": "2026-08-22T13:00:00Z",
|
|
"status": "ingested",
|
|
}
|
|
|
|
monkeypatch.setattr(settings, "sbom_nexus_read_mode", "legacy")
|
|
monkeypatch.setattr(settings, "sbom_nexus_write_mode", "nexus")
|
|
monkeypatch.setattr("api.routers.sbom.post_json", fake_post)
|
|
response = await client.post(
|
|
"/sbom/ingest/",
|
|
json={
|
|
"repo_slug": "testrepo",
|
|
"entries": [
|
|
{
|
|
"package_name": "fastapi",
|
|
"package_version": "0.115.0",
|
|
"ecosystem": "python",
|
|
"license_spdx": "MIT",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"repo_slug": "testrepo",
|
|
"snapshot_id": snapshot_id,
|
|
"ingested": 1,
|
|
"snapshot_at": "2026-08-22T13:00:00Z",
|
|
}
|
|
assert observed["path"] == "/sbom/ingest/"
|
|
assert observed["body"]["entries"][0]["ecosystem"] == "python"
|
|
|
|
legacy_snapshots = await client.get("/sbom/snapshots/?repo_slug=testrepo")
|
|
projected_repo = await client.get("/repos/testrepo")
|
|
assert legacy_snapshots.json() == []
|
|
assert projected_repo.json()["id"] == repo["id"]
|
|
assert projected_repo.json()["last_sbom_at"] == "2026-08-22T13:00:00Z"
|
|
assert projected_repo.json()["sbom_source"] == "sbom-nexus"
|
|
|
|
|
|
async def test_nexus_failure_is_visible_and_does_not_fall_back(client, monkeypatch):
|
|
await _create_repo(client)
|
|
|
|
async def unavailable(*args, **kwargs):
|
|
raise SBOMNexusError(502, "SBOM Nexus is unavailable: ConnectError")
|
|
|
|
monkeypatch.setattr(settings, "sbom_nexus_read_mode", "nexus")
|
|
monkeypatch.setattr("api.routers.sbom.get_json", unavailable)
|
|
response = await client.get("/sbom/snapshots/")
|
|
|
|
assert response.status_code == 502
|
|
assert response.json()["detail"] == "SBOM Nexus is unavailable: ConnectError"
|
|
|
|
|
|
async def test_repo_reads_project_nexus_last_attempt_at(client, monkeypatch):
|
|
await _create_repo(client)
|
|
|
|
async def repositories(path: str, **kwargs):
|
|
assert path == "/repositories/"
|
|
return [
|
|
{
|
|
"slug": "testrepo",
|
|
"last_attempt_at": "2026-08-22T14:00:00Z",
|
|
}
|
|
]
|
|
|
|
monkeypatch.setattr(settings, "sbom_nexus_read_mode", "nexus")
|
|
monkeypatch.setattr("api.routers.repos.get_sbom_nexus_json", repositories)
|
|
|
|
collection = await client.get("/repos/")
|
|
detail = await client.get("/repos/testrepo")
|
|
|
|
assert collection.status_code == 200
|
|
assert detail.status_code == 200
|
|
assert collection.json()[0]["last_sbom_at"] == "2026-08-22T14:00:00Z"
|
|
assert collection.json()[0]["sbom_source"] == "sbom-nexus"
|
|
assert detail.json()["last_sbom_at"] == "2026-08-22T14:00:00Z"
|
|
assert detail.json()["sbom_source"] == "sbom-nexus"
|
|
|
|
|
|
async def test_state_summaries_use_nexus_metrics(client, monkeypatch):
|
|
async def sbom_metrics(path: str, **kwargs):
|
|
if path == "/sbom/snapshots/":
|
|
return [
|
|
{
|
|
"snapshot_at": "2026-08-22T12:00:00Z",
|
|
"entry_count": 3,
|
|
},
|
|
{
|
|
"snapshot_at": "2026-08-22T14:00:00Z",
|
|
"entry_count": 5,
|
|
},
|
|
]
|
|
if path == "/sbom/report/licences/":
|
|
return {"groups": [], "copyleft_direct_count": 7}
|
|
raise AssertionError(f"unexpected Nexus path {path}")
|
|
|
|
monkeypatch.setattr(settings, "sbom_nexus_read_mode", "nexus")
|
|
monkeypatch.setattr("api.routers.state.get_sbom_nexus_json", sbom_metrics)
|
|
monkeypatch.setattr("api.services.summary_cache.get_sbom_nexus_json", sbom_metrics)
|
|
|
|
summary = await client.get("/state/summary", params={"refresh": "true"})
|
|
overview = await client.get("/state/overview", params={"refresh": "true"})
|
|
|
|
assert summary.status_code == 200
|
|
assert overview.status_code == 200
|
|
assert summary.json()["licence_risk_count"] == 7
|
|
assert overview.json()["licence_risk_count"] == 7
|
|
assert overview.json()["sbom_snapshot_count"] == 2
|
|
assert overview.json()["sbom_package_total"] == 8
|