policy-nexus/tests/test_source_inventory.py

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()