mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
778 lines
26 KiB
Python
778 lines
26 KiB
Python
import contextlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from ampg.addresses import address_registry_path
|
|
from ampg.apply import (
|
|
CommandRunResult,
|
|
apply_health,
|
|
apply_packages,
|
|
apply_start,
|
|
apply_state,
|
|
apply_supervisor,
|
|
)
|
|
from ampg.cli import main
|
|
from ampg.config import load_config
|
|
from ampg.platforms import PlatformProvider
|
|
from ampg.status import DaemonProbe
|
|
|
|
|
|
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, *, platform: str | None = None) -> None:
|
|
apply_args = [
|
|
"--config",
|
|
str(config_path),
|
|
"apply",
|
|
"--dry-run",
|
|
"--write-artifacts",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
approve_args = [
|
|
"--config",
|
|
str(config_path),
|
|
"approvals",
|
|
"approve",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
"--all",
|
|
]
|
|
if platform is not None:
|
|
apply_args.extend(("--platform", platform))
|
|
approve_args.extend(("--platform", platform))
|
|
status, _, error = _run_cli(apply_args)
|
|
if status != 0:
|
|
raise AssertionError(error)
|
|
status, _, error = _run_cli(approve_args)
|
|
if status != 0:
|
|
raise AssertionError(error)
|
|
|
|
|
|
def _missing_probe(adapter):
|
|
return DaemonProbe(
|
|
installed=False,
|
|
running=False,
|
|
executable_path=None,
|
|
)
|
|
|
|
|
|
def _linux_user_provider(root: Path) -> PlatformProvider:
|
|
return PlatformProvider(
|
|
name="linux-user",
|
|
process_supervisor="user-process",
|
|
can_manage_daemons=True,
|
|
can_write_system_config=False,
|
|
state_root=root / "supervisor-root",
|
|
)
|
|
|
|
|
|
def _termux_provider(root: Path) -> PlatformProvider:
|
|
return PlatformProvider(
|
|
name="android-termux",
|
|
process_supervisor="termux-services",
|
|
can_manage_daemons=True,
|
|
can_write_system_config=False,
|
|
state_root=root / "termux/var/lib/ampg",
|
|
)
|
|
|
|
|
|
def _run_state_apply(config_path: Path, root: Path) -> None:
|
|
config = load_config(config_path)
|
|
results = apply_state(
|
|
config,
|
|
dry_run=False,
|
|
platform_provider=_linux_user_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
blocked = [result for result in results if result.status == "blocked"]
|
|
if blocked:
|
|
raise AssertionError(blocked[0].message)
|
|
|
|
|
|
def _run_termux_state_apply(config_path: Path, root: Path) -> None:
|
|
config = load_config(config_path)
|
|
results = apply_state(
|
|
config,
|
|
dry_run=False,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
blocked = [result for result in results if result.status == "blocked"]
|
|
if blocked:
|
|
raise AssertionError(blocked[0].message)
|
|
|
|
|
|
def _run_termux_supervisor_apply(config_path: Path, root: Path) -> None:
|
|
config = load_config(config_path)
|
|
results = apply_supervisor(
|
|
config,
|
|
dry_run=False,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
blocked = [result for result in results if result.status == "blocked"]
|
|
if blocked:
|
|
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_packages_dry_run_plans_termux_installs(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
config = load_config(config_path)
|
|
|
|
results = apply_packages(
|
|
config,
|
|
dry_run=True,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
|
|
self.assertTrue(all(result.status == "planned" for result in results))
|
|
self.assertIn(("pkg", "install", "i2pd"), [result.command for result in results])
|
|
self.assertIn(("pkg", "install", "nginx"), [result.command for result in results])
|
|
|
|
def test_deploy_apply_packages_live_invokes_runner(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
config = load_config(config_path)
|
|
commands: list[tuple[str, ...]] = []
|
|
|
|
def runner(command: tuple[str, ...]) -> CommandRunResult:
|
|
commands.append(command)
|
|
return CommandRunResult(return_code=0)
|
|
|
|
results = apply_packages(
|
|
config,
|
|
dry_run=False,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
command_runner=runner,
|
|
)
|
|
|
|
self.assertTrue(all(result.status == "installed" for result in results))
|
|
self.assertIn(("pkg", "install", "i2pd"), commands)
|
|
self.assertIn(("pkg", "install", "nginx"), commands)
|
|
|
|
def test_deploy_apply_packages_blocks_manual_linux_user_install(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
config = load_config(config_path)
|
|
|
|
results = apply_packages(
|
|
config,
|
|
dry_run=True,
|
|
platform_provider=_linux_user_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
|
|
self.assertTrue(any(result.status == "blocked" for result in results))
|
|
self.assertTrue(
|
|
any("automatic package install command" in result.message for result in results)
|
|
)
|
|
|
|
def test_deploy_apply_packages_cli_prints_summary(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
config_path.write_text(
|
|
config_path.read_text(encoding="utf-8").replace(
|
|
'daemon_policy = "auto"',
|
|
'daemon_policy = "manage"',
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"packages",
|
|
"--dry-run",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(0, status, error)
|
|
self.assertIn("AMPG_DEPLOY_PACKAGE", output)
|
|
self.assertIn("AMPG_DEPLOY_APPLY_SUMMARY stage=packages mode=dry-run", output)
|
|
|
|
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)
|
|
|
|
def test_deploy_apply_supervisor_blocks_until_state_is_copied(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_and_approve_artifacts(config_path, platform="linux-user")
|
|
config = load_config(config_path)
|
|
|
|
results = apply_supervisor(
|
|
config,
|
|
dry_run=True,
|
|
platform_provider=_linux_user_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
|
|
self.assertTrue(any(result.status == "blocked" for result in results))
|
|
self.assertTrue(
|
|
any("deploy apply --stage state first" in result.message for result in results)
|
|
)
|
|
|
|
def test_deploy_apply_supervisor_dry_run_after_state_apply(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",
|
|
"--yes",
|
|
]
|
|
)
|
|
if status != 0:
|
|
raise AssertionError(error)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"supervisor",
|
|
"--dry-run",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(0, status, error)
|
|
self.assertIn("AMPG_DEPLOY_SUPERVISOR", output)
|
|
self.assertIn("status=planned", output)
|
|
self.assertIn("AMPG_DEPLOY_APPLY_SUMMARY stage=supervisor mode=dry-run", output)
|
|
|
|
def test_deploy_apply_supervisor_live_writes_user_service_files(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_and_approve_artifacts(config_path, platform="linux-user")
|
|
_run_state_apply(config_path, root)
|
|
config = load_config(config_path)
|
|
|
|
results = apply_supervisor(
|
|
config,
|
|
dry_run=False,
|
|
platform_provider=_linux_user_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
daemon_run = (
|
|
root / "supervisor-root/services/ampg-example-i2p-i2pd/i2pd.run"
|
|
)
|
|
loopback_run = (
|
|
root
|
|
/ "supervisor-root/services/ampg-example-i2p-nginx-loopback/nginx-loopback.run"
|
|
)
|
|
daemon_run_exists = daemon_run.exists()
|
|
loopback_run_exists = loopback_run.exists()
|
|
daemon_run_executable = bool(daemon_run.stat().st_mode & 0o111)
|
|
|
|
self.assertTrue(all(result.status == "written" for result in results))
|
|
self.assertTrue(daemon_run_exists)
|
|
self.assertTrue(loopback_run_exists)
|
|
self.assertTrue(daemon_run_executable)
|
|
|
|
def test_deploy_apply_start_cli_reports_missing_supervisor(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",
|
|
"--yes",
|
|
]
|
|
)
|
|
if status != 0:
|
|
raise AssertionError(error)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"start",
|
|
"--dry-run",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(1, status)
|
|
self.assertEqual("", error)
|
|
self.assertIn("AMPG_DEPLOY_START", output)
|
|
self.assertIn("deploy apply --stage supervisor first", output)
|
|
|
|
def test_deploy_apply_start_blocks_until_supervisor_is_installed(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_and_approve_artifacts(config_path)
|
|
_run_termux_state_apply(config_path, root)
|
|
config = load_config(config_path)
|
|
|
|
results = apply_start(
|
|
config,
|
|
dry_run=True,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
|
|
self.assertTrue(any(result.status == "blocked" for result in results))
|
|
self.assertTrue(
|
|
any("deploy apply --stage supervisor first" in result.message for result in results)
|
|
)
|
|
|
|
def test_deploy_apply_start_dry_run_after_supervisor_apply(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_and_approve_artifacts(config_path)
|
|
_run_termux_state_apply(config_path, root)
|
|
_run_termux_supervisor_apply(config_path, root)
|
|
config = load_config(config_path)
|
|
|
|
results = apply_start(
|
|
config,
|
|
dry_run=True,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
|
|
self.assertTrue(all(result.status == "planned" for result in results))
|
|
self.assertIn(("sv-enable", "ampg-example-i2p-i2pd"), [result.command for result in results])
|
|
|
|
def test_deploy_apply_start_live_invokes_runner(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_and_approve_artifacts(config_path)
|
|
_run_termux_state_apply(config_path, root)
|
|
_run_termux_supervisor_apply(config_path, root)
|
|
config = load_config(config_path)
|
|
commands: list[tuple[str, ...]] = []
|
|
|
|
def runner(command: tuple[str, ...]) -> CommandRunResult:
|
|
commands.append(command)
|
|
return CommandRunResult(return_code=0)
|
|
|
|
results = apply_start(
|
|
config,
|
|
dry_run=False,
|
|
platform_provider=_termux_provider(root),
|
|
daemon_probe=_missing_probe,
|
|
command_runner=runner,
|
|
)
|
|
|
|
self.assertTrue(all(result.status == "started" for result in results))
|
|
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)
|
|
|
|
def test_deploy_apply_health_blocks_placeholder_addresses(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",
|
|
"health",
|
|
"--dry-run",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(1, status)
|
|
self.assertEqual("", error)
|
|
self.assertIn("AMPG_DEPLOY_HEALTH", output)
|
|
self.assertIn("status=blocked", output)
|
|
self.assertIn("placeholder address", output)
|
|
|
|
def test_deploy_apply_health_dry_run_after_address_capture(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_i2p_address_file(root)
|
|
status, _, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"addresses",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
"--yes",
|
|
]
|
|
)
|
|
if status != 0:
|
|
raise AssertionError(error)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"health",
|
|
"--dry-run",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(0, status, error)
|
|
self.assertIn("AMPG_DEPLOY_HEALTH", output)
|
|
self.assertIn("status=planned", output)
|
|
self.assertIn("curl --proxy http://127.0.0.1:4444 -fsSL", output)
|
|
self.assertIn("http://captureexample.b32.i2p/", output)
|
|
|
|
def test_deploy_apply_health_live_invokes_runner(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _init_i2p_config(root)
|
|
_write_i2p_address_file(root)
|
|
status, _, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"addresses",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
"--yes",
|
|
]
|
|
)
|
|
if status != 0:
|
|
raise AssertionError(error)
|
|
config = load_config(config_path)
|
|
commands: list[tuple[str, ...]] = []
|
|
|
|
def runner(command: tuple[str, ...]) -> CommandRunResult:
|
|
commands.append(command)
|
|
return CommandRunResult(return_code=0)
|
|
|
|
results = apply_health(
|
|
config,
|
|
dry_run=False,
|
|
command_runner=runner,
|
|
)
|
|
|
|
self.assertTrue(all(result.status == "passed" for result in results))
|
|
self.assertIn(
|
|
(
|
|
"curl",
|
|
"--proxy",
|
|
"http://127.0.0.1:4444",
|
|
"-fsSL",
|
|
"http://captureexample.b32.i2p/",
|
|
),
|
|
commands,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|