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`.
473 lines
21 KiB
Python
473 lines
21 KiB
Python
"""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_the_screen_sized_request_does_not_break_the_fake_device(self, device):
|
|
"""Asking for screen x DPR (capped 3840x2160) must not break the fake
|
|
device's path.
|
|
|
|
The old "always 1280x720" expectation held only while the request
|
|
carried no size. Once a size is *named*, the fake UI clamps the capture
|
|
to the fake browser window (1100x720 context -> 1100x620 with chrome)
|
|
— PLAN §5's window scaling, even for the synthetic device. So the
|
|
assertion here is: the share starts, the viewer decodes exactly what
|
|
the sharer's delivered settings are, moving, and publish() reports
|
|
those delivered settings, not the request.
|
|
"""
|
|
errors: list = []
|
|
sharer = device("/mirror", errors)
|
|
start_sharing(sharer)
|
|
|
|
# What the page asked for vs what the fake device delivered.
|
|
reported = sharer.evaluate(
|
|
"""() => {
|
|
const track = document.querySelector('#preview').srcObject.getVideoTracks()[0];
|
|
const s = track.getSettings();
|
|
const dpr = window.devicePixelRatio || 1;
|
|
return {
|
|
requested: {
|
|
width: Math.min(Math.round(screen.width * dpr), 3840),
|
|
height: Math.min(Math.round(screen.height * dpr), 2160),
|
|
},
|
|
delivered: {width: s.width, height: s.height},
|
|
stat: document.querySelector('#stat-resolution').textContent,
|
|
};
|
|
}"""
|
|
)
|
|
assert reported["requested"] == {"width": 1100, "height": 720}, reported
|
|
# The full requested width is honoured; the height is clamped to the
|
|
# fake window's visible area (the 100px of chrome).
|
|
assert reported["delivered"]["width"] == 1100, reported
|
|
assert 0 < reported["delivered"]["height"] < 720, reported
|
|
|
|
# publish() reports the delivered settings, not the request.
|
|
# The page renders the separator as U+00D7 (multiplication sign).
|
|
w, h = reported["delivered"]["width"], reported["delivered"]["height"]
|
|
assert reported["stat"] == f"{w} \u00d7 {h}", reported
|
|
|
|
viewer = device("/watch", errors)
|
|
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
|
|
frames = frames_matching(viewer, ".tile video")
|
|
assert (frames["width"], frames["height"]) == (w, h), (
|
|
f"the viewer must decode what the sharer reported: {frames} vs {reported}"
|
|
)
|
|
assert frames["distinct"] >= 2, f"the video is frozen: {frames}"
|
|
assert errors == [], f"uncaught browser errors: {errors}"
|
|
|
|
def test_the_fake_device_keeps_exactly_1280x720(self, base_url, browser, request, device):
|
|
"""The fake-device counterpart of the realdisplay sharpness floor.
|
|
|
|
With the sharer's window filling the fake 1280x720 monitor (a 1280x820
|
|
viewport: the ~100px of window chrome clamps the 820 request back to
|
|
720), the deterministic device delivers exactly 1280x720 — the classic
|
|
"always 1280x720" synthetic path. The phase 02 ``ideal`` size
|
|
constraints (screen x DPR, capped 3840x2160) must not change what the
|
|
synthetic path delivers: the viewer decodes exactly 1280x720, moving.
|
|
|
|
Measured on the reference host (phase 02, 2026-11-01): a 1280x820
|
|
sharer window requests 1280x820 and the fake device delivers exactly
|
|
1280x720; the viewer decodes 1280x720. (A 1100x720 window instead
|
|
delivers 1100x620 — the same window clamp, covered by the sibling
|
|
test.)
|
|
"""
|
|
errors: list = []
|
|
context = browser.new_context(viewport={"width": 1280, "height": 820})
|
|
request.addfinalizer(context.close)
|
|
sharer = context.new_page()
|
|
sharer.on("pageerror", lambda error: errors.append(str(error)))
|
|
sharer.goto(f"{base_url}/mirror", wait_until="domcontentloaded")
|
|
expect(sharer.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
|
|
start_sharing(sharer)
|
|
|
|
delivered = sharer.evaluate(
|
|
"""() => {
|
|
const track = document.querySelector('#preview').srcObject.getVideoTracks()[0];
|
|
const s = track.getSettings();
|
|
return {width: s.width, height: s.height};
|
|
}"""
|
|
)
|
|
assert delivered == {"width": 1280, "height": 720}, delivered
|
|
# The page's own stats agree with the delivered settings.
|
|
expect(sharer.locator("#stat-resolution")).to_have_text("1280 \u00d7 720")
|
|
|
|
viewer = device("/watch", errors)
|
|
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
|
|
frames = frames_matching(viewer, ".tile video")
|
|
assert (frames["width"], frames["height"]) == (1280, 720), (
|
|
f"the fake device must keep delivering exactly 1280x720: {frames}"
|
|
)
|
|
assert frames["distinct"] >= 2, f"the video is frozen: {frames}"
|
|
assert errors == [], f"uncaught browser errors: {errors}"
|
|
|
|
def test_the_staged_tile_bytes_keep_growing(self, base_url, browser, request, device):
|
|
"""Inbound-bitrate sanity on a staged tile: an observation, not an equality.
|
|
|
|
The viewer's ``inbound-rtp.bytesReceived`` must grow while a share is
|
|
live. With the default 25 Mbit/s per-leg cap, a live 1280x720 fake
|
|
share delivered ~126 KB over a ~2 s sample (≈ 0.5 Mbit/s) on the
|
|
reference host (phase 02, 2026-11-01) — the figure is
|
|
environment-dependent (codec, motion, cap), so the assertion is growth
|
|
from a non-zero start, never a specific number.
|
|
"""
|
|
context = browser.new_context(viewport={"width": 1100, "height": 720})
|
|
request.addfinalizer(context.close)
|
|
# Test scaffolding (AGENTS.md §6): collect the RTCPeerConnection
|
|
# instances watch.js creates so the test can read the viewer's
|
|
# inbound-rtp stats through the one allowed global, ``window.__probe``.
|
|
context.add_init_script(
|
|
"""
|
|
window.__probe = {pcs: []};
|
|
const RealPC = window.RTCPeerConnection;
|
|
window.RTCPeerConnection = function (config) {
|
|
const pc = new RealPC(config);
|
|
window.__probe.pcs.push(pc);
|
|
return pc;
|
|
};
|
|
window.RTCPeerConnection.prototype = RealPC.prototype;
|
|
window.RTCPeerConnection.static = RealPC.static;
|
|
"""
|
|
)
|
|
page = context.new_page()
|
|
page.goto(f"{base_url}/watch", wait_until="domcontentloaded")
|
|
expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
|
|
|
|
sharer = device("/mirror")
|
|
start_sharing(sharer)
|
|
|
|
expect(page.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
|
|
# Take it fullscreen: the staged <video> is the same element moved onto
|
|
# the stage, so the peer connection — and its stats — survive.
|
|
page.locator(".tile").click()
|
|
expect(page.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
|
|
expect(page.locator("#watch-stage video")).to_have_count(1)
|
|
|
|
read_bytes = """async () => {
|
|
let total = 0;
|
|
for (const pc of window.__probe.pcs) {
|
|
try {
|
|
const stats = await pc.getStats();
|
|
stats.forEach((entry) => {
|
|
if (entry.type === 'inbound-rtp' && entry.kind === 'video') {
|
|
total += entry.bytesReceived || 0;
|
|
}
|
|
});
|
|
} catch { /* a closed pc has no stats; the live one does */ }
|
|
}
|
|
return total;
|
|
}"""
|
|
|
|
# Wait for the leg to actually be connected, then sample twice.
|
|
deadline = time.time() + 30
|
|
while time.time() < deadline:
|
|
states = page.evaluate("() => window.__probe.pcs.map(pc => pc.connectionState)")
|
|
if "connected" in states:
|
|
break
|
|
time.sleep(0.3)
|
|
first = page.evaluate(read_bytes)
|
|
assert first > 0, f"no bytes had arrived by the first sample: {first}"
|
|
time.sleep(2.0)
|
|
second = page.evaluate(read_bytes)
|
|
assert second > first, (
|
|
f"the viewer's inbound byte count did not grow while the share was "
|
|
f"live: {first} -> {second}"
|
|
)
|
|
|
|
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")
|
|
|
|
|
|
class TestAddressGate:
|
|
"""`--allow-cidr` in a real process, refused by a real browser.
|
|
|
|
``tests/test_app.py::TestAddressGate`` covers the JSON refusal the pages key
|
|
their own notice on, the proxy unwrapping, and the socket closing before it
|
|
accepts. This covers what a person actually gets when a server refuses them:
|
|
an explanation naming the address it judged, not an empty wall that looks
|
|
like a bug in the tool.
|
|
"""
|
|
|
|
@pytest.mark.parametrize("path", ["/mirror", "/watch"])
|
|
def test_a_refused_address_sees_why_and_none_of_the_app(self, browser, restricted_url, path):
|
|
context = browser.new_context(viewport={"width": 1100, "height": 720})
|
|
page = context.new_page()
|
|
try:
|
|
response = page.goto(f"{restricted_url}{path}")
|
|
assert response.status == 403
|
|
expect(page.locator("body")).to_contain_text("not open to your address")
|
|
expect(page.locator("body")).to_contain_text("203.0.113.0/24")
|
|
# Which address it judged — without this, "you are not allowed" in
|
|
# front of a proxy is undiagnosable.
|
|
expect(page.locator("body")).to_contain_text("127.0.0.1")
|
|
# None of the app shell leaked past the gate.
|
|
expect(page.locator("#start, #grid, #notice")).to_have_count(0)
|
|
finally:
|
|
context.close()
|
|
|
|
def test_the_health_probe_is_not_gated(self, restricted_url):
|
|
import httpx
|
|
|
|
assert httpx.get(f"{restricted_url}/healthz", timeout=5).status_code == 200
|