mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
Add deployment profiles
This commit is contained in:
@@ -6,7 +6,12 @@ from pathlib import Path
|
||||
|
||||
from ampg.cli import main
|
||||
from ampg.config import load_config
|
||||
from ampg.selection import parse_protocol_filters, select_protocols
|
||||
from ampg.selection import (
|
||||
parse_protocol_filters,
|
||||
protocols_for_selection,
|
||||
select_profile,
|
||||
select_protocols,
|
||||
)
|
||||
|
||||
|
||||
def _write_config(root: Path) -> Path:
|
||||
@@ -20,6 +25,18 @@ def _write_config(root: Path) -> Path:
|
||||
config_path = root / "gateway.toml"
|
||||
config_path.write_text(
|
||||
"""
|
||||
[profiles.mobile-i2p]
|
||||
protocols = ["i2p"]
|
||||
platform = "android-termux"
|
||||
|
||||
[profiles.tor-i2p]
|
||||
protocols = ["tor", "i2p"]
|
||||
|
||||
[profiles.artifact-profile]
|
||||
protocols = ["i2p"]
|
||||
platform = "android-termux"
|
||||
write_artifacts = true
|
||||
|
||||
[[site]]
|
||||
id = "example"
|
||||
domain = "example.test"
|
||||
@@ -69,6 +86,37 @@ class ProtocolSelectionTest(unittest.TestCase):
|
||||
|
||||
self.assertEqual(["i2p"], list(selected.sites[0].protocols))
|
||||
|
||||
def test_config_parses_profiles(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config = load_config(_write_config(Path(tmp)))
|
||||
|
||||
profile = select_profile(config, "mobile-i2p")
|
||||
self.assertIsNotNone(profile)
|
||||
self.assertEqual(("i2p",), profile.protocols)
|
||||
self.assertEqual("android-termux", profile.platform)
|
||||
self.assertFalse(profile.write_artifacts)
|
||||
|
||||
def test_profile_supplies_protocols_when_no_cli_protocol_is_given(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config = load_config(_write_config(Path(tmp)))
|
||||
|
||||
self.assertEqual(
|
||||
("i2p",),
|
||||
protocols_for_selection(
|
||||
config,
|
||||
raw_protocols=[],
|
||||
profile_name="mobile-i2p",
|
||||
),
|
||||
)
|
||||
self.assertEqual(
|
||||
("tor",),
|
||||
protocols_for_selection(
|
||||
config,
|
||||
raw_protocols=["tor"],
|
||||
profile_name="mobile-i2p",
|
||||
),
|
||||
)
|
||||
|
||||
def test_select_protocols_rejects_unknown_protocol(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config = load_config(_write_config(Path(tmp)))
|
||||
@@ -122,6 +170,87 @@ class ProtocolSelectionTest(unittest.TestCase):
|
||||
self.assertIn("blocked=0", output_text)
|
||||
self.assertNotIn("protocol=clearnet", output_text)
|
||||
|
||||
def test_apply_profile_filters_protocol_and_uses_platform(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
config_path = _write_config(root)
|
||||
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",
|
||||
"--profile",
|
||||
"mobile-i2p",
|
||||
]
|
||||
)
|
||||
|
||||
output_text = stdout.getvalue()
|
||||
|
||||
self.assertEqual(0, status)
|
||||
self.assertIn("protocol=i2p", output_text)
|
||||
self.assertIn("i2pd is missing; AMPG would prepare managed setup", output_text)
|
||||
self.assertIn("blocked=0", output_text)
|
||||
self.assertNotIn("protocol=clearnet", output_text)
|
||||
|
||||
def test_cli_protocol_overrides_profile_protocols(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_path = _write_config(Path(tmp))
|
||||
|
||||
stdout = io.StringIO()
|
||||
with contextlib.redirect_stdout(stdout):
|
||||
status = main(
|
||||
[
|
||||
"--config",
|
||||
str(config_path),
|
||||
"status",
|
||||
"--profile",
|
||||
"mobile-i2p",
|
||||
"--protocol",
|
||||
"tor",
|
||||
]
|
||||
)
|
||||
|
||||
output_text = stdout.getvalue()
|
||||
|
||||
self.assertEqual(0, status)
|
||||
self.assertIn("protocol=tor", output_text)
|
||||
self.assertIn("platform=android-termux", output_text)
|
||||
self.assertNotIn("protocol=i2p", output_text)
|
||||
|
||||
def test_profile_write_artifacts_defaults_apply_artifacts(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
config_path = _write_config(root)
|
||||
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",
|
||||
"--profile",
|
||||
"artifact-profile",
|
||||
]
|
||||
)
|
||||
|
||||
output_text = stdout.getvalue()
|
||||
|
||||
self.assertEqual(0, status)
|
||||
self.assertIn("AMPG_APPLY_ARTIFACT", output_text)
|
||||
self.assertIn("blocked=0", output_text)
|
||||
|
||||
def test_cli_rejects_unknown_protocol_filter(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_path = _write_config(Path(tmp))
|
||||
@@ -133,6 +262,17 @@ class ProtocolSelectionTest(unittest.TestCase):
|
||||
self.assertEqual(1, status)
|
||||
self.assertIn("selected protocol(s) not declared", stderr.getvalue())
|
||||
|
||||
def test_cli_rejects_unknown_profile(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_path = _write_config(Path(tmp))
|
||||
|
||||
stderr = io.StringIO()
|
||||
with contextlib.redirect_stderr(stderr):
|
||||
status = main(["--config", str(config_path), "status", "--profile", "nope"])
|
||||
|
||||
self.assertEqual(1, status)
|
||||
self.assertIn("unknown profile", stderr.getvalue())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user