Automate policy source freshness and inventory

tests/test_source_inventory.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).parents[1]
+sys.path.insert(0, str(ROOT / "tools"))
+from source_inventory import check, refresh
+
+
+REVISION = "a" * 40
+
+
+def _fixture(root: Path) -> tuple[Path, Path, Path, Path]:
+    source = root / "docs/adr/ADR-0001.md"
+    source.parent.mkdir(parents=True)
+    source.write_text("# ADR-0001\n", encoding="utf-8")
+    config = root / "source-inventory.config.json"
+    config.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "repositories": {
+                    "policy-nexus": {
+                        "local": True,
+                        "selectors": ["docs/adr/*.md"],
+                    }
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    publication = root / "publication.json"
+    publication.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "documents": [
+                    {
+                        "source_repo": "policy-nexus",
+                        "source_path": "docs/adr/ADR-0001.md",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    revisions = {"policy-nexus": REVISION}
+    digest = hashlib.sha256(
+        json.dumps(revisions, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest()
+    lock = root / "source-lock.json"
+    lock.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "source_set_digest": digest,
+                "repositories": {"policy-nexus": {"revision": REVISION}},
+            }
+        ),
+        encoding="utf-8",
+    )
+    return config, root / "source-inventory.json", publication, lock
+
+
+class SourceInventoryTest(unittest.TestCase):
+    def test_refresh_and_check_bind_publication_to_inventory(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config, inventory, publication, lock = _fixture(root)
+            refresh(
+                config,
+                inventory,
+                publication,
+                policy_root=root,
+                source_root=root.parent,
+            )
+            report = check(
+                config,
+                inventory,
+                publication,
+                policy_root=root,
+                source_root=root.parent,
+                lock_path=lock,
+            )
+            self.assertEqual(1, report["dispositions"]["published"])
+            self.assertEqual(REVISION, report["repositories"][0]["revision"])
+
+    def test_check_rejects_unreviewed_new_source(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config, inventory, publication, lock = _fixture(root)
+            refresh(
+                config,
+                inventory,
+                publication,
+                policy_root=root,
+                source_root=root.parent,
+            )
+            (root / "docs/adr/ADR-0002.md").write_text("# New\n", encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "unreviewed sources"):
+                check(
+                    config,
+                    inventory,
+                    publication,
+                    policy_root=root,
+                    source_root=root.parent,
+                    lock_path=lock,
+                )
+
+    def test_check_rejects_published_disposition_drift(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config, inventory, publication, lock = _fixture(root)
+            refresh(
+                config,
+                inventory,
+                publication,
+                policy_root=root,
+                source_root=root.parent,
+            )
+            value = json.loads(inventory.read_text(encoding="utf-8"))
+            value["sources"][0]["disposition"] = "metadata-pending"
+            inventory.write_text(json.dumps(value), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "must exactly match"):
+                check(
+                    config,
+                    inventory,
+                    publication,
+                    policy_root=root,
+                    source_root=root.parent,
+                    lock_path=lock,
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()