adaptive-multi-protocol-gateway/tests/test_addresses.py

import contextlib
import io
import json
import tempfile
import unittest
from pathlib import Path

from ampg.addresses import (
    address_registry_path,
    capture_addresses,
    effective_address_records,
    set_address,
)
from ampg.cli import main
from ampg.config import load_config
from ampg.health import health_plan
from ampg.manifest import fixture_manifest


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"""
[gateway]
state_dir = "./state"
cache_dir = "./cache"
run_dir = "./run"
user = "ampg"

[profiles.mobile-i2p]
protocols = ["i2p"]

[[site]]
id = "example"
domain = "example.test"

[site.source]
kind = "static-html"
path = "./site"
canonical_url = "https://example.test"

[site.outputs]
root = "./out"
plan_root = "./plan"

{protocol_block}
""",
        encoding="utf-8",
    )
    return config_path


def _mark_output(root: Path, protocol: str) -> None:
    output = root / f"out/{protocol}"
    output.mkdir(parents=True)
    (output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")


class AddressRegistryTest(unittest.TestCase):
    def test_gateway_paths_are_parsed(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config = load_config(
                _write_config(
                    root,
                    """
[site.protocols.i2p]
enabled = true
renderer = "privacy-html"
""",
                )
            )

        self.assertEqual(root / "state", config.paths.state_dir)
        self.assertEqual(root / "cache", config.paths.cache_dir)
        self.assertEqual(root / "run", config.paths.run_dir)
        self.assertEqual("ampg", config.paths.user)

    def test_manual_address_drives_manifest_and_health_plan(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config = load_config(
                _write_config(
                    root,
                    """
[site.protocols.i2p]
enabled = true
renderer = "privacy-html"
""",
                )
            )
            record = set_address(
                config,
                site_id="example",
                protocol="i2p",
                url="manualexample.b32.i2p",
            )
            _mark_output(root, "i2p")

            records = effective_address_records(config)
            manifest = fixture_manifest(config, config.sites[0])
            checks = health_plan(config)

        self.assertEqual("http://manualexample.b32.i2p/", record.url)
        self.assertEqual(root / "state/addresses.json", address_registry_path(config))
        self.assertEqual("http://manualexample.b32.i2p/", records[0].url)
        self.assertEqual("captured", records[0].address_status)
        self.assertEqual("http://manualexample.b32.i2p/", manifest["fixtures"][0]["url"])
        self.assertEqual("captured", manifest["fixtures"][0]["address_status"])
        self.assertEqual("planned", checks[0].status)

    def test_capture_reads_i2p_address_from_state_file(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config = load_config(
                _write_config(
                    root,
                    """
[site.protocols.i2p]
enabled = true
renderer = "privacy-html"
""",
                )
            )
            address_file = root / "state/example/i2p/b32.txt"
            address_file.parent.mkdir(parents=True)
            address_file.write_text("capturedexample.b32.i2p\n", encoding="utf-8")

            results = capture_addresses(config)
            registry = json.loads(address_registry_path(config).read_text(encoding="utf-8"))

        self.assertEqual("captured", results[0].status)
        self.assertEqual(address_file, results[0].path)
        self.assertEqual("http://capturedexample.b32.i2p/", registry["addresses"][0]["url"])

    def test_addresses_cli_set_and_list_respect_profile_scope(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _write_config(
                root,
                """
[site.protocols.clearnet]
enabled = true
renderer = "clearnet"

[site.protocols.i2p]
enabled = true
renderer = "privacy-html"
""",
            )

            set_stdout = io.StringIO()
            with contextlib.redirect_stdout(set_stdout):
                set_status = main(
                    [
                        "--config",
                        str(config_path),
                        "addresses",
                        "set",
                        "--site",
                        "example",
                        "--protocol",
                        "i2p",
                        "--url",
                        "cliexample.b32.i2p",
                    ]
                )

            list_stdout = io.StringIO()
            with contextlib.redirect_stdout(list_stdout):
                list_status = main(
                    [
                        "--config",
                        str(config_path),
                        "addresses",
                        "list",
                        "--profile",
                        "mobile-i2p",
                    ]
                )

        self.assertEqual(0, set_status)
        self.assertEqual(0, list_status)
        self.assertIn("AMPG_ADDRESS_SET status=written", set_stdout.getvalue())
        self.assertIn("protocol=i2p", list_stdout.getvalue())
        self.assertIn("captured=1", list_stdout.getvalue())
        self.assertNotIn("protocol=clearnet", list_stdout.getvalue())

    def test_addresses_cli_capture_uses_profile_scope(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            config_path = _write_config(
                root,
                """
[site.protocols.clearnet]
enabled = true
renderer = "clearnet"

[site.protocols.i2p]
enabled = true
renderer = "privacy-html"
""",
            )
            address_file = root / "state/example/i2p/hostname.txt"
            address_file.parent.mkdir(parents=True)
            address_file.write_text("capturecli.b32.i2p\n", encoding="utf-8")

            stdout = io.StringIO()
            with contextlib.redirect_stdout(stdout):
                status = main(
                    [
                        "--config",
                        str(config_path),
                        "addresses",
                        "capture",
                        "--profile",
                        "mobile-i2p",
                    ]
                )

        output = stdout.getvalue()
        self.assertEqual(0, status)
        self.assertIn("protocol=i2p", output)
        self.assertIn("status=captured", output)
        self.assertIn("captured=1", output)
        self.assertNotIn("protocol=clearnet", output)


if __name__ == "__main__":
    unittest.main()