"""Phase 07 task 05 — the flagship playable suite: keyboard input, end to end. The first time the whole stack is proven by a single test family: a real browser, real key events (CDP `Input.dispatchKeyEvent` — trusted, the `window.__uc.input` module's WASD/arrow + Space/Enter bindings), a real live server on the `tests/fixtures/levels_us04/` campaign (the shipped level 1 layout, `target_score` 100 / `time_limit` 120 s — one served soup scores 44-66, so a single serve does *not* end the round and the HUD stays observable), and the test-only fixed-seed seam (`UNDERCOOKED_TEST_MODE=1` → the room is born with `TEST_MODE_SEED` = 20240701, so the page's `?seed=20240701` plays the seed the sim actually draws — `tests/e2e/README.md`). * **`test_moves_and_faces`** — a key press produces the expected position delta (the server-observed `app.view`, quantized to 2 decimals), the facing follows the movement (`f` 1 = right, 3 = left), the facing is sticky after a stop, and a diagonal is normalised on the client (the wire's magnitude never exceeds 1 — the `stats()` census reads the exact `1/√2`). * **`test_stops_at_walls`** — walking into a wall does not pass through (the body's centre settles at `wall ± 0.35`, the radius) and does not jitter (repeated reads are equal to the wire's quantization). * **`test_serves_one_soup`** — the full chain, keyboard-driven: the tomato off the spawn shelf, chop, pot, cook, plate, serve — the HUD score bumps and the order ticket disappears. (The campaign's recipe is kind-based in v1 — `sim/scoring.py`'s `_candidates`: the hand holds one item, a `tomato_plated` dish composes the `soup`, and the two-ingredient combine is a later phase's mechanic.) * **`test_denied_out_of_reach`** — an early press with no station in reach shows the deny feedback (the FX census's `denied`) and the item stays (the shelf's stock, the hand, both untouched). * **`test_two_players_same_target`** — two contexts press at the same marked set-down simultaneously; exactly one gets the item, the other sees the `denied` (a same-tick press is `contested_lost` — the straddle case, where the later tick's press finds an empty fixture, re-sets the down and retries), and the server's item count is conserved (in-process `room.sim_state`). * **`test_alt_tab_parks_the_cook`** — a hidden tab releases every held key and sends one zero-intent frame with the action released (the module's `stats()` census), and the server-observed position freezes (a short coast, then equal reads) — no cook running away. * **`test_readme_controls_match_keymap`** — the README's "Controls" block is exactly `getKeyMap()` (the task's completion criterion, enforced; the in-game help overlay renders the same rows). The harness is the phase-07 e2e machine (`conftest.py`): the `e2e_server` factory (one module-scoped test-mode server for the whole suite), fresh contexts per test, the `window.__undercooked` app seam (the `view` — the server-observed state the choreography steers against) and the `window.__uc` census seam (`input.stats()`, the FX's census). Steering is real: the helper holds WASD keys via `page.keyboard.down/up` on a ~60 ms cadence and reads the server's own view — no `sendInput` shortcuts in this suite. """ from __future__ import annotations import math import re import time from itertools import pairwise from pathlib import Path from typing import TYPE_CHECKING, Any, cast import pytest from conftest import app_screen, open_two_pages if TYPE_CHECKING: from playwright.sync_api import Browser, Page pytestmark = pytest.mark.e2e REPO_ROOT = Path(__file__).resolve().parents[2] #: The single-serve campaign (target 100: one soup is 44-66 — the HUD #: stays observable after the serve). US04_LEVELS_DIR: Path = REPO_ROOT / "tests" / "fixtures" / "levels_us04" #: The documented test-mode seed (the `?seed=` value the page loads #: with; `server/room.py`'s `TEST_MODE_SEED`). SEED = "20240701" #: The kitchen geometry (the campaign's grid): the floor spans x 1-11, #: y 1-9 in tile units, so a body of radius 0.35 stops at 1.35 / 10.65 #: (x) and 1.35 / 7.65 (y — the row-8 wall; row 7 is the last floor). WALL_X_MIN, WALL_X_MAX = 1.35, 10.65 WALL_Y_MIN, WALL_Y_MAX = 1.35, 7.65 #: The body radius (the `sim`'s 0.35 — the wall band's tolerance). WALL_BAND = 0.06 #: The phase-06 waypoints and staging points, verbatim (the approach #: leg decides the facing: a cook arrives facing the station). SHELF_STAGE = (2.5, 1.5) # facing LEFT at the shelf (1, 1) BOARD_STAGE = (3.5, 1.5) # facing RIGHT at the board (4, 1) STOVE_STAGE = (7.5, 1.5) # facing RIGHT at the stove (8, 1) PREP_STAGE = (3.5, 3.5) # facing LEFT at the prep (2, 3) SERVE_STAGE = (8.8, 3.5) # in front of the window (9, 3): a stop west of its # centre keeps the window ahead of the facing plane (`not_facing`) #: The server-observed "me" (the `app.view` — the raw merged snapshot; #: `pi` is the seat index, `f` the facing index, `h` the carried #: item's short form). ME_PROBE = """() => { const app = window.__undercooked; if (app.you === null) return null; const me = Object.values(app.view.players).find((p) => p.pi === app.you.seat); return me === undefined ? null : { x: me.x, y: me.y, f: me.f, h: me.h, s: me.s, pi: me.pi }; }""" #: The FX's census (the `denied` marker's consumption). FX_PROBE = "() => (window.__uc.fx !== null ? window.__uc.fx.stats() : null)" #: The module's input census (the held keys, the last frame sent). INPUT_PROBE = "() => window.__uc.input.stats()" #: The `?seed=` seam's page half (the app's own param, echoed once). SEED_PROBE = "() => window.__undercooked.seed ?? ''" def _seeded_url(url: str) -> str: return f"{url}/?seed={SEED}" def _join_ui(page: Page, code: str) -> None: """The keyboard-first join (the us03 machine).""" page.locator("#join-code").focus() page.keyboard.type(code) page.keyboard.press("Enter") def _start_solo(page: Page) -> str: """A single-player room: create, ready, start (the solo-host start, PLAN §6); the room code.""" page.evaluate("() => window.__undercooked.createLobby({ name: 'Ada' })") page.locator("#room-code").wait_for(state="visible", timeout=20_000) code = page.locator("#room-code").inner_text() page.get_by_role("button", name="Ready up").focus() page.keyboard.press("Enter") page.wait_for_function( "() => { const b = document.querySelector('#start-btn'); " "return b !== null && !b.disabled; }", timeout=20_000, ) page.get_by_role("button", name="Start").focus() page.keyboard.press("Enter") page.locator("#kcanvas").wait_for(state="attached", timeout=30_000) assert app_screen(page) == "playing" page.wait_for_timeout(600) # the first frame + the spawn settle return code def _start_duo(page_a: Page, page_b: Page) -> str: """The us03 lobby (both ready, the host starts); the room code.""" page_a.evaluate("() => window.__undercooked.createLobby({ name: 'Ada' })") page_a.locator("#room-code").wait_for(state="visible", timeout=20_000) code = page_a.locator("#room-code").inner_text() _join_ui(page_b, code) page_b.locator("#room-code").wait_for(state="visible", timeout=20_000) page_b.get_by_role("button", name="Ready up").focus() page_b.keyboard.press("Enter") start_enabled = ( "() => { const b = document.querySelector('#start-btn'); " "return b !== null && !b.disabled; }" ) page_a.wait_for_function(start_enabled, timeout=20_000) page_a.get_by_role("button", name="Start").focus() page_a.keyboard.press("Enter") page_a.locator("#kcanvas").wait_for(state="attached", timeout=30_000) page_b.locator("#kcanvas").wait_for(state="attached", timeout=30_000) assert app_screen(page_a) == "playing" assert app_screen(page_b) == "playing" page_a.wait_for_timeout(600) page_b.wait_for_timeout(600) return code def _walk(page: Page, waypoints: list[tuple[float, float]], band: float = 0.45, timeout_s: float = 90.0) -> None: """Steer the real keyboard to the waypoints (the ~60 ms cadence: the key set is the dead-reckoning of the server-observed position — the dead zone (0.15 per axis) is inside the band, so a stop always lands in it; the deceleration tail is the settle).""" for index, (tx, ty) in enumerate(waypoints): target_band = band if index == len(waypoints) - 1 else 0.3 deadline = time.monotonic() + timeout_s held: set[str] = set() while True: me = page.evaluate(ME_PROBE) if me is None: msg = "the cook vanished from the server's view" raise AssertionError(msg) dx = tx - me["x"] dy = ty - me["y"] want: set[str] = set() if abs(dx) > 0.15: want.add("d" if dx > 0 else "a") if abs(dy) > 0.15: want.add("s" if dy > 0 else "w") for key in sorted(held - want): page.keyboard.up(key) for key in sorted(want - held): page.keyboard.down(key) held = want if math.hypot(dx, dy) <= target_band: break if time.monotonic() > deadline: msg = f"timed out steering to ({tx}, {ty}) from ({me['x']}, {me['y']})" raise AssertionError(msg) page.wait_for_timeout(60) for key in sorted(held): page.keyboard.up(key) page.wait_for_timeout(300) # the deceleration tail (the sim brakes on zero) def _press(page: Page, hold_ms: int = 120) -> None: """One action-button press (a rising edge: down, a beat, up).""" page.keyboard.down("Space") page.wait_for_timeout(hold_ms) page.keyboard.up("Space") def _face(page: Page, key: str) -> None: """A short nudge in the `key` direction: it re-sets the sim's facing deterministically (a walk-band stop can land facing a diagonal tail — the station's front tile must be the facing one) and keeps the body inside the stage band (a ~0.3 nudge).""" page.keyboard.down(key) page.wait_for_timeout(100) page.keyboard.up(key) page.wait_for_timeout(250) # the deceleration tail + the facing settle #: The wire's facing index per nudge key (`f` on the player entry). FACING_INDEX = {"a": 3, "d": 1, "w": 0, "s": 2} def _aim(page: Page, face_key: str) -> None: """Re-set the `face_key` facing only when the wire says it is off (a blind nudge moves the body ~0.36 and can step it across a cell boundary — off the station's candidate grid, where a press is a `nothing_here` no-op).""" me = page.evaluate(ME_PROBE) if me is None or me["f"] == FACING_INDEX[face_key]: return _face(page, face_key) def _sim_item(room: Any, seat: int) -> str | None: """The item the sim's `seat` carries, as `kind_form` (the wire's `h` shape), or None (an empty hand). The in-process sim is the ground truth: the lerp'd view lags the sim ~100 ms and can sit stale under snapshot backpressure (two browser contexts on one room), so a view-based wait can pass on a frame the sim has not applied yet.""" carrying = room.sim_state.players[seat].carrying if carrying is None: return None return f"{carrying.kind.value}_{carrying.form.value}" def _sim_wait_item( room: Any, seat: int, expected: str | None, timeout_s: float = 8.0 ) -> None: """Poll the sim until `seat` carries exactly `expected` (None = the empty hand) — the ground-truth replacement for the view's `h` waits.""" deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: if _sim_item(room, seat) == expected: return time.sleep(0.02) msg = f"the sim never carried {expected!r} at seat {seat}" raise AssertionError(msg) def _sim_settled( room: Any, seat: int, xy: tuple[float, float], tol: float = 0.45, timeout_s: float = 20.0, ) -> None: """Poll the sim until `seat` is inside `tol` of `xy` and *at rest* (three consecutive samples ~150 ms: a body in flight passes the band before its deceleration tail stops it — a press sent while it moves resolves off the candidate grid).""" deadline = time.monotonic() + timeout_s still = 0 while time.monotonic() < deadline: pos = room.sim_state.players[seat].pos if math.hypot(pos.x - xy[0], pos.y - xy[1]) <= tol: still += 1 if still >= 3: return else: still = 0 time.sleep(0.05) pos = room.sim_state.players[seat].pos msg = f"the sim never settled seat {seat} near {xy} (at {pos.x:.2f}, {pos.y:.2f})" raise AssertionError(msg) def _pick( page: Page, room: Any, seat: int, face_key: str, short: str, *, tries: int = 3 ) -> None: """A press retried until the sim (the ground truth) carries `short` (a press that lands while the facing is off the station is a silent no-op — the next nudge re-aims and re-presses).""" for _ in range(tries): _aim(page, face_key) _press(page) try: _sim_wait_item(room, seat, short, timeout_s=5.0) return except AssertionError: sim = room.sim_state.players[seat] denied = [ (e.payload.get("reason"), e.payload.get("player_id")) for _, e in room.event_buffer[-12:] if e.kind.value == "denied" ] print( f"[pick {short}] miss: sim=({sim.pos.x:.2f},{sim.pos.y:.2f})" f" f={sim.facing.value if hasattr(sim.facing, 'value') else sim.facing}" f" denied={denied[-3:] if denied else None}" ) continue # the press missed — re-aim, re-press msg = f"the pick of {short!r} never landed after {tries} presses" raise AssertionError(msg) def _take(page: Page, room: Any, seat: int, short: str, tries: int = 3) -> None: """The pick after a hold, retried (a follow-up press that lands off-facing finds the station still `ready` — a second press picks; the hold's release already happened).""" for _ in range(tries): try: _sim_wait_item(room, seat, short, timeout_s=3.0) return except AssertionError: _press(page) # the station is still `ready` — a fresh press picks msg = f"never carried {short!r} after {tries} presses" raise AssertionError(msg) def _set_down( page: Page, room: Any, seat: int, face_key: str = "a", tries: int = 3 ) -> None: """A set-down press, retried until the sim (the ground truth) shows the empty hand (a press that lands off-facing leaves the item in the hand — the nudge re-aims and the press retries).""" for _ in range(tries): _aim(page, face_key) _press(page) try: _sim_wait_item(room, seat, None, timeout_s=5.0) return except AssertionError: continue # the item is still in the hand — re-aim, re-press me = page.evaluate(ME_PROBE) denied = [ev for ev in _events(page) if ev.get("k") == "denied"] msg = ( f"the set-down never landed after {tries} presses " f"(me={me}, last_denial={denied[-1] if denied else None})" ) raise AssertionError(msg) def _serve(page: Page, timeout_s: float = 45.0) -> int: """The window press, retried until the HUD's score bumps (a denied serve keeps the dish in the hand — a re-press is a fresh rising edge; the denial's named reason rides the event's `r` and is reported on the final failure).""" deadline = time.monotonic() + timeout_s last_denial: dict | None = None for _ in range(4): _press(page) while time.monotonic() < deadline: score = page.evaluate( "() => Number(document.querySelector('#hud-score')" ".textContent || 0)" ) if score > 0: return int(score) denied = [ev for ev in _events(page) if ev.get("k") == "denied"] if denied: last_denial = denied[-1] page.wait_for_timeout(150) msg = f"the serve never scored (last denial: {last_denial})" raise AssertionError(msg) def _station_ready_js(kind: str) -> str: return ( f"() => {{ const app = window.__undercooked; " f"const st = Object.values(app.view.stations).find((s) => s.k === '{kind}'); " "return st !== undefined && st.s === 'ready'; }" ) def _hold_until(page: Page, predicate_js: str, timeout_s: float = 90.0) -> None: """Hold the action (the module's ~48 ms resend keeps the hold continuous above the 20 Hz tick) until `predicate_js`, then the release + one press (the pick's rising edge). A hold that has run 10 s without the predicate is re-pressed (the press may have landed off-facing — the station never started; the release + keydown is a fresh rising edge).""" deadline = time.monotonic() + timeout_s attempt_deadline = time.monotonic() + 10 page.keyboard.down("Space") try: while not page.evaluate(predicate_js): if time.monotonic() > deadline: me = page.evaluate(ME_PROBE) st = page.evaluate( "() => Object.values(window.__undercooked.view.stations)" ".map((s) => s.k + ':' + s.s + ':' + s.g)" ) msg = ( f"timed out holding the action ({predicate_js}) me={me} " f"stations={st} last_events={_events(page)[-8:]}" ) raise AssertionError(msg) if time.monotonic() > attempt_deadline: page.keyboard.up("Space") page.wait_for_timeout(150) page.keyboard.down("Space") attempt_deadline = time.monotonic() + 10 page.wait_for_timeout(40) finally: page.keyboard.up("Space") page.wait_for_timeout(150) # the release frame lands before the press _press(page) def _wait_carrying(page: Page, short: str, timeout_s: float = 30.0) -> None: deadline = time.monotonic() + timeout_s while True: me = page.evaluate(ME_PROBE) if me is not None and me["h"] == short: return if time.monotonic() > deadline: msg = f"timed out waiting to carry {short!r} (carrying {me and me['h']!r})" raise AssertionError(msg) page.wait_for_timeout(50) def _events(page: Page) -> list[dict]: """The wire's events as the client saw them (the `snap` frames' `e` arrays — the encoder's short form: `k`, `i` = the player id, `r` = the denied reason).""" return page.evaluate( """() => { const out = []; for (const frame of window.__undercooked.messageLog) { const e = frame.raw !== undefined ? frame.raw.e : frame.events; if (Array.isArray(e)) for (const ev of e) out.push(ev); } return out; }""" ) def _item_count(sim_state: Any) -> int: """The server's item census (a station's contents + a hand).""" stations = sim_state.stations players = sim_state.players total = sum(len(station.contents) for station in stations) for player in players: if getattr(player, "carrying", None) is not None: total += 1 return total @pytest.fixture(scope="module") def us04_server(e2e_server): """The module's one test-mode server (the fixed-seed seam).""" live = e2e_server(US04_LEVELS_DIR, test_mode=True) yield live live.close() def test_moves_and_faces(browser: Browser, us04_server) -> None: """Key presses produce the expected position deltas; the facing follows the movement and is sticky after a stop; a diagonal is normalised on the client (the census reads the exact 1/√2).""" url = us04_server.url # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(_seeded_url(url), wait_until="networkidle") _start_solo(page) p0 = page.evaluate(ME_PROBE) assert p0 is not None and abs(p0["x"] - 2.5) < 0.1 and abs(p0["y"] - 7.5) < 0.1, ( f"the spawn is not at (2.5, 7.5): {p0}" ) # Hold D (right) — mid-hold: the facing is right (1), x is # moving, y is the row (the 2-decimal quantization's band). page.keyboard.down("d") page.wait_for_timeout(500) mid = page.evaluate(ME_PROBE) page.keyboard.up("d") assert mid is not None assert mid["f"] == 1, f"the facing is not right while moving right: {mid}" assert mid["x"] > p0["x"] + 0.2, f"no x delta while moving right: {p0} -> {mid}" assert abs(mid["y"] - 7.5) < 0.1, f"y drifted off the row: {mid}" # The stop: the deceleration tail, then equal reads (the # facing is sticky — still right). page.wait_for_timeout(400) p1 = page.evaluate(ME_PROBE) page.wait_for_timeout(250) p2 = page.evaluate(ME_PROBE) page.wait_for_timeout(250) p3 = page.evaluate(ME_PROBE) assert p1 is not None and p2 is not None and p3 is not None assert abs(p2["x"] - p1["x"]) < 0.05 and abs(p3["x"] - p2["x"]) < 0.05, ( f"the stop jitters: {p1} -> {p2} -> {p3}" ) assert p3["f"] == 1, f"the facing was not sticky after the stop: {p3}" # A short A (left) hold: the facing turns left (3) and x drops. before = page.evaluate(ME_PROBE) page.keyboard.down("a") page.wait_for_timeout(350) page.keyboard.up("a") page.wait_for_timeout(400) after = page.evaluate(ME_PROBE) assert before is not None and after is not None assert after["f"] == 3, f"the facing is not left after the A hold: {after}" assert 0.05 < (before["x"] - after["x"]) < 3.0, ( f"the A hold did not move the cook left: {before} -> {after}" ) # A diagonal (D+W): the client normalises — the census's last # frame is the exact 1/√2 on both axes (the server's clamp must # never see a diagonal; the magnitude stays ≤ 1). page.keyboard.down("d") page.keyboard.down("w") page.wait_for_timeout(250) stats = page.evaluate(INPUT_PROBE) page.keyboard.up("d") page.keyboard.up("w") last = stats["last"] assert abs(last["x"] - math.sqrt(2) / 2) < 1e-9, ( f"the diagonal's x is not normalised: {last}" ) assert abs(last["y"] + math.sqrt(2) / 2) < 1e-9, ( f"the diagonal's y is not normalised: {last}" ) finally: context.close() def test_stops_at_walls(browser: Browser, us04_server) -> None: """Walking into a wall does not pass through (the body's centre settles at `wall - 0.35`) and does not jitter (repeated reads are equal to the wire's 2-decimal quantization).""" url = us04_server.url # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(_seeded_url(url), wait_until="networkidle") _start_solo(page) # The left wall (from the spawn's x 2.5: a short walk). page.keyboard.down("a") page.wait_for_timeout(1500) page.keyboard.up("a") page.wait_for_timeout(500) me = page.evaluate(ME_PROBE) assert me is not None assert WALL_X_MIN - WALL_BAND <= me["x"] <= WALL_X_MIN + WALL_BAND, ( f"the cook is not stopped at the left wall: x={me['x']} (expect {WALL_X_MIN})" ) # The right wall (the long walk: x 1.35 -> 10.65). page.keyboard.down("d") page.wait_for_timeout(5000) page.keyboard.up("d") page.wait_for_timeout(500) me = page.evaluate(ME_PROBE) assert me is not None assert WALL_X_MAX - WALL_BAND <= me["x"] <= WALL_X_MAX + WALL_BAND, ( f"the cook is not stopped at the right wall: x={me['x']} (expect {WALL_X_MAX})" ) # The bottom wall (y 7.5 -> 7.65: a short walk). page.keyboard.down("s") page.wait_for_timeout(1000) page.keyboard.up("s") page.wait_for_timeout(500) me = page.evaluate(ME_PROBE) assert me is not None assert WALL_Y_MAX - WALL_BAND <= me["y"] <= WALL_Y_MAX + WALL_BAND, ( f"the cook is not stopped at the bottom wall: y={me['y']} (expect {WALL_Y_MAX})" ) # No jitter: three reads, 300 ms apart, are equal (the wire's # quantization is 2 decimals — a stopped body is a constant). reads = [page.evaluate(ME_PROBE)] for _ in range(2): page.wait_for_timeout(300) reads.append(page.evaluate(ME_PROBE)) for first, second in pairwise(reads): assert first is not None and second is not None assert abs(first["x"] - second["x"]) <= 0.01 and abs( first["y"] - second["y"] ) <= 0.01, f"the stopped body jitters: {first} -> {second}" finally: context.close() def test_serves_one_soup(browser: Browser, us04_server) -> None: """The full chain, keyboard-driven: the tomato off the spawn shelf, chop, pot, cook, plate, serve — the HUD score bumps and the order ticket disappears (the `levels_us04` campaign: target 100, so the serve does not end the round).""" url = us04_server.url # type: ignore[attr-defined] app = us04_server.app # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(_seeded_url(url), wait_until="networkidle") code = _start_solo(page) room = app.state.supervisor.get(code) assert room is not None, "the room is not in the supervisor" page.set_default_timeout(60_000) # 1. The tomato, off the spawn shelf (the approach from the # east: facing left at the shelf; the nudge re-aims a # walk-band stop and the press retries a missed one). _walk(page, [(3.5, 7.5), (3.5, 1.5), SHELF_STAGE]) _sim_settled(room, 0, SHELF_STAGE) _pick(page, room, 0, "a", "tomato_raw") # 2. Chop it (the hold: the module's resend keeps it # continuous; the press after the release is the pick). _walk(page, [BOARD_STAGE]) _sim_settled(room, 0, BOARD_STAGE) _aim(page, "d") _hold_until(page, _station_ready_js("chop_board")) _take(page, room, 0, "tomato_chopped") # 3. Pot + cook it (6 s of work — the longest hold). _walk(page, [(3.5, 2.5), (5.5, 2.5), (5.5, 1.5), STOVE_STAGE]) _sim_settled(room, 0, STOVE_STAGE) _aim(page, "d") _hold_until(page, _station_ready_js("stove")) _take(page, room, 0, "tomato_cooked") # 4. Plate it. _walk(page, [(7.5, 3.5), PREP_STAGE]) _sim_settled(room, 0, PREP_STAGE) _aim(page, "a") _hold_until(page, _station_ready_js("prep")) _take(page, room, 0, "tomato_plated") # 5. Serve it — but wait for the order first (the campaign's # 16-20 s spawn interval; the serve would be `no_orders` # without one). _walk(page, [SERVE_STAGE]) _sim_settled(room, 0, SERVE_STAGE) page.wait_for_function( "() => Object.values(window.__undercooked.view.orders).length >= 1", timeout=30_000, ) ticket = page.locator(".ticket").first order_id = ticket.get_attribute("data-order-id") assert order_id is not None and order_id != "" assert "0" in page.locator("#hud-score").inner_text() # The HUD's score bumps (the 44-66 band at a first streak; a # denied serve is retried — the dish stays in the hand). score = _serve(page) assert 20 <= score <= 100, f"the serve's score is outside the band: {score}" # The order ticket disappears (the `gone` tombstone's DOM half). page.locator(f'.ticket[data-order-id="{order_id}"]').wait_for( state="detached", timeout=15_000 ) # The FX's census: the scripted `served` event landed. fx = page.evaluate(FX_PROBE) assert fx is not None and fx["processed"].get("served", 0) >= 1, ( f"no `served` FX: {fx and fx['processed']}" ) # The hand is empty (the dish left it at the window) — the # sim is the ground truth (the view lags ~100 ms). assert room.sim_state.players[0].carrying is None, ( "the sim's hand still carries the dish" ) finally: context.close() def test_denied_out_of_reach(browser: Browser, us04_server) -> None: """An early press with no station in reach shows the deny feedback (the FX's `denied`) and the item stays (the shelf's stock, the hand, both untouched).""" url = us04_server.url # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(_seeded_url(url), wait_until="networkidle") _start_solo(page) # The spawn (2.5, 7.5) is 2.24 tiles from the nearest station # (the sink (3, 5)) — beyond REACH + FORGIVE (1.6): the press # has nothing to point at. _press(page) page.wait_for_timeout(700) # The deny feedback: the FX's census shows the `denied`. fx = page.evaluate(FX_PROBE) assert fx is not None and fx["processed"].get("denied", 0) >= 1, ( f"no `denied` FX: {fx and fx['processed']}" ) # The item stays: the hand is empty and the spawn shelf's # stock is untouched (the wire's `g` — the first content's # short form). me = page.evaluate(ME_PROBE) assert me is not None and me["h"] is None, f"the hand is not empty: {me}" shelf = page.evaluate( """() => Object.values(window.__undercooked.view.stations) .find((s) => s.k === 'spawn_shelf' && s.x === 1 && s.y === 1)""" ) assert shelf is not None and shelf["g"] == "tomato_raw", ( f"the shelf's stock was touched: {shelf}" ) finally: context.close() def _hand_empty_js() -> str: return ( "() => { const me = Object.values(window.__undercooked.view.players)" ".find((p) => p.pi === window.__undercooked.you.seat); " "return me !== undefined && me.h === null; }" ) def _contended_attempt( page_a: Page, page_b: Page, room: Any, attempt: int ) -> dict[str, object]: """One simultaneous-press attempt (the module's hold keeps both frames' presses continuous). Returns the census: the item count (conserved across the attempt), who — if anyone — carried the item, and the loser's `contested_lost` denial when one landed. The straddle case (a carried winner, no denial: the winner's frame landed one tick first, the loser's tick saw an empty fixture — a silent no-op) is re-set here: the winner puts the item back on the prep so the next pair can land in one tick.""" before = _item_count(room.sim_state) last_tick_before = room.event_buffer[-1][0] if room.event_buffer else 0 # The sim's ground truth before the press: a press sent while a # body is still in flight resolves off the candidate grid. _sim_settled(room, 0, PREP_STAGE) _sim_settled(room, 1, (3.5, 4.5)) # The facing check (both arrive from the east; the nudge happens # only when the wire says the facing is off — a blind nudge can # step a cook off the prep's candidate grid). _aim(page_a, "a") _aim(page_b, "a") page_a.keyboard.down("Space") page_b.keyboard.down("Space") page_a.wait_for_timeout(300) page_a.keyboard.up("Space") page_b.keyboard.up("Space") page_a.wait_for_timeout(800) after = _item_count(room.sim_state) assert before == after, ( f"the item count is not conserved (attempt {attempt}): {before} -> {after}" ) sim_a = room.sim_state.players[0] sim_b = room.sim_state.players[1] me_a = page_a.evaluate(ME_PROBE) me_b = page_b.evaluate(ME_PROBE) assert me_a is not None and me_b is not None # The sim is the ground truth for the winner (the view lags ~100 # ms and a stale `h` would flip the loser's side). winner: int | None = None if sim_a.carrying is not None: winner = me_a["pi"] elif sim_b.carrying is not None: winner = me_b["pi"] denied = None if winner is not None: loser_token = ( page_b if winner == me_a["pi"] else page_a ).evaluate( "() => Object.values(window.__undercooked.view.players)" ".find((p) => p.pi === window.__undercooked.you.seat).i" ) # The server-side buffer (the client queue TTL-expires ~1 s # after the tick the event left it): a `contested_lost` # denial attached to the loser, after this attempt's start. denied = next( ( ev for tick, ev in room.event_buffer if tick > last_tick_before and ev.kind.value == "denied" and ev.payload.get("reason") == "contested_lost" and ev.payload.get("player_id") == loser_token ), None, ) if denied is None: # the straddle case — the winner re-sets the down assert winner is not None winner_page = page_a if winner == me_a["pi"] else page_b winner_page.wait_for_timeout(300) _set_down(winner_page, room, winner) # back on the prep return {"winner": winner, "denied": denied} def test_two_players_same_target(browser: Browser, us04_server) -> None: """Two contexts press at the same marked set-down simultaneously: exactly one gets the item, the other sees the `denied` (a same-tick press is `contested_lost`), and the server's item count is conserved (the in-process `room.sim_state`). Up to six attempts: a straddle pair consumes the item without a `denied` — the winner re-sets the down (the marked object is back on the prep) and the next pair retries.""" url = us04_server.url # type: ignore[attr-defined] app = us04_server.app # type: ignore[attr-defined] context_a, page_a, context_b, page_b = open_two_pages(browser, url) try: page_a.goto(_seeded_url(url), wait_until="networkidle") page_b.goto(_seeded_url(url), wait_until="networkidle") code = _start_duo(page_a, page_b) page_a.set_default_timeout(90_000) page_b.set_default_timeout(90_000) # The in-process room (the item census + the seed's server half). room = app.state.supervisor.get(code) assert room is not None, "the room is not in the supervisor" # A (spawn (2.5, 7.5)): the tomato, then a marked set-down on # the prep (a raw set-down is a marked object, no work). _walk(page_a, [(3.5, 7.5), (3.5, 1.5), SHELF_STAGE], timeout_s=120) _sim_settled(room, 0, SHELF_STAGE) _pick(page_a, room, 0, "a", "tomato_raw") _walk( page_a, [(5.5, 2.5), (5.5, 3.5), PREP_STAGE], timeout_s=120, ) _sim_settled(room, 0, PREP_STAGE) _set_down(page_a, room, 0) # the set-down (the marked raw on the prep) # B (spawn (9.5, 7.5)): the staging diagonal (facing left at # (3.5, 4.5): the front-diagonal covers the prep (2, 3)). _walk(page_b, [(9.5, 4.5), (3.5, 4.5)], timeout_s=120) _sim_settled(room, 1, (3.5, 4.5)) # The simultaneous press (up to six attempts — the docstring). denied = None winner_seat: int | None = None for attempt in range(1, 7): outcome = _contended_attempt(page_a, page_b, room, attempt) winner_seat = cast(int | None, outcome["winner"]) denied = outcome["denied"] if winner_seat is None: page_a.wait_for_timeout(500) # neither carried — retry the pair elif denied is not None: break # a carried winner with a `contested_lost` denial assert winner_seat is not None, ( "neither cook ever carried the item " f"(A={page_a.evaluate(ME_PROBE)}, B={page_b.evaluate(ME_PROBE)}, " "station_items=" f"{[(s.kind.value, len(s.contents)) for s in room.sim_state.stations if s.contents]}, " "hands=" f"{[None if p.carrying is None else p.carrying.kind.value for p in room.sim_state.players]}, " f"marks={len(room.sim_state.item_marks)}, " f"events={_events(page_a)[-12:]})" ) assert denied is not None, ( "no `denied` (contested_lost) for the loser after three simultaneous presses" ) # Exactly one carries (the other's hand is empty) — the sim # is the ground truth (the view lags ~100 ms). a_holds = room.sim_state.players[0].carrying is not None b_holds = room.sim_state.players[1].carrying is not None assert a_holds != b_holds, ( f"exactly one should carry: A={a_holds!r}, B={b_holds!r}" ) finally: context_a.close() context_b.close() def test_alt_tab_parks_the_cook(browser: Browser, us04_server) -> None: """A hidden tab releases every held key and sends one zero-intent frame with the action released (the module's `stats()` census — the last frame is {0, 0, false}, nothing held), and the server-observed position freezes (a short coast, then equal reads) — no cook running away when you alt-tab.""" url = us04_server.url # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(_seeded_url(url), wait_until="networkidle") _start_solo(page) # Hold D (right) and let the cook walk. page.keyboard.down("d") page.wait_for_timeout(700) moving = page.evaluate(ME_PROBE) assert moving is not None and moving["x"] > 2.5, ( f"the cook was not walking before the tab hid: {moving}" ) # The alt-tab: `document.hidden` forced (the us04 # `HIDDEN_PAUSE_PROBE` trick) — the module's `visibilitychange` # branch clears the held keys and sends the forced zero. page.evaluate( """() => { Object.defineProperty(document, 'hidden', { value: true, configurable: true }); document.dispatchEvent(new Event('visibilitychange')); }""" ) page.wait_for_timeout(400) # The module's census: the zero-intent frame, nothing held. stats = page.evaluate(INPUT_PROBE) assert stats["last"] == {"x": 0, "y": 0, "act": False}, ( f"the hidden tab did not send the forced zero: {stats['last']}" ) assert not any(stats["held"].values()), f"keys are still held: {stats['held']}" assert stats["actionHeld"] is False # The server-observed freeze: a short coast (the deceleration), # then equal reads (the 2-decimal quantization's band). page.wait_for_timeout(500) p1 = page.evaluate(ME_PROBE) page.wait_for_timeout(350) p2 = page.evaluate(ME_PROBE) page.wait_for_timeout(350) p3 = page.evaluate(ME_PROBE) assert p1 is not None and p2 is not None and p3 is not None for first, second in ((p1, p2), (p2, p3)): assert abs(first["x"] - second["x"]) <= 0.01 and abs( first["y"] - second["y"] ) <= 0.01, f"the hidden cook is still moving: {first} -> {second}" assert p3["x"] >= moving["x"], ( "the cook moved backwards after the park " f"(a coast stop, not a reversal): {moving} -> {p3}" ) # The cleanup: the key up (the module's `keyup` clears — it # does), the tab back (a `visibilitychange` the module ignores # when visible). page.keyboard.up("d") page.evaluate( """() => { Object.defineProperty(document, 'hidden', { value: false, configurable: true }); document.dispatchEvent(new Event('visibilitychange')); }""" ) finally: context.close() def test_readme_controls_match_keymap(browser: Browser, us04_server) -> None: """The README's "Controls" block is exactly `getKeyMap()` (the task's completion criterion — the in-game help overlay renders the same rows, so the README, the help and the module are one truth).""" url = us04_server.url # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(url, wait_until="networkidle") keymap = page.evaluate( "() => window.__uc.input" ".getKeyMap().map((row) => ({ keys: row.keys.join(' / '), action: row.action }))" ) text = (REPO_ROOT / "README.md").read_text(encoding="utf-8") match = re.search(r"### Controls\n(.*?)\n###? ", text, re.DOTALL) assert match is not None, "the README's `### Controls` block is missing" rows = [ parsed.groups() for line in match.group(1).splitlines() if (parsed := re.match(r"^- \*\*(.+?)\*\* — (.+)$", line.strip())) ] assert len(rows) == len(keymap), ( f"the README has {len(rows)} control rows, the key map has {len(keymap)}" ) for (keys_md, action_md), row in zip(rows, keymap, strict=True): assert keys_md == row["keys"], ( f"the README's keys {keys_md!r} != the key map's {row['keys']!r}" ) assert action_md == row["action"], ( f"the README's action {action_md!r} != the key map's {row['action']!r}" ) # The in-game help renders the same rows (the overlay's # `getKeyMap()` — the module re-exports the help's map). help_rows = page.evaluate( "() => import('/static/js/ui/help.js').then((m) => " "m.getKeyMap().map((row) => ({ keys: row.keys.join(' / '), action: row.action })))" ) assert help_rows == keymap, "the help's key map diverged from the module's" finally: context.close() def test_seed_param_matches_the_test_mode_server(browser: Browser, us04_server) -> None: """The `?seed=` seam's whole point: the page's `?seed=20240701` is the seed the test-mode room actually draws (the in-process `room._seed_cursor` — the supervisor's `test_mode` branch), and the app echoes it (the `seed` field).""" url = us04_server.url # type: ignore[attr-defined] app = us04_server.app # type: ignore[attr-defined] context = browser.new_context() page = context.new_page() try: page.goto(_seeded_url(url), wait_until="networkidle") page.evaluate("() => window.__undercooked.createLobby({ name: 'Ada' })") page.locator("#room-code").wait_for(state="visible", timeout=20_000) code = page.locator("#room-code").inner_text() room = app.state.supervisor.get(code) assert room is not None assert room._seed_cursor == 20240701, ( f"the test-mode room's base seed is not the documented constant: {room._seed_cursor}" ) seed = page.evaluate(SEED_PROBE) assert seed == SEED, f"the app did not echo the ?seed= param: {seed!r}" finally: context.close()