mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
Add state stage deploy apply
This commit is contained in:
jwinterm
@@ -0,0 +1,217 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user