"""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