feat: prepare postgres sbom cutover

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834

tests/test_import_state_hub.py

@@ -0,0 +1,103 @@
+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