mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
139 lines
4.2 KiB
Python
139 lines
4.2 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.selection import parse_protocol_filters, 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(
|
|
"""
|
|
[[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_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):
|
|
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_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())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|