ducoterra
7359ee4e07
130 tests, ~20s, 94% coverage. Four rungs, each proving something the one below it cannot: registry admission, fan-out, signal authorization and teardown, no I/O wire camelCase round-trip, malformed frames, WebSocket lifecycle e2e real capture -> encode -> SRTP -> decode -> painted frames realdisplay the actual X11 capture backend on a private Xvfb display The e2e rung drives Chromium's fake media device, which produces a genuine moving 1280x720 monitor stream, so the assertion is that a watcher's <video> has a size and that its pixels change over time -- a frozen or never-started peer connection fails the suite, which a "signal arrived" assertion would not have caught. Findings encoded as flags and comments, because each cost a debugging cycle: the headless shell has no media stack at all (use channel="chromium"); mediaDevices is undefined off a secure context (serve over 127.0.0.1); and --use-fake-ui-for-media-stream breaks real X11 capture by approving the request without ever selecting a source, so the realdisplay suite selects the source explicitly instead. The realdisplay fixture strips WAYLAND_DISPLAY and DBUS_SESSION_BUS_ADDRESS from the browser environment. Without that, a headed Chromium reaches xdg-desktop-portal, mirrors the developer's actual desktop and raises a permission dialog on their real screen -- which is exactly what happened before this was isolated. Skipped unless MIRROR_REALDISPLAY=1. Also guards the frontend's structure: scripts must parse as ES modules, every id a script looks up must exist in its page, ids must not collide across pages that share one stylesheet, and nothing may load from a CDN. LLM-Generated: true
117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
"""Configuration parsing, the CLI surface, and the debugpy hook."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from mirror import __main__ as entry
|
|
from mirror.config import 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",
|
|
):
|
|
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_streams == 0
|
|
|
|
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"}]')
|
|
settings = Settings.from_env()
|
|
assert (settings.host, settings.port) == ("127.0.0.1", 9123)
|
|
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_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_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
|