ducoterra
1da466a412
**Phase 02 final verification pass — complete, 100% green.** **Verified (all 4 tasks in `complete/02_sharp_capture/`):** - Task 1: `pickSurface()` requests `screen × devicePixelRatio` (capped 3840×2160) instead of window size - Task 2: `degradationPreference: "maintain-resolution"` + per-leg `maxBitrate`, re-applied after `setLocalDescription` *and* on `connected` (both call sites asserted in `test_app.py`) - Task 3: default cap raised to 25 Mbit/s — `config.py` constant, `net.js` offline fallback, `.env.example`, README, and the drift-guard test all agree - Task 4: realdisplay sharpness floor (viewer decoded width ≥ 1152px = 90% of Xvfb 1280px) in `test_sharp_picture_reaches_the_wall` **Gates (all green):** - `uv run ruff check . && uv run ruff format --check .` → clean - `uv run pyright` → 0 errors - `uv run pytest` → 231 passed, 3 skipped (realdisplay), coverage 95% - `MIRROR_REALDISPLAY=1 uv run pytest -m realdisplay` → 3 passed on a real Xvfb + headed Chromium **Completion criteria:** (1) 90% width floor asserted on viewer's decoded pixels — PASS · (2) `/api/config` `maxBitrateBps: 25000000` default, net.js fallback agrees — PASS · (3) `#stat-resolution` matches viewer-decoded size incl. documented window downscale (1100×620, 1280×720 fake-device tests) — PASS · (4) both suites green, aspect-ratio guard (`1280/800` ± 0.06 + `≠ 1280×720`) still distinguishes real from fake capture — PASS · (5) README Configuration documents `25000000` — PASS; `PLAN.md` §6 "default 12 Mbit/s" left untouched as instructed — **flag for the user to amend**. **Docs:** README — fixed suite timing `~20s → ~50s` (measured 49s); all other claims verified (CLI flags, dev extra, config table, kiosk key table vs. `watch.js`/`share.js` routers, kiosk commands vs. `docs/kiosk-silverblue.md`). AGENTS.md — same stale timing fixed, nothing else stale (all referenced test names/files exist). **Notable:** `docs/kiosk-silverblue.md` + its scripts use `mirror.reeselink.com` while README/SECURITY.md/Containerfile use `mirror.reeseapps.com` — a pre-existing cross-file mismatch I could not resolve from the repo (both in same-day commits; live-infra fact). Left untouched; user should confirm the live kiosk URL. No live changes made. **Next pending phase:** `03_audio_share`.
586 lines
27 KiB
Python
586 lines
27 KiB
Python
"""HTTP surface + signaling socket tests. No browser required."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from mirror.access import parse_cidrs
|
|
from mirror.app import create_app
|
|
from mirror.config import Settings
|
|
|
|
STATIC = Path(__file__).resolve().parent.parent / "src/mirror/static"
|
|
|
|
|
|
def gated(*allow, client_address="203.0.113.9", **kwargs):
|
|
"""A client speaking as ``client_address`` to a server restricted to ``allow``."""
|
|
settings = Settings(allow_cidrs=parse_cidrs(",".join(allow)), **kwargs)
|
|
return TestClient(create_app(settings), client=(client_address, 44444))
|
|
|
|
|
|
EXTERNAL_REF = re.compile(r"""(?:src|href)\s*=\s*["']https?://""", re.IGNORECASE)
|
|
|
|
|
|
def handshake(socket, role, name=""):
|
|
socket.send_json({"type": "hello", "role": role, "name": name})
|
|
message = socket.receive_json()
|
|
assert message["type"] == "welcome", message
|
|
return message["peerId"]
|
|
|
|
|
|
def wait_for(socket, kind):
|
|
"""Read frames until one has ``type == kind``.
|
|
|
|
The roster legitimately pushes ``stream_updated`` frames whenever the
|
|
audience changes, so tests must not assume a one-message-per-action cadence.
|
|
Any ``error`` frame seen on the way is a failure worth surfacing.
|
|
"""
|
|
seen = []
|
|
while True:
|
|
message = socket.receive_json()
|
|
if message.get("type") == kind:
|
|
return message
|
|
assert message.get("type") != "error", (
|
|
f"unexpected error while waiting for {kind}: {message}"
|
|
)
|
|
seen.append(message)
|
|
assert len(seen) < 25, f"never saw {kind!r}; got {seen}"
|
|
|
|
|
|
class TestPages:
|
|
@pytest.mark.parametrize(
|
|
("path", "marker"),
|
|
[
|
|
("/", "Watch everything"),
|
|
("/mirror", "Share my screen"),
|
|
("/share", "Share my screen"),
|
|
("/watch", "tile-template"),
|
|
],
|
|
)
|
|
def test_pages_render(self, client, path, marker):
|
|
response = client.get(path)
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/html")
|
|
assert marker in response.text
|
|
|
|
def test_pages_are_not_cacheable(self, client):
|
|
assert client.get("/watch").headers["cache-control"] == "no-store"
|
|
|
|
def test_the_two_pages_use_the_media_apis_they_claim(self):
|
|
share = (STATIC / "share.js").read_text()
|
|
watch = (STATIC / "watch.js").read_text()
|
|
assert "getDisplayMedia" in share
|
|
assert "RTCPeerConnection" in share and "RTCPeerConnection" in watch
|
|
assert "new WebSocket" in (STATIC / "net.js").read_text()
|
|
|
|
def test_nothing_is_loaded_from_a_cdn(self):
|
|
"""The no-CDN rule is a hard product requirement, so assert it."""
|
|
offenders = []
|
|
for file in sorted(STATIC.iterdir()):
|
|
text = file.read_text()
|
|
if EXTERNAL_REF.search(text):
|
|
offenders.append(file.name)
|
|
if "@import" in text or "fonts.googleapis" in text:
|
|
offenders.append(f"{file.name} (@import)")
|
|
assert offenders == []
|
|
|
|
@pytest.mark.parametrize("asset", ["app.css", "net.js", "share.js", "watch.js"])
|
|
def test_static_assets_are_served(self, client, asset):
|
|
assert client.get(f"/static/{asset}").status_code == 200
|
|
|
|
def test_unknown_page_is_a_404(self, client):
|
|
assert client.get("/nope").status_code == 404
|
|
|
|
|
|
class TestClientScripts:
|
|
"""Guards for the hand-written ES modules: they must parse, and the ids they
|
|
reach for must exist. Both of these have already bitten once."""
|
|
|
|
NODE = shutil.which("node")
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is not installed")
|
|
@pytest.mark.parametrize("script", ["net.js", "share.js", "watch.js"])
|
|
def test_scripts_parse_as_es_modules(self, script, tmp_path):
|
|
copy = tmp_path / "subject.mjs"
|
|
copy.write_text((STATIC / script).read_text())
|
|
result = subprocess.run([self.NODE, "--check", str(copy)], capture_output=True, text=True)
|
|
assert result.returncode == 0, f"{script}: {result.stderr}"
|
|
|
|
@pytest.mark.parametrize(
|
|
("script", "page"),
|
|
[("share.js", "mirror.html"), ("watch.js", "watch.html")],
|
|
)
|
|
def test_every_element_the_script_reaches_for_exists(self, script, page):
|
|
html = (STATIC / page).read_text()
|
|
wanted = set(re.findall(r"""\$\(\"#([A-Za-z0-9_-]+)\"\)""", (STATIC / script).read_text()))
|
|
missing = {name for name in wanted if f'id="{name}"' not in html}
|
|
assert not missing, f"{script} looks up {sorted(missing)} which {page} does not render"
|
|
|
|
def test_both_pages_load_the_stylesheet_and_their_module(self):
|
|
for page, script in (("mirror.html", "share.js"), ("watch.html", "watch.js")):
|
|
html = (STATIC / page).read_text()
|
|
assert 'href="/static/app.css"' in html
|
|
assert f'src="/static/{script}"' in html
|
|
assert 'type="module"' in html
|
|
|
|
def test_share_never_asks_for_an_exact_capture_size(self):
|
|
"""A hard size breaks pickers: `exact` fails or hides picker entries on
|
|
displays/DPRs the browser cannot honour. The size request must stay a
|
|
preference the browser can decline while still capturing."""
|
|
source = (STATIC / "share.js").read_text()
|
|
assert not re.search(r"\bexact\s*:", source), (
|
|
"share.js must not use an `exact` constraint key"
|
|
)
|
|
|
|
def test_server_and_client_agree_on_the_bitrate_default(self):
|
|
"""`/api/config` advertises the server's per-leg cap; a browser that cannot
|
|
reach it falls back to the hardcoded literal in `net.js`. If the two drift,
|
|
offline clients silently get a different picture quality and nothing fails.
|
|
One number, two places, one test that notices."""
|
|
server = int(
|
|
re.search(
|
|
r"DEFAULT_MAX_BITRATE_BPS\s*=\s*(\d[\d_]*)",
|
|
(STATIC.parent / "config.py").read_text(),
|
|
)
|
|
.group(1)
|
|
.replace("_", "")
|
|
)
|
|
client = int(re.search(r"maxBitrateBps:\s*(\d+)", (STATIC / "net.js").read_text()).group(1))
|
|
assert server == client, (
|
|
f"the server default ({server}) and the net.js fallback ({client}) "
|
|
"have drifted: a client that cannot reach /api/config would use "
|
|
"a different bitrate silently"
|
|
)
|
|
|
|
def test_sender_quality_is_reapplied_on_negotiation_and_connection(self):
|
|
"""`setParameters` is per-negotiation: a re-offer (a surface switch,
|
|
a viewer that re-joined) resets the encodings. The cap must therefore
|
|
be re-applied after `setLocalDescription` *and* when the leg becomes
|
|
`connected`. Both call sites are asserted so a future edit cannot drop
|
|
one and silently soften the wall."""
|
|
source = (STATIC / "share.js").read_text()
|
|
assert source.count('"maintain-resolution"') == 1, (
|
|
"the degradation preference must be a single, named decision"
|
|
)
|
|
assert "async function applySenderQuality" in source
|
|
|
|
def section(start: str, stop: str) -> str:
|
|
begin = source.index(start)
|
|
return source[begin : source.index(stop, begin)]
|
|
|
|
negotiate = section("async function negotiate(viewerId)", "async function onSignal")
|
|
assert "applySenderQuality" in negotiate, (
|
|
"the cap is not re-applied after setLocalDescription"
|
|
)
|
|
assert negotiate.index("setLocalDescription") < negotiate.index("applySenderQuality"), (
|
|
"the cap is applied before the encodings exist"
|
|
)
|
|
|
|
connected = section(
|
|
"pc.onconnectionstatechange = () => {",
|
|
"for (const track of state.stream.getTracks())",
|
|
)
|
|
assert re.search(
|
|
r'connectionState\s*===\s*"connected".*applySenderQuality', connected, re.S
|
|
), "the cap is not re-applied when the leg becomes connected"
|
|
|
|
def test_the_two_pages_share_no_element_ids_but_the_shared_chrome(self):
|
|
"""Both pages consume one stylesheet, so a duplicated id means a colliding rule.
|
|
|
|
This is not hypothetical: a `#stage` on both pages once made the share
|
|
page's preview inherit `display: none` from the watch page's overlay.
|
|
"""
|
|
shared = {"net", "notice", "toast"}
|
|
ids = {
|
|
page: set(re.findall(r'id="([A-Za-z0-9_-]+)"', (STATIC / page).read_text()))
|
|
for page in ("mirror.html", "watch.html", "index.html")
|
|
}
|
|
assert ids["mirror.html"] & ids["watch.html"] == shared
|
|
assert ids["index.html"] & ids["mirror.html"] <= shared | {"livebar"}
|
|
for page, found in ids.items():
|
|
assert len(found) == len(set(found)), f"{page} declares an id twice"
|
|
|
|
def test_videos_are_muted_and_inline_so_autoplay_is_not_blocked(self):
|
|
for page in ("mirror.html", "watch.html"):
|
|
for tag in re.findall(r"<video[^>]*>", (STATIC / page).read_text()):
|
|
assert "muted" in tag, f"{page}: {tag} would be blocked by autoplay policy"
|
|
assert "playsinline" in tag, f"{page}: {tag} would hijack iOS into fullscreen"
|
|
|
|
|
|
class TestStylesheet:
|
|
def test_hidden_always_means_gone(self):
|
|
"""Every component that sets its own `display` outranks the UA's `[hidden]`.
|
|
|
|
Without this rule the share page shows the LIVE badge and the Stop button
|
|
while nothing is being shared, and every wall tile offers an Exit button
|
|
it is not on stage for.
|
|
"""
|
|
css = (STATIC / "app.css").read_text()
|
|
assert re.search(r"\[hidden\]\s*\{[^}]*display:\s*none\s*!important", css), (
|
|
"app.css must declare a global `[hidden] { display: none !important }`"
|
|
)
|
|
# Components that set `display` and are toggled with `hidden` in JS.
|
|
for selector in (".btn", ".livetag", ".icon-btn", ".toast", ".net"):
|
|
assert selector in css, f"{selector} is toggled by `hidden`; it lives in app.css"
|
|
|
|
def test_no_rule_reintroduces_a_conflicting_display_for_hidden(self):
|
|
"""Every rule whose selector mentions `[hidden]` must actually hide."""
|
|
css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
|
|
offenders = [
|
|
selector.strip()
|
|
for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)
|
|
if "[hidden]" in selector and not re.search(r"display\s*:\s*none", body)
|
|
]
|
|
assert offenders == []
|
|
|
|
def test_a_hidden_control_is_not_revealed_only_by_hover(self):
|
|
"""Kiosk rule: a control hidden by default must be reachable without a pointer.
|
|
|
|
The tile's controls (`.actions`) start at `opacity: 0`. If the only rule
|
|
that raises them to full opacity were a `:hover` rule, a mouseless kiosk
|
|
would have no mute/expand controls at all. So while any `.actions` rule
|
|
hides them, the stylesheet must also contain a `.actions` reveal rule
|
|
with a non-hover trigger — keyboard focus or the selection state.
|
|
"""
|
|
css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
|
|
actions_rules = [
|
|
(selector.strip(), body)
|
|
for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)
|
|
if ".actions" in selector
|
|
]
|
|
hiding = [
|
|
selector
|
|
for selector, body in actions_rules
|
|
if re.search(r"opacity\s*:\s*0(?![0-9])", body)
|
|
]
|
|
non_hover_reveals = [
|
|
selector
|
|
for selector, body in actions_rules
|
|
if re.search(r"opacity\s*:\s*1(?![0-9])", body)
|
|
and re.search(r":focus-visible|:focus-within|data-selected", selector)
|
|
]
|
|
if hiding:
|
|
assert non_hover_reveals, (
|
|
f".actions starts hidden ({sorted(hiding)}) but no rule reveals it "
|
|
"without a pointer — a mouseless kiosk would have no tile controls"
|
|
)
|
|
|
|
|
|
class TestAddressGate:
|
|
"""mirror has no accounts; the address is the access control."""
|
|
|
|
ALLOW = "10.2.0.0/22"
|
|
|
|
def test_unset_allows_everyone_which_is_the_development_default(self):
|
|
with gated(client_address="203.0.113.9") as client:
|
|
assert client.get("/watch").status_code == 200
|
|
|
|
@pytest.mark.parametrize("path", ["/", "/mirror", "/watch", "/api/streams", "/static/app.css"])
|
|
def test_a_refused_address_gets_no_page_no_roster_no_assets(self, path):
|
|
with gated(self.ALLOW) as client:
|
|
assert client.get(path).status_code == 403
|
|
|
|
def test_the_probe_stays_answerable_from_anywhere(self):
|
|
"""The container health check comes from loopback, which the list excludes."""
|
|
with gated(self.ALLOW) as client:
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
def test_a_browser_gets_an_explanation_rather_than_json(self):
|
|
with gated(self.ALLOW) as client:
|
|
response = client.get("/mirror", headers={"accept": "text/html"})
|
|
assert response.status_code == 403
|
|
assert "text/html" in response.headers["content-type"]
|
|
body = response.text
|
|
assert self.ALLOW in body
|
|
assert "203.0.113.9" in body # which address was judged, or this is undiagnosable
|
|
|
|
def test_the_refusal_page_cannot_depend_on_assets_the_client_cannot_fetch(self):
|
|
with gated(self.ALLOW) as client:
|
|
body = client.get("/", headers={"accept": "text/html"}).text
|
|
assert not EXTERNAL_REF.search(body)
|
|
assert "/static/" not in body
|
|
|
|
def test_a_fetch_gets_the_machine_readable_refusal(self):
|
|
with gated(self.ALLOW) as client:
|
|
response = client.get("/api/config")
|
|
assert response.status_code == 403
|
|
assert response.json() == {
|
|
"error": "forbidden",
|
|
"message": "This mirror only accepts connections from 10.2.0.0/22.",
|
|
"you": "203.0.113.9",
|
|
}
|
|
|
|
def test_a_refusal_is_never_cacheable(self):
|
|
with gated(self.ALLOW) as client:
|
|
for headers in ({"accept": "text/html"}, {}):
|
|
response = client.get("/api/config", headers=headers)
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
def test_the_socket_closes_before_it_accepts(self):
|
|
with (
|
|
gated(self.ALLOW) as client,
|
|
pytest.raises(WebSocketDisconnect),
|
|
client.websocket_connect("/ws") as socket,
|
|
):
|
|
socket.send_json({"type": "hello", "role": "viewer"})
|
|
|
|
def test_the_real_address_is_in_and_connects(self):
|
|
with gated(self.ALLOW, client_address="10.2.1.10") as client:
|
|
assert client.get("/watch").status_code == 200
|
|
with client.websocket_connect("/ws") as socket:
|
|
assert handshake(socket, "viewer")
|
|
|
|
def test_a_trusted_local_proxy_carries_the_client_through(self):
|
|
"""The fleet shape: nginx on the same host, X-Forwarded-For set by nginx."""
|
|
with gated(self.ALLOW, client_address="127.0.0.1") as client:
|
|
assert client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"}).status_code == 200
|
|
with client.websocket_connect("/ws", headers={"x-forwarded-for": "10.2.1.5"}) as socket:
|
|
assert handshake(socket, "viewer")
|
|
|
|
def test_a_trusted_proxy_carrying_a_foreign_client_is_still_refused(self):
|
|
with gated(self.ALLOW, client_address="127.0.0.1") as client:
|
|
assert client.get("/watch", headers={"x-forwarded-for": "8.8.8.8"}).status_code == 403
|
|
with (
|
|
pytest.raises(WebSocketDisconnect),
|
|
client.websocket_connect("/ws", headers={"x-forwarded-for": "8.8.8.8"}),
|
|
):
|
|
pass
|
|
|
|
def test_an_untrusted_peer_cannot_write_its_own_pass(self):
|
|
with gated(self.ALLOW, client_address="203.0.113.9") as client:
|
|
response = client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"})
|
|
assert response.status_code == 403
|
|
assert response.json()["you"] == "203.0.113.9"
|
|
|
|
def test_both_address_families_can_be_admitted_side_by_side(self):
|
|
with gated(f"{self.ALLOW}, fd00::/64", client_address="fd00::1234") as client:
|
|
assert client.get("/watch").status_code == 200
|
|
|
|
|
|
class TestJsonApi:
|
|
def test_config_advertises_the_media_defaults(self, client):
|
|
body = client.get("/api/config").json()
|
|
assert body["iceServers"] == []
|
|
assert body["maxBitrateBps"] == 2_000_000
|
|
assert "version" in body
|
|
|
|
def test_config_is_a_whitelist(self, client):
|
|
body = client.get("/api/config").json()
|
|
assert set(body) == {"version", "iceServers", "maxBitrateBps", "pingIntervalMs"}
|
|
|
|
def test_streams_starts_empty(self, client):
|
|
body = client.get("/api/streams").json()
|
|
assert body["mirrors"] == []
|
|
assert body["stats"]["mirrors"] == 0
|
|
|
|
def test_healthz(self, client):
|
|
assert client.get("/healthz").json()["status"] == "ok"
|
|
|
|
|
|
class TestSignalingSocket:
|
|
def test_hello_must_come_first(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_json({"type": "publish", "label": "sneaky"})
|
|
message = socket.receive_json()
|
|
assert message["type"] == "error"
|
|
assert message["code"] == "hello_required"
|
|
assert message["fatal"] is True
|
|
|
|
def test_malformed_json_is_rejected(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_text("{not json")
|
|
assert socket.receive_json()["code"] == "bad_json"
|
|
|
|
def test_a_non_object_frame_is_rejected(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_text("[1, 2, 3]")
|
|
assert socket.receive_json()["code"] == "bad_json"
|
|
|
|
def test_an_unknown_role_is_refused(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_json({"type": "hello", "role": "spectator"})
|
|
assert socket.receive_json()["code"] == "bad_role"
|
|
|
|
def test_unknown_message_type_is_reported_but_survivable(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
handshake(socket, "viewer", "V")
|
|
socket.send_json({"type": "telepathy"})
|
|
assert socket.receive_json()["code"] == "bad_message"
|
|
socket.send_json({"type": "ping"})
|
|
assert socket.receive_json()["type"] == "pong"
|
|
|
|
def test_ping_roundtrips(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
handshake(socket, "viewer")
|
|
socket.send_json({"type": "ping"})
|
|
assert socket.receive_json() == {"type": "pong"}
|
|
|
|
def test_an_oversized_frame_is_refused(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
handshake(socket, "viewer")
|
|
socket.send_text(json.dumps({"type": "ping", "padding": "x" * 70_000}))
|
|
assert socket.receive_json()["code"] == "too_large"
|
|
|
|
def test_roster_flows_from_sharer_to_viewer(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "Linux Box")
|
|
viewer_id = handshake(viewer, "viewer", "Phone")
|
|
|
|
sharer.send_json({"type": "publish", "label": "Terminal", "width": 1280, "height": 720})
|
|
added = wait_for(viewer, "stream_added")["stream"]
|
|
assert added["id"] == sharer_id
|
|
assert added["label"] == "Terminal"
|
|
assert added["name"] == "Linux Box"
|
|
assert added["width"] == 1280
|
|
|
|
viewer.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
joined = wait_for(sharer, "peer_joined")
|
|
assert joined["viewerId"] == viewer_id
|
|
assert joined["name"] == "Phone"
|
|
|
|
sharer.send_json({"type": "signal", "to": viewer_id, "payload": {"sdp": {"s": "o"}}})
|
|
relayed = wait_for(viewer, "signal")
|
|
assert relayed["from"] == sharer_id
|
|
assert relayed["payload"] == {"sdp": {"s": "o"}}
|
|
|
|
viewer.send_json({"type": "signal", "to": sharer_id, "payload": {"c": "ice"}})
|
|
assert wait_for(sharer, "signal")["payload"] == {"c": "ice"}
|
|
|
|
def test_capture_metadata_survives_the_wire(self, client):
|
|
"""Regression: `displaySurface` once arrived as "" because the model expected
|
|
snake_case and unknown keys were being ignored."""
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json(
|
|
{
|
|
"type": "publish",
|
|
"label": "proj.gl — zsh",
|
|
"displaySurface": "window",
|
|
"width": 1920,
|
|
"height": 1080,
|
|
"audio": True,
|
|
}
|
|
)
|
|
stream = wait_for(viewer, "stream_added")["stream"]
|
|
assert stream["displaySurface"] == "window"
|
|
assert stream["audio"] is True
|
|
assert (stream["width"], stream["height"]) == (1920, 1080)
|
|
assert client.get("/api/streams").json()["mirrors"][0]["displaySurface"] == "window"
|
|
|
|
def test_both_viewers_of_one_mirror_see_their_own_leg(self, client):
|
|
with (
|
|
client.websocket_connect("/ws") as sharer,
|
|
client.websocket_connect("/ws") as one,
|
|
client.websocket_connect("/ws") as two,
|
|
):
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
first = handshake(one, "viewer", "one")
|
|
second = handshake(two, "viewer", "two")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(one, "stream_added")
|
|
wait_for(two, "stream_added")
|
|
|
|
one.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
two.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
assert wait_for(sharer, "peer_joined")["viewerId"] == first
|
|
assert wait_for(sharer, "peer_joined")["viewerId"] == second
|
|
|
|
sharer.send_json({"type": "signal", "to": first, "payload": {"n": 1}})
|
|
sharer.send_json({"type": "signal", "to": second, "payload": {"n": 2}})
|
|
assert wait_for(one, "signal")["payload"] == {"n": 1}
|
|
assert wait_for(two, "signal")["payload"] == {"n": 2}
|
|
|
|
def test_a_mirror_disappears_for_viewers_when_the_sharer_leaves(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer", "V")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
wait_for(sharer, "peer_joined")
|
|
|
|
sharer.close()
|
|
assert wait_for(viewer, "stream_removed")["streamId"] == sharer_id
|
|
|
|
def test_unpublish_removes_the_mirror(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
sharer.send_json({"type": "unpublish"})
|
|
assert wait_for(viewer, "stream_removed")["streamId"] == sharer_id
|
|
assert client.get("/api/streams").json()["mirrors"] == []
|
|
|
|
def test_signalling_a_peer_that_never_subscribed_is_refused(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer")
|
|
viewer_id = handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
assert wait_for(viewer, "stream_added")["stream"]["sharerId"] == sharer_id
|
|
# The viewer knows the mirror exists but has not subscribed to it.
|
|
sharer.send_json({"type": "signal", "to": viewer_id, "payload": {}})
|
|
assert wait_for(sharer, "error")["code"] == "not_subscribed"
|
|
|
|
def test_a_viewer_cannot_reach_a_sharer_it_never_subscribed_to(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "signal", "to": sharer_id, "payload": {}})
|
|
assert wait_for(viewer, "error")["code"] == "not_subscribed"
|
|
|
|
def test_subscribing_to_a_mirror_that_is_not_there_reports_cleanly(self, client):
|
|
with client.websocket_connect("/ws") as viewer:
|
|
handshake(viewer, "viewer")
|
|
viewer.send_json({"type": "subscribe", "streamId": "p404"})
|
|
message = wait_for(viewer, "error")
|
|
assert message["code"] == "unknown_stream"
|
|
assert message["fatal"] is False
|
|
|
|
def test_editing_the_display_name_reaches_viewers(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "Old name")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
sharer.send_json({"type": "hello", "role": "sharer", "name": "Design Laptop"})
|
|
assert wait_for(viewer, "stream_updated")["stream"]["name"] == "Design Laptop"
|
|
assert client.get("/api/streams").json()["mirrors"][0]["sharerId"] == sharer_id
|
|
|
|
def test_stats_track_the_fan_out(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer", "V")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
wait_for(sharer, "peer_joined")
|
|
stats = client.get("/api/streams").json()["stats"]
|
|
assert stats["mirrors"] == 1
|
|
assert stats["watching"] == 1
|
|
assert stats["connections"] == 2
|
|
|
|
def test_one_peer_cannot_impersonate_another(self, client):
|
|
"""`to` is addressed by peer id; a viewer must not be able to spoof a sharer."""
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer", "V")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "publish", "label": "forged"})
|
|
assert wait_for(viewer, "error")["code"] == "not_sharer"
|
|
assert client.get("/api/streams").json()["stats"]["mirrors"] == 1
|
|
assert sharer_id and viewer
|