Add dry-run apply activation plan

tests/test_activation.py

@@ -0,0 +1,180 @@
+import contextlib
+import io
+import tempfile
+import unittest
+from pathlib import Path
+
+from ampg.activation import activation_steps, blocked_steps
+from ampg.cli import main
+from ampg.config import load_config
+from ampg.platforms import platform_by_name
+from ampg.status import DaemonProbe
+
+
+def _probe(*, installed: bool = False, running: bool = False):
+    def probe(adapter):
+        executable_path = f"/usr/bin/{adapter.executable}" if installed else None
+        return DaemonProbe(
+            installed=installed,
+            running=running,
+            executable_path=executable_path,
+        )
+
+    return probe
+
+
+def _write_config(root: Path, protocol_block: str) -> Path:
+    source = root / "site"
+    source.mkdir()
+    (source / "index.html").write_text("<h1>Hello</h1>", encoding="utf-8")
+    config_path = root / "gateway.toml"
+    config_path.write_text(
+        f"""
+[[site]]
+id = "example"
+domain = "example.test"
+
+[site.source]
+kind = "static-html"
+path = "./site"
+
+[site.outputs]
+root = "./out"
+plan_root = "./plan"
+
+{protocol_block}
+""",
+        encoding="utf-8",
+    )
+    return config_path
+
+
+class ActivationTest(unittest.TestCase):
+    def test_activation_plan_is_ready_when_output_and_daemon_policy_are_ready(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            config_path = _write_config(
+                root,
+                """
+[site.protocols.i2p]
+enabled = true
+renderer = "privacy-html"
+daemon = "i2pd"
+daemon_policy = "auto"
+local_port = 19081
+""",
+            )
+            output = root / "out/i2p"
+            output.mkdir(parents=True)
+            (output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
+            config = load_config(config_path)
+
+            steps = activation_steps(
+                config,
+                platform_provider=platform_by_name("android-termux"),
+                daemon_probe=_probe(installed=False, running=False),
+            )
+
+        self.assertEqual([], blocked_steps(steps))
+        by_stage = {step.stage: step for step in steps if step.stage != "artifact"}
+        self.assertEqual("ready", by_stage["output"].status)
+        self.assertEqual("manage-owned", by_stage["daemon"].action)
+        self.assertEqual("planned", by_stage["health"].status)
+        self.assertTrue(any(step.stage == "artifact" for step in steps))
+
+    def test_missing_output_blocks_activation(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            config_path = _write_config(
+                Path(tmp),
+                """
+[site.protocols.tor]
+enabled = true
+renderer = "privacy-html"
+daemon = "tor"
+daemon_policy = "auto"
+""",
+            )
+            config = load_config(config_path)
+
+            steps = activation_steps(
+                config,
+                platform_provider=platform_by_name("linux-user"),
+                daemon_probe=_probe(installed=False, running=False),
+            )
+
+        blocked = blocked_steps(steps)
+        self.assertTrue(any(step.stage == "output" for step in blocked))
+        self.assertTrue(any(step.stage == "health" for step in blocked))
+
+    def test_apply_cli_requires_dry_run(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            config_path = _write_config(
+                Path(tmp),
+                """
+[site.protocols.gemini]
+enabled = true
+renderer = "gemtext"
+daemon = "agate"
+daemon_policy = "external"
+""",
+            )
+
+            stderr = io.StringIO()
+            with contextlib.redirect_stderr(stderr):
+                status = main(["--config", str(config_path), "apply"])
+
+        self.assertEqual(1, status)
+        self.assertIn("live apply is not implemented", stderr.getvalue())
+
+    def test_apply_dry_run_can_write_review_artifacts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            config_path = _write_config(
+                root,
+                """
+[site.protocols.gemini]
+enabled = true
+renderer = "gemtext"
+daemon = "agate"
+daemon_policy = "external"
+port = 1965
+""",
+            )
+            output = root / "out/gemini"
+            output.mkdir(parents=True)
+            (output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
+
+            stdout = io.StringIO()
+            with contextlib.redirect_stdout(stdout):
+                status = main(
+                    [
+                        "--config",
+                        str(config_path),
+                        "apply",
+                        "--dry-run",
+                        "--write-artifacts",
+                        "--platform",
+                        "unknown",
+                    ]
+                )
+
+            self.assertEqual(0, status)
+            output_text = stdout.getvalue()
+            self.assertIn("AMPG_APPLY_ARTIFACT", output_text)
+            self.assertIn("AMPG_APPLY_STEP", output_text)
+            self.assertIn("blocked=0", output_text)
+            artifact_line = next(
+                line for line in output_text.splitlines() if line.startswith("AMPG_APPLY_ARTIFACT")
+            )
+            artifact_path = Path(
+                next(
+                    part.removeprefix("path=")
+                    for part in artifact_line.split()
+                    if part.startswith("path=")
+                )
+            )
+            self.assertTrue(artifact_path.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()