github-such-software/adaptive-multi-protocol-gateway
1
0
Fork 0
mirror of https://github.com/Such-Software/adaptive-multi-protocol-gateway.git synced 2026-10-17 15:41:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add address capture deploy apply stage

Browse Source
This commit is contained in:
jwinterm
2026-07-07 05:56:47 -04:00
parent c3d44a5e5b
commit 8f4724089b
12 changed files with 256 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_deploy_apply.py
+87
View File
@@ -4,6 +4,7 @@ import tempfile
import unittest
from pathlib import Path
from ampg.addresses import address_registry_path
from ampg.apply import CommandRunResult, apply_start, apply_state, apply_supervisor
from ampg.cli import main
from ampg.config import load_config
@@ -148,6 +149,13 @@ def _run_termux_supervisor_apply(config_path: Path, root: Path) -> None:
raise AssertionError(blocked[0].message)
def _write_i2p_address_file(root: Path, value: str = "captureexample.b32.i2p") -> Path:
path = root / ".ampg/state/example/i2p/b32.txt"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(value + "\n", encoding="utf-8")
return path
class DeployApplyTest(unittest.TestCase):
def test_deploy_apply_state_blocks_unapproved_artifacts(self):
with tempfile.TemporaryDirectory() as tmp:
@@ -479,6 +487,85 @@ class DeployApplyTest(unittest.TestCase):
self.assertIn(("sv-enable", "ampg-example-i2p-i2pd"), commands)
self.assertIn(("sv-enable", "ampg-example-i2p-nginx-loopback"), commands)
def test_deploy_apply_addresses_blocks_when_address_is_missing(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",
"addresses",
"--dry-run",
"--profile",
"mobile-i2p",
]
)
self.assertEqual(1, status)
self.assertEqual("", error)
self.assertIn("AMPG_DEPLOY_ADDRESS", output)
self.assertIn("status=blocked", output)
self.assertIn("no address file found", output)
def test_deploy_apply_addresses_dry_run_does_not_write_registry(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
config_path = _init_i2p_config(root)
_write_i2p_address_file(root)
config = load_config(config_path)
status, output, error = _run_cli(
[
"--config",
str(config_path),
"deploy",
"apply",
"--stage",
"addresses",
"--dry-run",
"--profile",
"mobile-i2p",
]
)
registry_exists = address_registry_path(config).exists()
self.assertEqual(0, status, error)
self.assertIn("AMPG_DEPLOY_ADDRESS", output)
self.assertIn("status=planned", output)
self.assertIn("http://captureexample.b32.i2p/", output)
self.assertFalse(registry_exists)
def test_deploy_apply_addresses_live_writes_registry(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
config_path = _init_i2p_config(root)
_write_i2p_address_file(root)
config = load_config(config_path)
status, output, error = _run_cli(
[
"--config",
str(config_path),
"deploy",
"apply",
"--stage",
"addresses",
"--profile",
"mobile-i2p",
"--yes",
]
)
registry_text = address_registry_path(config).read_text(encoding="utf-8")
self.assertEqual(0, status, error)
self.assertIn("status=written", output)
self.assertIn("http://captureexample.b32.i2p/", registry_text)
if __name__ == "__main__":
unittest.main()