mirror/tests/test_config.py

"""Configuration parsing, the CLI surface, and the debugpy hook."""

from __future__ import annotations

import pytest

from mirror import __main__ as entry
from mirror.access import ConfigError, parse_cidrs
from mirror.config import (
    DEFAULT_MAX_BITRATE_BPS,
    DEFAULT_PORT,
    Settings,
    lan_addresses,
    parse_ice_servers,
)
from mirror.debug import maybe_enable_debugpy


class TestIceServers:
    @pytest.mark.parametrize("raw", ["", "   ", "[]", "none", "null", "not json", "{}", "[1,2]"])
    def test_unusable_input_means_no_turn_stun(self, raw):
        assert parse_ice_servers(raw) == []

    def test_valid_servers_survive_and_junk_is_dropped(self):
        raw = '[{"urls": "stun:router:3478"}, "junk", {"urls": ["turn:h:3478"]}]'
        assert parse_ice_servers(raw) == [
            {"urls": "stun:router:3478"},
            {"urls": ["turn:h:3478"]},
        ]


class TestSettings:
    def test_defaults(self, monkeypatch):
        for name in (
            "MIRROR_HOST",
            "MIRROR_PORT",
            "MIRROR_ICE_SERVERS",
            "MIRROR_MAX_BITRATE",
            "MIRROR_MAX_MIRRORS",
            "MIRROR_ALLOW_CIDRS",
            "MIRROR_TRUSTED_PROXIES",
        ):
            monkeypatch.delenv(name, raising=False)
        settings = Settings.from_env()
        assert settings.port == DEFAULT_PORT
        assert settings.host == "0.0.0.0"
        assert settings.ice_servers == []
        assert settings.max_bitrate_bps == DEFAULT_MAX_BITRATE_BPS
        assert settings.max_streams == 0
        # Unrestricted until told otherwise, and only loopback may speak for a client.
        assert settings.allow_cidrs == ()
        assert [str(network) for network in settings.trusted_proxies] == [
            "127.0.0.0/8",
            "::1/128",
        ]
        assert settings.gate().enabled is False

    def test_environment_wins(self, monkeypatch):
        monkeypatch.setenv("MIRROR_HOST", "127.0.0.1")
        monkeypatch.setenv("MIRROR_PORT", "9123")
        monkeypatch.setenv("MIRROR_MAX_BITRATE", "250000")
        monkeypatch.setenv("MIRROR_MAX_MIRRORS", "3")
        monkeypatch.setenv("MIRROR_ICE_SERVERS", '[{"urls": "stun:box:3478"}]')
        monkeypatch.setenv("MIRROR_ALLOW_CIDRS", "10.2.0.0/22, fd00::/64")
        monkeypatch.setenv("MIRROR_TRUSTED_PROXIES", "10.1.1.27")
        settings = Settings.from_env()
        assert (settings.host, settings.port) == ("127.0.0.1", 9123)
        assert [str(network) for network in settings.allow_cidrs] == ["10.2.0.0/22", "fd00::/64"]
        assert [str(network) for network in settings.trusted_proxies] == ["10.1.1.27/32"]
        assert settings.max_bitrate_bps == 250_000
        assert settings.max_streams == 3
        assert settings.ice_servers == [{"urls": "stun:box:3478"}]

    @pytest.mark.parametrize("value", ["abc", "", "80x"])
    def test_a_bogus_number_falls_back_instead_of_crashing(self, monkeypatch, value):
        monkeypatch.setenv("MIRROR_PORT", value)
        assert Settings.from_env().port == DEFAULT_PORT

    def test_client_config_is_a_whitelist(self, monkeypatch):
        monkeypatch.setenv("MIRROR_ICE_SERVERS", '[{"urls": "stun:box:3478"}]')
        config = Settings.from_env().client_config()
        assert set(config) == {"iceServers", "maxBitrateBps", "pingIntervalMs"}

    def test_a_bad_cidr_refuses_to_start_rather_than_failing_open(self, monkeypatch):
        monkeypatch.setenv("MIRROR_ALLOW_CIDRS", "office")
        with pytest.raises(ConfigError, match="MIRROR_ALLOW_CIDRS"):
            Settings.from_env()

    def test_the_gate_carries_both_halves_of_the_setting(self):
        settings = Settings(
            allow_cidrs=parse_cidrs("10.2.0.0/22"), trusted_proxies=parse_cidrs("127.0.0.1")
        )
        gate = settings.gate()
        assert gate.enabled is True
        assert gate.check("127.0.0.1", "10.2.1.5") == (True, "10.2.1.5")

    def test_lan_addresses_are_real_addresses(self):
        for address in lan_addresses():
            assert not address.startswith("127.")
            assert address.count(".") == 3 or ":" in address


class TestCli:
    def test_bare_invocation_means_serve(self):
        parser = entry.build_parser()
        argv = ["serve", "--port", "8123", "--host", "127.0.0.1"]
        args = parser.parse_args(argv)
        assert args.command == "serve"
        assert args.port == 8123

    def test_allow_cidr_is_repeatable_and_replaces_the_environment(self, monkeypatch):
        monkeypatch.setenv("MIRROR_ALLOW_CIDRS", "192.168.0.0/16")
        args = entry.build_parser().parse_args(
            ["serve", "--allow-cidr", "10.2.0.0/22", "--allow-cidr", "fd00::/64"]
        )
        assert args.allow_cidr == ["10.2.0.0/22", "fd00::/64"]
        # cmd_serve replaces rather than merges, so the environment cannot leave a
        # second, wider network open by accident.
        assert [str(network) for network in parse_cidrs(",".join(args.allow_cidr))] == [
            "10.2.0.0/22",
            "fd00::/64",
        ]

    @pytest.mark.parametrize(
        ("given", "expected"),
        [
            ([], ["serve"]),
            (["--port", "9999"], ["serve", "--port", "9999"]),
            (["--allow-cidr", "10.2.0.0/22"], ["serve", "--allow-cidr", "10.2.0.0/22"]),
            (["serve"], ["serve"]),
            (["serve", "--version"], ["serve", "--version"]),
            (["--version"], ["--version"]),
            (["--help"], ["--help"]),
            (["-h"], ["-h"]),
        ],
    )
    def test_only_informational_flags_escape_the_implicit_serve(self, given, expected):
        assert entry.with_serve(given) == expected

    @pytest.mark.parametrize("argv", [["--version"], ["serve", "--version"]])
    def test_version_prints_and_exits_cleanly(self, argv, capsys):
        with pytest.raises(SystemExit) as caught:
            entry.main(argv)
        assert caught.value.code == 0
        assert f"mirror {entry.__version__}" in capsys.readouterr().out

    def test_help_lists_the_subcommand_it_is_not_being_rewritten_into(self, capsys):
        with pytest.raises(SystemExit):
            entry.main(["--help"])
        out = capsys.readouterr().out
        assert "usage: mirror" in out
        assert "serve" in out

    def test_a_bad_allow_cidr_exits_with_a_message_not_a_traceback(self, capsys):
        # Config is parsed before anything binds or imports, so this cannot stall
        # on a stray DEBUGPY in the operator's shell.
        code = entry.cmd_serve(
            entry.build_parser().parse_args(["serve", "--allow-cidr", "nonsense"])
        )
        assert code == 2
        assert "--allow-cidr" in capsys.readouterr().err

    def test_port_validation_rejects_nonsense(self):
        parser = entry.build_parser()
        with pytest.raises(SystemExit):
            parser.parse_args(["serve", "--port", "nope"])
        with pytest.raises(SystemExit):
            parser.parse_args(["serve", "--port", "70000"])

    def test_version_exits_cleanly(self, capsys):
        with pytest.raises(SystemExit) as excinfo:
            entry.build_parser().parse_args(["--version"])
        assert excinfo.value.code == 0
        assert "mirror" in capsys.readouterr().out

    def test_flags_without_a_subcommand_still_serve(self, monkeypatch):
        """`mirror --port 8901` must work as well as `mirror serve --port 8901`."""
        import sys

        captured = {}

        class StubUvicorn:
            @staticmethod
            def run(_app, **kwargs):
                captured.update(kwargs)
                raise KeyboardInterrupt

        for name in ("MIRROR_HOST", "MIRROR_PORT", "MIRROR_ICE_SERVERS", "DEBUGPY"):
            monkeypatch.delenv(name, raising=False)
        monkeypatch.setitem(sys.modules, "uvicorn", StubUvicorn)
        with pytest.raises(KeyboardInterrupt):
            entry.main(["--port", "8901", "--host", "127.0.0.1"])
        assert captured["port"] == 8901
        assert captured["host"] == "127.0.0.1"
        assert captured["access_log"] is False


class TestDebugpy:
    @pytest.mark.parametrize("value", ["", "0", "off", "no", "FALSE"])
    def test_stays_asleep_unless_explicitly_asked(self, monkeypatch, value):
        monkeypatch.setenv("DEBUGPY", value)
        assert maybe_enable_debugpy() is False