tegwick
ba535e1f8f
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sbom_nexus import importer
|
|
|
|
|
|
def test_import_reconciles_history_licences_and_is_idempotent(monkeypatch) -> None:
|
|
target_snapshots: list[dict[str, Any]] = []
|
|
groups = [
|
|
{
|
|
"license_spdx": "MIT",
|
|
"count": 1,
|
|
"repos": ["demo"],
|
|
"is_copyleft": False,
|
|
},
|
|
{
|
|
"license_spdx": None,
|
|
"count": 2,
|
|
"repos": ["zeta", "demo"],
|
|
"is_copyleft": False,
|
|
},
|
|
]
|
|
|
|
def fake_request(
|
|
base_url: str,
|
|
path: str,
|
|
*,
|
|
method: str = "GET",
|
|
body: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
if base_url == "source":
|
|
if path == "/repos/":
|
|
return [
|
|
{
|
|
"id": "repo-id",
|
|
"slug": "demo",
|
|
"status": "active",
|
|
"local_path": "/repos/demo",
|
|
}
|
|
]
|
|
if path == "/sbom/snapshots/":
|
|
return [
|
|
{
|
|
"id": "snapshot-id",
|
|
"repo_id": "repo-id",
|
|
"snapshot_at": "2026-01-02T03:04:05Z",
|
|
"source": "manual",
|
|
"entry_count": 1,
|
|
}
|
|
]
|
|
if path == "/sbom/snapshots/snapshot-id":
|
|
return {
|
|
"entries": [
|
|
{
|
|
"package_name": "example",
|
|
"package_version": "1.0",
|
|
"ecosystem": "other",
|
|
"license_spdx": "MIT",
|
|
"is_direct": True,
|
|
"is_dev": False,
|
|
}
|
|
]
|
|
}
|
|
if path == "/sbom/report/licences/":
|
|
return {"groups": list(reversed(groups)), "copyleft_direct_count": 0}
|
|
if base_url == "target":
|
|
if method == "PUT":
|
|
return {"slug": "demo"}
|
|
if method == "POST" and path == "/sbom/import/":
|
|
assert body is not None
|
|
already_present = any(
|
|
snapshot["legacy_id"] == body["legacy_id"]
|
|
for snapshot in target_snapshots
|
|
)
|
|
if not already_present:
|
|
target_snapshots.append(
|
|
{
|
|
"legacy_id": body["legacy_id"],
|
|
"repo_slug": body["repo_slug"],
|
|
"snapshot_at": body["snapshot_at"],
|
|
"entry_count": len(body["entries"]),
|
|
}
|
|
)
|
|
return {"imported": not already_present}
|
|
if path == "/sbom/snapshots/":
|
|
return target_snapshots
|
|
if path == "/sbom/report/licences/":
|
|
return {"groups": groups, "copyleft_direct_count": 0}
|
|
raise AssertionError((base_url, method, path))
|
|
|
|
monkeypatch.setattr(importer, "request_json", fake_request)
|
|
|
|
first = importer.import_history("source", "target", dry_run=False)
|
|
second = importer.import_history("source", "target", dry_run=False)
|
|
|
|
assert first["ok"] is True
|
|
assert first["results"] == {"imported": 1}
|
|
assert first["snapshot_reconciliation"]["matched_snapshot_count"] == 1
|
|
assert first["licence_reconciliation"]["groups_match"] is True
|
|
assert second["ok"] is True
|
|
assert second["results"] == {"already_present": 1}
|
|
assert len(target_snapshots) == 1
|