"""One live replay per public IP on the CPU server. The AWS video workers enforce this in their Python signaling server. On the CPU server every seat is its own relay process and the page picks the seat, so the rule lives in nginx, which fronts every seat's /signal socket. These tests pin the contract between the nginx config, the deploy script and the live page without starting nginx. """ import pathlib import re import unittest ROOT = pathlib.Path(__file__).resolve().parent NGINX = (ROOT / "deploy" / "nginx-w3cs.conf").read_text(encoding="utf-8") DEPLOY = (ROOT / "deploy" / "deploy-cpu-server.sh").read_text(encoding="utf-8") PAGE = (ROOT / "poc" / "interactive-live.html").read_text(encoding="utf-8") def location_blocks(config): """Yield (matcher, body) for every location block, in file order.""" for match in re.finditer( r"location\s+(.+?)\s*\{(.*?)\n \}", config, re.S): yield match.group(1).strip(), match.group(2) class NginxViewerGateTests(unittest.TestCase): def test_zone_keys_on_client_address_with_trusted_exemptions(self): # geo values are literal text: "default $binary_remote_addr;" in a # geo block is one shared key for every client (a global gate, # seen live 2026-08-21). geo flags the trusted addresses; the map # builds the key, and map values do expand variables. geo = NGINX.split("geo $w3cs_trusted_viewer {", 1)[1].split("}", 1)[0] self.assertIn("default 0;", geo) self.assertIn("127.0.0.1 1;", geo) self.assertIn("::1 1;", geo) self.assertIn("include /etc/nginx/w3cs-trusted-ips.conf;", geo) self.assertNotIn("$", geo) key = NGINX.split( "map $w3cs_trusted_viewer $w3cs_viewer_key {", 1)[1].split("}", 1)[0] self.assertIn('1 "";', key) self.assertIn("default $binary_remote_addr;", key) self.assertNotIn("geo $w3cs_viewer_key", NGINX) example = (ROOT / "deploy" / "w3cs-trusted-ips.conf.example" ).read_text(encoding="utf-8") self.assertIn("# 203.0.113.7 1;", example) self.assertNotIn('"";', example) self.assertIn( "limit_conn_zone $w3cs_viewer_key zone=w3cs_viewer:1m;", NGINX) self.assertIn("limit_conn_status 429;", NGINX) def test_only_seat_holding_sockets_and_the_probe_are_limited(self): limited = {matcher for matcher, body in location_blocks(NGINX) if "limit_conn w3cs_viewer 1;" in body} self.assertEqual({ "~ ^/seat(?[1-5])/signal$", "= /signal", "= /viewer-gate", }, limited) # Data planes, status probes and bundles of the session that holds # the seat must keep flowing while its socket is counted. for matcher, body in location_blocks(NGINX): if matcher not in limited: self.assertNotIn("limit_conn", body, matcher) def test_signal_route_is_matched_before_the_generic_seat_route(self): matchers = [matcher for matcher, _ in location_blocks(NGINX)] signal = matchers.index("~ ^/seat(?[1-5])/signal$") generic = matchers.index( "~ ^/seat(?[1-5])(?/.*)$") self.assertLess(signal, generic) # The legacy origin planes no longer swallow /signal unlimited. self.assertIn("~ ^/(reliable|frame|control)(/|$)", matchers) self.assertNotIn("~ ^/(signal|reliable|frame|control)(/|$)", matchers) def test_signal_route_keeps_the_websocket_proxy_contract(self): body = dict(location_blocks(NGINX))[ "~ ^/seat(?[1-5])/signal$"] self.assertIn( "proxy_pass http://127.0.0.1:$seat_port/signal$is_args$args;", body) self.assertIn("proxy_set_header Upgrade $http_upgrade;", body) self.assertIn("proxy_read_timeout 3600s;", body) def test_probe_reaches_the_preaccess_limiter(self): body = dict(location_blocks(NGINX))["= /viewer-gate"] # `return` runs in the rewrite phase, before limit_conn (preaccess), # and would answer 204 beside a held socket. A content handler does # not. self.assertIn("empty_gif;", body) self.assertNotIn("return", body) self.assertIn("add_header Cache-Control no-store always;", body) class DeployTests(unittest.TestCase): def test_deploy_guarantees_the_allowlist_include_before_nginx_t(self): create = DEPLOY.index( "install -m 0644 /dev/null /etc/nginx/w3cs-trusted-ips.conf") self.assertIn("[ -f /etc/nginx/w3cs-trusted-ips.conf ] ||", DEPLOY) self.assertLess(create, DEPLOY.index("\nnginx -t\n")) self.assertTrue( (ROOT / "deploy" / "w3cs-trusted-ips.conf.example").exists()) class PageTests(unittest.TestCase): def test_page_asks_the_gate_before_and_after_a_failed_connect(self): self.assertIn('fetch("/viewer-gate",{cache:"no-store"})', PAGE) self.assertIn("return response.status===429;", PAGE) connect = PAGE[PAGE.index("const connectWithSeat=async()=>{"): PAGE.index("throw NO_FREE_SEATS;\n};")] self.assertIn("if(await viewerGateBusy())throw VIEWER_BUSY;", connect) # Pre-connect check comes before the seat search; the retry after a # refused socket comes after the 4001 seat-busy branch. self.assertLess(connect.index("if(await viewerGateBusy()){"), connect.index("resolveSeat()")) self.assertLess(connect.index("lastSignalingClose?.code===4001"), connect.rindex("throw VIEWER_BUSY")) def test_page_never_gates_lab_http_origins(self): self.assertIn( 'if(location.protocol!=="https:"||explicitBridgePort)return false;', PAGE) def test_busy_card_waits_and_restarts_without_a_seat_search(self): self.assertIn("}else if(error===VIEWER_BUSY){", PAGE) self.assertIn( 'textContent="Please watch one replay at a time";', PAGE) busy = PAGE[PAGE.index("}else if(error===VIEWER_BUSY){"):] busy = busy[:busy.index("}else{")] self.assertIn("if(!await viewerGateBusy()){", busy) self.assertIn("location.reload();", busy) self.assertNotIn("probeSeats", busy) class LivePageAudioDefaultTests(unittest.TestCase): """Sound on at 50% for a first visit, same as the AWS streaming page.""" def test_volume_slider_starts_at_half(self): self.assertIn( '=0&&parsed<=1)" "initialVolume=parsed;", PAGE) self.assertNotIn( 'Number(localStorage.getItem("w3replays-volume"))', PAGE) self.assertIn("let lastAudibleVolume=initialVolume>0?initialVolume:.5;", PAGE) class LivePageWheelDirectionTests(unittest.TestCase): """Wheel up zooms in on every platform: Apple natural scrolling flips the browser's sign, so Apple platforms invert by default; ?wheel overrides.""" def test_zoom_sign_follows_the_physical_wheel(self): self.assertIn("let wheelInverted=(()=>{", PAGE) # One click flips a wrong platform guess (Windows touchpads invert # too) and persists it. self.assertIn('id="wheelbtn"', PAGE) self.assertIn("wheelInverted=!wheelInverted;", PAGE) self.assertIn('localStorage.setItem("w3wheel",wheelInverted?"natural":"classic");', PAGE) self.assertIn('localStorage.setItem("w3wheel",wheelParam);', PAGE) # No platform guess: the game's own default (wheel up zooms in) for # everyone; the button/param invert for an OS that flips the wheel. self.assertNotIn('navigator.platform);', PAGE.split('const zoomDelta', 1)[0][-600:]) self.assertIn('title="mouse wheel zoom direction"', PAGE) self.assertNotIn('click if wheel up zooms out', PAGE) self.assertIn("const zoomDelta=deltaY=>wheelInverted?-deltaY:deltaY;", PAGE) self.assertIn('delta:zoomDelta(event.deltaY)', PAGE) self.assertNotIn('delta:event.deltaY,', PAGE) class LivePageParkingBlockOrderTests(unittest.TestCase): """The signaling open handler runs while the module is suspended in the top-level `await connectWithSeat()`; everything it calls must already be initialized or the page throws a ReferenceError on every load.""" def test_parking_block_precedes_the_top_level_connect_await(self): connect = PAGE.index("await connectWithSeat());") for name in ("let gameplayReached=false;", "const parkingEnabled=(()=>{", "const viewerAway=()=>", "const sendVisibility=hidden=>{"): self.assertLess(PAGE.index(name), connect, name) self.assertIn("sendVisibility(viewerAway());", PAGE) if __name__ == "__main__": unittest.main() class SpectatorLanesTests(unittest.TestCase): """The relay's bridge-only lanes are not reachable through the public origin, and spectators never touch the per-IP gate.""" CONF = (ROOT / "deploy" / "nginx-w3cs.conf").read_text(encoding="utf-8") def test_tap_and_input_are_refused_by_nginx(self): deny = self.CONF.index("location ~ ^/seat[1-5]/(tap|input)$ {") self.assertIn("return 404;", self.CONF[deny:deny + 120]) # listed before the generic seat route so it wins the regex order self.assertLess(deny, self.CONF.index("location ~ ^/seat(?[1-5])(?/.*)$ {")) def test_gate_zone_is_untouched_by_spectators(self): self.assertEqual(3, self.CONF.count("limit_conn w3cs_viewer 1;")) self.assertNotIn("spectate", self.CONF.lower().replace("spectator copy", ""))