Experimental/mirror
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
24cb10ee5c53e10778af4ef531f0148029f0d06d
mirror/tests/e2e/test_live_mirror.py
T
ducoterra 7359ee4e07 test: four-rung suite that asserts on decoded pixels, not log lines 
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
2026-11-01 16:54:28 -05:00

270 lines
11 KiB
Python
Raw Blame History

"""End-to-end: a real `mirror serve`, real Chromium, real captured frames.
Every assertion about "it works" is made on pixels that reached the viewer's
<video>, not on log lines. If the media path breaks, these tests break.
"""
from __future__ import annotations
import re
import time
import pytest
from playwright.sync_api import expect
from .conftest import frames_matching, start_sharing, stop_sharing
pytestmark = [pytest.mark.e2e, pytest.mark.timeout(180)]
@pytest.fixture
def device(base_url, browser, request):
"""Open a page as its own device (isolated context, fresh peer id)."""
opened = []
def _open(path: str = "/", errors: list | None = None):
context = browser.new_context(viewport={"width": 1100, "height": 720})
opened.append(context)
request.addfinalizer(context.close)
page = context.new_page()
if errors is not None:
page.on("pageerror", lambda error: errors.append(str(error)))
page.goto(f"{base_url}{path}", wait_until="domcontentloaded")
expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
return page
return _open
def tile_ids(page) -> list[str]:
return page.locator(".tile").evaluate_all("els => els.map(e => e.dataset.stream)")
class TestShareThenWatch:
def test_a_shared_window_reaches_a_viewer_as_moving_pixels(self, device):
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
expect(sharer.locator("#stat-resolution")).to_have_text(
re.compile(r"\d{3,4}\s.\s\d{3,4}"), timeout=10_000
)
expect(sharer.locator("#stat-surface")).to_contain_text("Screen", timeout=5_000)
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
frames = frames_matching(viewer, ".tile video")
assert not frames.get("missing"), "no <video> element in the tile"
assert frames["width"] > 0 and frames["height"] > 0, f"no decoded frame size: {frames}"
assert frames["distinct"] >= 2, f"video is frozen: {frames}"
# The sharer learns that someone is watching.
expect(sharer.locator("#viewer-count")).to_have_text("1", timeout=10_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_clicking_a_tile_takes_it_on_stage(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
viewer.locator(".tile").click()
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
expect(viewer.locator("#watch-stage video")).to_have_count(1)
# Still moving while staged — staging must not restart the peer connection.
frames = frames_matching(viewer, "#watch-stage video")
assert frames["distinct"] >= 2, f"staged video froze: {frames}"
# And the mirror survives the stage: the sharer still has one viewer.
expect(sharer.locator("#viewer-count")).to_have_text("1", timeout=10_000)
def test_stopping_the_share_clears_the_wall(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
stop_sharing(sharer)
expect(viewer.locator(".tile")).to_have_count(0, timeout=10_000)
expect(viewer.locator("#empty")).to_be_visible()
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"))
def test_closing_the_sharer_tab_clears_the_wall(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
sharer.close()
expect(viewer.locator(".tile")).to_have_count(0, timeout=10_000)
class TestChromeVisibility:
"""The page must never advertise a control that does not apply to its state."""
def test_the_idle_share_page_offers_only_a_start(self, device):
sharer = device("/mirror")
expect(sharer.locator("#start")).to_be_visible()
expect(sharer.locator("#placeholder")).to_be_visible()
expect(sharer.locator("#stop")).not_to_be_visible()
expect(sharer.locator("#livetag")).not_to_be_visible()
start_sharing(sharer)
expect(sharer.locator("#stop")).to_be_visible()
expect(sharer.locator("#livetag")).to_be_visible()
expect(sharer.locator("#start")).not_to_be_visible()
expect(sharer.locator("#placeholder")).not_to_be_visible()
def test_exit_only_appears_on_the_tile_that_is_on_stage(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
expect(viewer.locator(".tile .exit")).not_to_be_visible()
viewer.locator(".tile").click()
expect(viewer.locator("#watch-stage .exit")).to_be_visible()
viewer.locator("#watch-stage .exit").click()
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator(".tile .exit")).not_to_be_visible()
# Leaving the stage must not have disturbed the stream.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=20_000)
class TestManyToMany:
def test_one_mirror_fans_out_to_several_viewers(self, device):
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
viewers = [device("/watch", errors) for _ in range(3)]
for viewer in viewers:
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
expect(sharer.locator("#viewer-count")).to_have_text("3", timeout=15_000)
expect(sharer.locator("#audience li")).to_have_count(3, timeout=10_000)
for viewer in viewers:
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"a viewer froze: {frames}"
# One viewer leaving must not disturb the others.
viewers[0].close()
expect(sharer.locator("#viewer-count")).to_have_text("2", timeout=15_000)
for viewer in viewers[1:]:
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"collateral damage: {frames}"
assert errors == [], f"uncaught browser errors: {errors}"
def test_several_mirrors_land_on_one_wall(self, device):
errors: list = []
sharers = []
for _ in range(2):
page = device("/mirror", errors)
start_sharing(page)
sharers.append(page)
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(2, timeout=15_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=30_000)
ids = tile_ids(viewer)
assert len(ids) == 2 and len(set(ids)) == 2, "tiles must be distinct mirrors"
# Stage the second mirror, then jump to the first.
viewer.locator(".tile").nth(1).click()
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
staged = viewer.locator("#watch-stage .tile").get_attribute("data-stream")
assert staged == ids[1]
viewer.locator("#watch-stage .exit").click()
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
viewer.locator(".tile").nth(0).click()
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
# Both mirrors are still alive and untouched by the staging gymnastics.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_stopping_one_mirror_leaves_the_other_running(self, device):
first = device("/mirror")
start_sharing(first)
second = device("/mirror")
start_sharing(second)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=30_000)
stop_sharing(first)
expect(viewer.locator(".tile")).to_have_count(1, timeout=10_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=20_000)
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"the surviving mirror froze: {frames}"
class TestTimingAndRecovery:
def test_a_viewer_already_on_the_wall_sees_a_late_mirror(self, device):
viewer = device("/watch")
expect(viewer.locator("#empty")).to_be_visible()
sharer = device("/mirror")
start_sharing(sharer)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"late mirror did not animate: {frames}"
def test_a_reloaded_viewer_recovers_the_live_mirror(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
viewer.reload(wait_until="domcontentloaded")
expect(viewer.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"reload did not restore motion: {frames}"
expect(sharer.locator("#viewer-count")).to_have_text("1", timeout=15_000)
def test_the_index_page_counts_what_is_live(self, device):
index = device("/")
expect(index.locator("#livebar")).to_contain_text("No mirrors live", timeout=10_000)
sharer = device("/mirror")
start_sharing(sharer)
expect(index.locator("#livebar")).to_contain_text("1 mirror", timeout=10_000)
class TestRosterTruthfulness:
def test_the_api_agrees_with_what_the_browser_shows(self, device, base_url):
import httpx
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
# Give the stats ticker a beat, then compare server truth to pixels.
time.sleep(1.0)
payload = httpx.get(f"{base_url}/api/streams", timeout=5).json()
assert payload["stats"]["mirrors"] == 1
assert payload["stats"]["watching"] == 1
mirror = payload["mirrors"][0]
assert mirror["width"] and mirror["height"]
assert mirror["displaySurface"] in ("monitor", "window", "browser")
Reference in New Issue View Git Blame Copy Permalink