feat: gate infrastructure changes on provider inventory

tests/test_deployment_installer.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 from contextlib import redirect_stderr, redirect_stdout
+from datetime import UTC, datetime, timedelta
 import io
 import json
+import os
 from pathlib import Path
 import stat
 import subprocess
@@ -36,6 +38,7 @@ import govoplan_deploy.cli as deployment_cli  # noqa: E402
 from govoplan_deploy.capabilities import (  # noqa: E402
     capability_change_impacts,
     infrastructure_capability_document,
+    infrastructure_dependency_inventory_from_mapping,
 )
 from govoplan_deploy.cluster_evidence import (  # noqa: E402
     collect_kubernetes_evidence,
@@ -87,6 +90,41 @@ def _kubernetes_test_deployment(component: str, replicas: int) -> dict:
     }
 
 
+def _dependency_inventory(
+    installation_id: str,
+    *,
+    generated_at: datetime | None = None,
+) -> dict:
+    return {
+        "schema_version": 1,
+        "installation_id": installation_id,
+        "generated_at": (generated_at or datetime.now(UTC)).isoformat(),
+        "complete": True,
+        "inspected_capability_ids": ["coordination.redis", "mail.smtp"],
+        "providers": [
+            {
+                "module_id": "mail",
+                "state": "complete",
+                "capability_ids": ["mail.smtp"],
+                "dependency_count": 1,
+            }
+        ],
+        "dependencies": [
+            {
+                "capability_id": "mail.smtp",
+                "module_id": "mail",
+                "dependency_type": "smtp_endpoint",
+                "dependency_ref": "endpoint:17",
+                "state": "active",
+                "scope": "system",
+                "summary": "Persisted SMTP endpoint has one credential binding.",
+                "metrics": {"credential_binding_count": 1},
+                "required_action": "Rebind or migrate this SMTP endpoint.",
+            }
+        ],
+    }
+
+
 class DeploymentInstallerTests(unittest.TestCase):
     def test_kubernetes_evidence_requires_two_node_spread_and_safe_runtime(
         self,
@@ -1034,6 +1072,28 @@ class DeploymentInstallerTests(unittest.TestCase):
         self.assertNotIn("old-secret", impacts["database.postgresql"].detail)
         self.assertNotIn("new-secret", impacts["database.postgresql"].detail)
 
+    def test_capability_impact_includes_provider_dependency_evidence(self) -> None:
+        previous_spec = default_spec(mail_mode="test-mail", module_set="full")
+        desired_spec = default_spec(mail_mode="disabled", module_set="full")
+        inventory = infrastructure_dependency_inventory_from_mapping(
+            _dependency_inventory(previous_spec.installation_id)
+        )
+
+        impacts = {
+            item.capability_id: item
+            for item in capability_change_impacts(
+                infrastructure_capability_document(previous_spec, {}),
+                infrastructure_capability_document(desired_spec, {}),
+                dependency_inventory=inventory,
+            )
+        }
+
+        mail = impacts["mail.smtp"]
+        self.assertTrue(mail.inventory_inspected)
+        self.assertEqual("endpoint:17", mail.actual_dependencies[0].dependency_ref)
+        self.assertIn("mail:endpoint:17", mail.detail)
+        self.assertIn("Rebind or migrate", mail.required_action)
+
     def test_replica_counts_drive_compose_and_load_balancer_discovery(self) -> None:
         spec = default_spec(
             storage_mode="garage",
@@ -1221,6 +1281,65 @@ class DeploymentInstallerTests(unittest.TestCase):
                     for check in second_plan.checks
                 )
             )
+            self.assertTrue(second_plan.blocked)
+            self.assertTrue(
+                any(
+                    check.id == "capability.dependency_inventory.missing"
+                    and check.level == "error"
+                    for check in second_plan.checks
+                )
+            )
+
+            atomic_write(
+                paths.dependency_inventory,
+                canonical_json(_dependency_inventory(second_spec.installation_id)),
+                mode=0o600,
+            )
+            evidenced_plan = build_plan(
+                second_spec,
+                paths,
+                include_host_checks=False,
+            )
+
+            self.assertFalse(
+                any(
+                    check.level == "error"
+                    and check.id.startswith("capability.dependency_inventory.")
+                    for check in evidenced_plan.checks
+                )
+            )
+            self.assertEqual(
+                "endpoint:17",
+                {
+                    item.capability_id: item
+                    for item in evidenced_plan.capability_impacts
+                }["mail.smtp"].actual_dependencies[0].dependency_ref,
+            )
+            self.assertTrue(
+                any(
+                    check.id == "capability.dependency_inventory.current"
+                    and check.level == "ok"
+                    for check in evidenced_plan.checks
+                )
+            )
+
+            stale = _dependency_inventory(
+                second_spec.installation_id,
+                generated_at=datetime.now(UTC) - timedelta(minutes=6),
+            )
+            atomic_write(
+                paths.dependency_inventory,
+                canonical_json(stale),
+                mode=0o600,
+            )
+            stale_plan = build_plan(second_spec, paths, include_host_checks=False)
+            self.assertTrue(stale_plan.blocked)
+            self.assertTrue(
+                any(
+                    check.id == "capability.dependency_inventory.stale"
+                    for check in stale_plan.checks
+                )
+            )
 
     def test_secret_change_is_planned_without_exposing_secret_values(self) -> None:
         with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory:
@@ -1330,6 +1449,54 @@ class DeploymentInstallerTests(unittest.TestCase):
                 )[0],
             )
 
+    def test_cli_collects_bounded_private_dependency_inventory(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory:
+            root = Path(directory) / "installation"
+            self.assertEqual(
+                0,
+                run_cli(
+                    [
+                        "init",
+                        "--non-interactive",
+                        "--directory",
+                        str(root),
+                    ]
+                )[0],
+            )
+            payload = _dependency_inventory("govoplan-local")
+            response = MagicMock()
+            response.__enter__.return_value = response
+            response.geturl.return_value = "https://ops.example.test/inventory"
+            response.read.return_value = json.dumps(payload).encode("utf-8")
+            fetch = MagicMock(return_value=response)
+
+            with (
+                patch.dict(os.environ, {"TEST_OPS_KEY": "secret-api-key"}),
+                patch.object(deployment_cli, "urlopen", fetch),
+            ):
+                result, stdout, stderr = run_cli(
+                    [
+                        "collect-infrastructure-inventory",
+                        "--directory",
+                        str(root),
+                        "--ops-url",
+                        "https://ops.example.test/inventory",
+                        "--api-key-env",
+                        "TEST_OPS_KEY",
+                    ]
+                )
+
+            self.assertEqual(0, result, stderr)
+            self.assertIn("1 record(s)", stdout)
+            evidence_path = root / "infrastructure-dependency-inventory.json"
+            self.assertEqual(0o600, stat.S_IMODE(evidence_path.stat().st_mode))
+            self.assertNotIn(
+                "secret-api-key",
+                evidence_path.read_text(encoding="utf-8"),
+            )
+            request = fetch.call_args.args[0]
+            self.assertEqual("secret-api-key", request.get_header("X-api-key"))
+
     def test_cli_requires_external_url_when_switching_from_managed(self) -> None:
         with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory:
             root = Path(directory) / "installation"