mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
257 lines
8.3 KiB
Python
257 lines
8.3 KiB
Python
import contextlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from ampg.activation import activation_steps, blocked_steps
|
|
from ampg.cli import main
|
|
from ampg.config import load_config
|
|
from ampg.platforms import platform_by_name
|
|
from ampg.status import DaemonProbe
|
|
|
|
|
|
def _probe(*, installed: bool = False, running: bool = False):
|
|
def probe(adapter):
|
|
executable_path = f"/usr/bin/{adapter.executable}" if installed else None
|
|
return DaemonProbe(
|
|
installed=installed,
|
|
running=running,
|
|
executable_path=executable_path,
|
|
)
|
|
|
|
return probe
|
|
|
|
|
|
def _write_config(root: Path, protocol_block: str) -> Path:
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>Hello</h1>", encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
f"""
|
|
[[site]]
|
|
id = "example"
|
|
domain = "example.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
{protocol_block}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
return config_path
|
|
|
|
|
|
class ActivationTest(unittest.TestCase):
|
|
def test_activation_plan_reviews_placeholder_address_before_health(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config_path = _write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
local_port = 19081
|
|
""",
|
|
)
|
|
output = root / "out/i2p"
|
|
output.mkdir(parents=True)
|
|
(output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
|
|
config = load_config(config_path)
|
|
|
|
steps = activation_steps(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
self.assertEqual([], blocked_steps(steps))
|
|
by_stage = {step.stage: step for step in steps if step.stage != "artifact"}
|
|
self.assertEqual("ready", by_stage["output"].status)
|
|
self.assertEqual("manage-owned", by_stage["daemon"].action)
|
|
self.assertEqual("capture-address", by_stage["address"].action)
|
|
self.assertEqual("review", by_stage["address"].status)
|
|
self.assertIn("addresses capture --protocol i2p", by_stage["address"].command)
|
|
self.assertEqual("review", by_stage["health"].status)
|
|
self.assertTrue(any(step.stage == "artifact" for step in steps))
|
|
|
|
def test_configured_address_allows_planned_activation_health(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config_path = _write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
i2p_url = "ready.b32.i2p"
|
|
""",
|
|
)
|
|
output = root / "out/i2p"
|
|
output.mkdir(parents=True)
|
|
(output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
|
|
config = load_config(config_path)
|
|
|
|
steps = activation_steps(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
by_stage = {step.stage: step for step in steps if step.stage != "artifact"}
|
|
self.assertEqual("use-address", by_stage["address"].action)
|
|
self.assertEqual("ready", by_stage["address"].status)
|
|
self.assertEqual("planned", by_stage["health"].status)
|
|
|
|
def test_missing_output_blocks_activation(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_path = _write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "tor"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
config = load_config(config_path)
|
|
|
|
steps = activation_steps(
|
|
config,
|
|
platform_provider=platform_by_name("linux-user"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
blocked = blocked_steps(steps)
|
|
self.assertTrue(any(step.stage == "output" for step in blocked))
|
|
self.assertTrue(any(step.stage == "health" for step in blocked))
|
|
|
|
def test_apply_cli_requires_dry_run(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_path = _write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.gemini]
|
|
enabled = true
|
|
renderer = "gemtext"
|
|
daemon = "agate"
|
|
daemon_policy = "external"
|
|
""",
|
|
)
|
|
|
|
stderr = io.StringIO()
|
|
with contextlib.redirect_stderr(stderr):
|
|
status = main(["--config", str(config_path), "apply"])
|
|
|
|
self.assertEqual(1, status)
|
|
self.assertIn("live apply is not implemented", stderr.getvalue())
|
|
|
|
def test_apply_dry_run_can_write_review_artifacts(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config_path = _write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.gemini]
|
|
enabled = true
|
|
renderer = "gemtext"
|
|
daemon = "agate"
|
|
daemon_policy = "external"
|
|
port = 1965
|
|
""",
|
|
)
|
|
output = root / "out/gemini"
|
|
output.mkdir(parents=True)
|
|
(output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
status = main(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"apply",
|
|
"--dry-run",
|
|
"--write-artifacts",
|
|
"--platform",
|
|
"unknown",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(0, status)
|
|
output_text = stdout.getvalue()
|
|
self.assertIn("AMPG_APPLY_ARTIFACT", output_text)
|
|
self.assertIn("AMPG_APPLY_STEP", output_text)
|
|
self.assertIn('command="-"', output_text)
|
|
self.assertIn("blocked=0", output_text)
|
|
artifact_line = next(
|
|
line for line in output_text.splitlines() if line.startswith("AMPG_APPLY_ARTIFACT")
|
|
)
|
|
artifact_path = Path(
|
|
next(
|
|
part.removeprefix("path=")
|
|
for part in artifact_line.split()
|
|
if part.startswith("path=")
|
|
)
|
|
)
|
|
self.assertTrue(artifact_path.exists())
|
|
|
|
def test_apply_dry_run_write_artifacts_prints_state_copy_plan(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config_path = _write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
output = root / "out/i2p"
|
|
output.mkdir(parents=True)
|
|
(output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
status = main(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"apply",
|
|
"--dry-run",
|
|
"--write-artifacts",
|
|
"--platform",
|
|
"android-termux",
|
|
]
|
|
)
|
|
|
|
output_text = stdout.getvalue()
|
|
self.assertEqual(0, status)
|
|
self.assertIn("AMPG_APPLY_STATE_COPY site=example protocol=i2p", output_text)
|
|
self.assertIn("kind=daemon-config", output_text)
|
|
self.assertIn("kind=loopback-config", output_text)
|
|
self.assertIn(str(root / ".ampg/state/example/i2p/i2pd-tunnels.conf"), output_text)
|
|
self.assertIn(str(root / ".ampg/state/example/i2p/nginx-loopback.conf"), output_text)
|
|
self.assertIn("AMPG_APPLY_SUPERVISOR site=example protocol=i2p", output_text)
|
|
self.assertIn("service=ampg-example-i2p-i2pd", output_text)
|
|
self.assertIn("service=ampg-example-i2p-nginx-loopback", output_text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|