"""Locust load test for the place game. Simulates real players: log in, fetch the canvas, open the same websocket the frontend uses, and place a pixel as soon as the per-user cooldown allows - while a background greenlet keeps draining the broadcast stream so each idle connection costs what a real one costs. A placement that comes back rejected or errors out (timeout, dropped connection, ...) has a configurable chance of triggering a "reload" - tearing the socket down and redoing the page load (HTML + canvas fetch + fresh websocket), the way a real, mildly frustrated player mashes refresh instead of quietly waiting. That reconnect-storm behavior - many clients reloading around the same time - is itself part of what's worth measuring under load, not just steady-state placement throughput. Authentication uses local Django accounts (ModelBackend), never the real CAS server at polylan.ch - see testing_auth_views.py. This ONLY works against a target where LOAD_TEST_LOGIN_ENABLED=true; never point it at production. Setup, on the target server: export LOAD_TEST_LOGIN_ENABLED=true Then, in the Django admin, open Accounts > Users and click "Create load test users" (see accounts/admin.py) - it's only visible with the env var above set. Creates/resets loadtest0..loadtest499. Run: uv run locust -f loadtest/locustfile.py --host http://localhost:7777 Env vars: PLACE_EVENT_PATH event slug under test (default: "loadtest") LOAD_TEST_USER_COUNT must match accounts/admin.py's LOAD_TEST_USER_COUNT (default: 500) LOAD_TEST_USER_PREFIX must match accounts/admin.py's LOAD_TEST_USER_PREFIX (default: "loadtest") LOAD_TEST_PASSWORD must match accounts/admin.py's LOAD_TEST_PASSWORD (default: "loadtest-only-password") RELOAD_ON_FAILURE_PROBABILITY chance a failed placement reloads instead of just waiting for the next cooldown (default: 0.7) """ import os import random import struct import time import gevent import websocket from gevent.queue import Empty, Queue from locust import HttpUser, between, events, task EVENT_PATH = os.environ.get("PLACE_EVENT_PATH", "loadtest") USER_COUNT = int(os.environ.get("LOAD_TEST_USER_COUNT", "500")) USER_PREFIX = os.environ.get("LOAD_TEST_USER_PREFIX", "loadtest") PASSWORD = os.environ.get("LOAD_TEST_PASSWORD", "loadtest-only-password") RELOAD_ON_FAILURE_PROBABILITY = float(os.environ.get("RELOAD_ON_FAILURE_PROBABILITY", "0.7")) # Mirrors place/models.py::PALETTE - kept as a plain constant here since # loadtest runs standalone, without Django settings loaded. PALETTE_SIZE = 16 CANVAS_HEADER = struct.Struct(" str: """Round-robins through the pool of pre-created accounts so N Locust users spread evenly across USER_COUNT real accounts even when N differs from USER_COUNT.""" global _next_account_index username = f"{USER_PREFIX}{_next_account_index % USER_COUNT}" _next_account_index += 1 return username class PlaceUser(HttpUser): wait_time = between(1, 3) def on_start(self): self.username = _next_username() self._login() self.width, self.height = self._fetch_canvas_size() self._connect() def on_stop(self): self._teardown_ws() def _login(self): response = self.client.post( "/testing/login/", data={"username": self.username, "password": PASSWORD}, name="/testing/login/", ) if response.status_code != 200: raise RuntimeError( f"login failed for {self.username}: {response.status_code} " f"{response.text}. Is LOAD_TEST_LOGIN_ENABLED set on the " "target, and did you run the 'Create load test users' admin " "action (accounts/admin.py)?" ) def _fetch_canvas_size(self) -> tuple[int, int]: response = self.client.get( f"/api/events/{EVENT_PATH}/place/canvas", name="/api/events/[event]/place/canvas", ) width, height = CANVAS_HEADER.unpack_from(response.content, 0) return width, height def _connect(self): """(Re)opens the websocket and a fresh reader greenlet, then syncs next_placement_at to the server's own cooldown clock - every connection starts with an 's||...' message (see place/consumers.py::_initial_state) - rather than trusting local bookkeeping. That sync is exactly what makes a page reload meaningful to simulate: it re-anchors client state to the server's truth instead of just reopening a blank socket.""" self.ws = self._open_ws() self.next_placement_at = self._read_initial_next_placement_at() self.reply_queue = Queue() self._reader = gevent.spawn(self._reader_loop) def _teardown_ws(self): reader = getattr(self, "_reader", None) if reader is not None: reader.kill(block=False) ws = getattr(self, "ws", None) if ws is not None: try: ws.close() except Exception: pass def _reload(self): """What a real player does after a placement fails: hit refresh. Tears the socket down and redoes the page load over HTTP plus a brand-new websocket - the same reconnect churn a real outage or rejection wave would cause.""" self._teardown_ws() self.client.get(f"/{EVENT_PATH}/place", name="/[event]/place (reload)") self.width, self.height = self._fetch_canvas_size() self._connect() def _open_ws(self): scheme = "wss" if self.client.base_url.startswith("https") else "ws" host = self.client.base_url.split("://", 1)[1].rstrip("/") url = f"{scheme}://{host}/ws/place/{EVENT_PATH}/" cookie_header = "; ".join( f"{name}={value}" for name, value in self.client.cookies.get_dict().items() ) start = time.perf_counter() exc = None ws = None try: ws = websocket.create_connection( url, header=[f"Cookie: {cookie_header}"], timeout=10 ) except Exception as e: exc = e events.request.fire( request_type="WS", name="ws_connect", response_time=(time.perf_counter() - start) * 1000, response_length=0, exception=exc, ) if exc is not None: raise exc return ws def _read_initial_next_placement_at(self) -> float: message = self.ws.recv() next_ms = message.split("|")[1] return int(next_ms) / 1000 if next_ms else 0.0 def _reader_loop(self): """The only greenlet that ever calls ws.recv() - place_pixel below only ever writes, and reads back via reply_queue, so there's no race between two greenlets reading the same socket. Broadcast pixels from other players (and 'e|...' resize messages) are read and dropped here, same as a real idle viewer.""" while True: try: message = self.ws.recv() except Exception: return if not message: return if message[0] in ("k", "r"): self.reply_queue.put(message) @task def place_pixel(self): now = time.time() if now < self.next_placement_at: return # still on cooldown - a real client wouldn't send either x = random.randrange(self.width) y = random.randrange(self.height) color = random.randrange(PALETTE_SIZE) message = f"p|{x}|{y}|{color}" start = time.perf_counter() exc = None reply = None try: self.ws.send(message) reply = self.reply_queue.get(timeout=10) except (Empty, Exception) as e: exc = e events.request.fire( request_type="WS", name="place_pixel", response_time=(time.perf_counter() - start) * 1000, response_length=len(reply) if reply else 0, exception=exc, ) accepted = reply is not None and reply.startswith("k|") # Both an accepted placement (k|) and a cooldown rejection # (r|c|) carry the server's authoritative cooldown end - sync # to it either way, for the same reason _connect syncs off 's|...': # it keeps us honest against clock drift and against several # simulated users sharing one round-robined account, instead of # hammering the server with attempts it will just keep rejecting. if reply is not None and reply.startswith(("k|", "r|c|")): next_ms = reply.rsplit("|", 1)[1] self.next_placement_at = int(next_ms) / 1000 if not accepted and random.random() < RELOAD_ON_FAILURE_PROBABILITY: self._reload()