adaptive-multi-protocol-gateway/tests/test_deploy_apply.py

import contextlib
import io
import tempfile
import unittest
from pathlib import Path

from ampg.cli import main


def _source(root: Path) -> Path:
    source = root / "site"
    source.mkdir()
    (source / "index.html").write_text("<h1>Hello</h1>", encoding="utf-8")
    return source


def _run_cli(args: list[str]) -> tuple[int, str, str]:
    stdout = io.StringIO()
    stderr = io.StringIO()
    with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
        status = main(args)
    return status, stdout.getvalue(), stderr.getvalue()


def _init_i2p_config(root: Path) -> Path:
    config_path = root / "gateway.toml"
    status, _, error = _run_cli(
        [
            "--config",
            str(config_path),
            "init",
            "site",
            "example",
            "--domain",
            "example.test",
            "--source",
            str(_source(root)),
            "--protocol",
            "i2p",
        ]
    )
    if status != 0:
        raise AssertionError(error)
    output = root / "dist/example/i2p"
    output.mkdir(parents=True)
    (output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
    return config_path


def _write_and_approve_artifacts(config_path: Path) -> None:
    status, _, error = _run_cli(
        [
            "--config",
            str(config_path),
            "apply",
            "--dry-run",
            "--write-artifacts",
            "--profile",
            "mobile-i2p",
        ]
    )
    if status != 0:
        raise AssertionError(error)
    status, _, error = _run_cli(
        [
            "--config",
            str(config_path),
            "approvals",
            "approve",
            "--profile",
            "mobile-i2p",
            "--all",
        ]
    )
    if status != 0:
        raise AssertionError(error)


class DeployApplyTest(unittest.TestCase):
    def test_deploy_apply_state_blocks_unapproved_artifacts(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _init_i2p_config(root)

            status, output, error = _run_cli(
                [
                    "--config",
                    str(config_path),
                    "deploy",
                    "apply",
                    "--stage",
                    "state",
                    "--dry-run",
                    "--profile",
                    "mobile-i2p",
                ]
            )

        self.assertEqual(1, status)
        self.assertEqual("", error)
        self.assertIn("AMPG_DEPLOY_STATE", output)
        self.assertIn("status=blocked", output)
        self.assertIn("artifact is missing", output)

    def test_deploy_apply_state_dry_run_after_approval(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _init_i2p_config(root)
            _write_and_approve_artifacts(config_path)

            status, output, error = _run_cli(
                [
                    "--config",
                    str(config_path),
                    "deploy",
                    "apply",
                    "--stage",
                    "state",
                    "--dry-run",
                    "--profile",
                    "mobile-i2p",
                ]
            )

        self.assertEqual(0, status, error)
        self.assertIn("AMPG_DEPLOY_STATE", output)
        self.assertIn("status=planned", output)
        self.assertIn("AMPG_DEPLOY_APPLY_SUMMARY stage=state mode=dry-run", output)
        self.assertIn("blocked=0", output)

    def test_deploy_apply_state_live_requires_yes(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _init_i2p_config(root)
            _write_and_approve_artifacts(config_path)

            status, _, error = _run_cli(
                [
                    "--config",
                    str(config_path),
                    "deploy",
                    "apply",
                    "--stage",
                    "state",
                    "--profile",
                    "mobile-i2p",
                ]
            )

        self.assertEqual(1, status)
        self.assertIn("requires --yes", error)

    def test_deploy_apply_state_blocks_stale_approved_artifacts(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _init_i2p_config(root)
            _write_and_approve_artifacts(config_path)
            artifact = (
                root / "dist/ampg-plan/example/i2p/managed/android-termux/i2pd-tunnels.conf"
            )
            artifact.write_text(
                artifact.read_text(encoding="utf-8") + "# edited\n",
                encoding="utf-8",
            )

            status, output, error = _run_cli(
                [
                    "--config",
                    str(config_path),
                    "deploy",
                    "apply",
                    "--stage",
                    "state",
                    "--dry-run",
                    "--profile",
                    "mobile-i2p",
                ]
            )

        self.assertEqual(1, status)
        self.assertEqual("", error)
        self.assertIn("status=blocked", output)
        self.assertIn("artifact differs from current generator output", output)

    def test_deploy_apply_state_live_copies_approved_artifacts(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _init_i2p_config(root)
            _write_and_approve_artifacts(config_path)

            status, output, error = _run_cli(
                [
                    "--config",
                    str(config_path),
                    "deploy",
                    "apply",
                    "--stage",
                    "state",
                    "--profile",
                    "mobile-i2p",
                    "--yes",
                ]
            )

            daemon_config = root / ".ampg/state/example/i2p/i2pd-tunnels.conf"
            loopback_config = root / ".ampg/state/example/i2p/nginx-loopback.conf"
            daemon_config_exists = daemon_config.exists()
            loopback_config_exists = loopback_config.exists()

        self.assertEqual(0, status, error)
        self.assertIn("status=written", output)
        self.assertTrue(daemon_config_exists)
        self.assertTrue(loopback_config_exists)


if __name__ == "__main__":
    unittest.main()