mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
276 lines
9.3 KiB
Python
276 lines
9.3 KiB
Python
import contextlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from ampg.cli import main
|
|
from ampg.config import load_config
|
|
from ampg.deploy import deploy_plan
|
|
from ampg.platforms import platform_by_name
|
|
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 _missing_probe(adapter):
|
|
return DaemonProbe(installed=False, running=False, executable_path=None)
|
|
|
|
|
|
class DeployPlanTest(unittest.TestCase):
|
|
def test_deploy_plan_guides_fresh_i2p_only_site(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = root / "gateway.toml"
|
|
init_status, _, init_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"init",
|
|
"site",
|
|
"example",
|
|
"--domain",
|
|
"example.test",
|
|
"--source",
|
|
str(_source(root)),
|
|
"--protocol",
|
|
"i2p",
|
|
]
|
|
)
|
|
self.assertEqual(0, init_status, init_error)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"plan",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(0, status, error)
|
|
self.assertIn("AMPG_DEPLOY_SUMMARY status=todo", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=source status=ready", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=build status=todo", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=dns status=skipped", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=artifacts status=todo", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=health status=todo", output)
|
|
self.assertIn("AMPG_DEPLOY_NEXT step=1 stage=build", output)
|
|
self.assertIn("build --profile mobile-i2p", output)
|
|
self.assertIn("deploy apply --stage addresses --dry-run --profile mobile-i2p", output)
|
|
|
|
def test_deploy_plan_guides_package_apply_for_missing_managed_daemons(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = root / "gateway.toml"
|
|
init_status, _, init_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"init",
|
|
"site",
|
|
"example",
|
|
"--domain",
|
|
"example.test",
|
|
"--source",
|
|
str(_source(root)),
|
|
"--protocol",
|
|
"i2p",
|
|
]
|
|
)
|
|
self.assertEqual(0, init_status, init_error)
|
|
|
|
plan = deploy_plan(
|
|
load_config(config_path),
|
|
profile="mobile-i2p",
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_missing_probe,
|
|
)
|
|
|
|
packages = next(step for step in plan.steps if step.stage == "packages")
|
|
self.assertEqual("todo", packages.status)
|
|
self.assertIn("deploy apply --stage packages --dry-run", packages.command)
|
|
self.assertIn("managed daemon package", packages.message)
|
|
|
|
def test_deploy_plan_shows_dns_review_for_clearnet_site(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = root / "gateway.toml"
|
|
init_status, _, init_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"init",
|
|
"site",
|
|
"example",
|
|
"--domain",
|
|
"example.test",
|
|
"--source",
|
|
str(_source(root)),
|
|
"--preset",
|
|
"full",
|
|
]
|
|
)
|
|
self.assertEqual(0, init_status, init_error)
|
|
|
|
_, output, _ = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"plan",
|
|
"--profile",
|
|
"vps-full",
|
|
]
|
|
)
|
|
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=dns status=todo", output)
|
|
self.assertIn("dns plan --profile vps-full", output)
|
|
|
|
def test_deploy_plan_after_build_moves_build_stage_to_ready(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = root / "gateway.toml"
|
|
init_status, _, init_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"init",
|
|
"site",
|
|
"example",
|
|
"--domain",
|
|
"example.test",
|
|
"--source",
|
|
str(_source(root)),
|
|
"--protocol",
|
|
"i2p",
|
|
]
|
|
)
|
|
self.assertEqual(0, init_status, init_error)
|
|
build_status, _, build_error = _run_cli(["--config", str(config_path), "build"])
|
|
self.assertEqual(0, build_status, build_error)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"plan",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(0, status, error)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=build status=ready", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=artifacts status=todo", output)
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=health status=review", output)
|
|
|
|
def test_deploy_plan_guides_staged_apply_after_approval(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = root / "gateway.toml"
|
|
init_status, _, init_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"init",
|
|
"site",
|
|
"example",
|
|
"--domain",
|
|
"example.test",
|
|
"--source",
|
|
str(_source(root)),
|
|
"--protocol",
|
|
"i2p",
|
|
]
|
|
)
|
|
self.assertEqual(0, init_status, init_error)
|
|
build_status, _, build_error = _run_cli(["--config", str(config_path), "build"])
|
|
self.assertEqual(0, build_status, build_error)
|
|
apply_status, _, apply_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"apply",
|
|
"--dry-run",
|
|
"--write-artifacts",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
self.assertEqual(0, apply_status, apply_error)
|
|
approve_status, _, approve_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"approvals",
|
|
"approve",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
"--all",
|
|
]
|
|
)
|
|
self.assertEqual(0, approve_status, approve_error)
|
|
|
|
status, output, error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"plan",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
self.assertEqual(0, status, error)
|
|
|
|
state_status, _, state_error = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"apply",
|
|
"--stage",
|
|
"state",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
"--yes",
|
|
]
|
|
)
|
|
self.assertEqual(0, state_status, state_error)
|
|
status_after_state, output_after_state, error_after_state = _run_cli(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"deploy",
|
|
"plan",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
self.assertIn("AMPG_DEPLOY_STEP stage=deploy-apply status=todo", output)
|
|
self.assertIn("deploy apply --stage state --dry-run", output)
|
|
self.assertEqual(0, status_after_state, error_after_state)
|
|
self.assertIn("deploy apply --stage supervisor --dry-run", output_after_state)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|