mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
280 lines
8.9 KiB
Python
280 lines
8.9 KiB
Python
import contextlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from ampg.cli import main
|
|
from ampg.config import load_config
|
|
from ampg.selection import (
|
|
parse_protocol_filters,
|
|
protocols_for_selection,
|
|
select_profile,
|
|
select_protocols,
|
|
)
|
|
|
|
|
|
def _write_config(root: Path) -> Path:
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text(
|
|
'<html><body onload="boot()"><script src="app.js"></script><h1>Hello</h1></body></html>',
|
|
encoding="utf-8",
|
|
)
|
|
(source / "app.js").write_text("boot()", encoding="utf-8")
|
|
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"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[site.protocols.clearnet]
|
|
enabled = true
|
|
renderer = "clearnet"
|
|
daemon = "nginx"
|
|
daemon_policy = "adopt"
|
|
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "tor"
|
|
daemon_policy = "auto"
|
|
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
return config_path
|
|
|
|
|
|
class ProtocolSelectionTest(unittest.TestCase):
|
|
def test_parse_protocol_filters_supports_repeated_and_comma_values(self):
|
|
self.assertEqual(
|
|
("tor", "i2p", "gemini"),
|
|
parse_protocol_filters(["tor,i2p", "gemini", "tor"]),
|
|
)
|
|
|
|
def test_select_protocols_filters_sites(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = load_config(_write_config(Path(tmp)))
|
|
selected = select_protocols(config, ("i2p",))
|
|
|
|
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)))
|
|
|
|
with self.assertRaisesRegex(ValueError, "not declared"):
|
|
select_protocols(config, ("gemini",))
|
|
|
|
def test_build_protocol_filter_only_writes_selected_output(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config_path = _write_config(root)
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
status = main(["--config", str(config_path), "build", "--protocol", "i2p"])
|
|
|
|
self.assertEqual(0, status)
|
|
self.assertTrue((root / "out/i2p/index.html").exists())
|
|
self.assertFalse((root / "out/clearnet").exists())
|
|
self.assertFalse((root / "out/tor").exists())
|
|
self.assertIn("protocol=i2p", stdout.getvalue())
|
|
self.assertNotIn("protocol=clearnet", stdout.getvalue())
|
|
|
|
def test_apply_protocol_filter_avoids_unselected_clearnet_block(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), patch("ampg.status.shutil.which", return_value=None):
|
|
status = main(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"apply",
|
|
"--dry-run",
|
|
"--protocol",
|
|
"i2p",
|
|
"--platform",
|
|
"android-termux",
|
|
]
|
|
)
|
|
|
|
output_text = stdout.getvalue()
|
|
|
|
self.assertEqual(0, status)
|
|
self.assertIn("protocol=i2p", output_text)
|
|
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), patch("ampg.status.shutil.which", return_value=None):
|
|
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))
|
|
|
|
stderr = io.StringIO()
|
|
with contextlib.redirect_stderr(stderr):
|
|
status = main(["--config", str(config_path), "status", "--protocol", "gemini"])
|
|
|
|
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()
|