"""Locust load test for the market (betting) page. Simulates real bettors: log in, load the market page the way the frontend does (user info, markets, own bets, own event points, then one bet-history fetch per market since the charts are open by default), then keep placing or increasing bets, refreshing the page, and occasionally reloading it. Bets follow the same rules as the real UI: one option per market, the amount sent is the new total (so an increase re-sends the same option with a bigger amount), and a user never bets more than their event points. A 400 the server is expected to give under load ("Insufficient points", "Market is not open for betting", ...) counts as a success - it's the business rule doing its job, not the server failing. 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 In the Django admin: - Accounts > Users > "Create load test users" (see accounts/admin.py) creates/resets loadtest0..loadtest499. - Market > Markets: create a few markets with status "Open" (and a few options each) on the event under test. Without open markets the users only browse. Run: uv run locust -f loadtest/market_locustfile.py --host http://localhost:7777 Env vars: MARKET_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") MIN_BET_INCREMENT smallest amount added per bet (default: 10) MAX_BET_INCREMENT largest amount added per bet (default: 50) """ import os import random from locust import HttpUser, between, task EVENT_PATH = os.environ.get("MARKET_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") MIN_BET_INCREMENT = int(os.environ.get("MIN_BET_INCREMENT", "10")) MAX_BET_INCREMENT = int(os.environ.get("MAX_BET_INCREMENT", "50")) API = f"/api/events/{EVENT_PATH}/market" # 400s the market API returns on purpose - concurrent bettors racing for # the same points, or a market closed mid-test. Anything else is a failure. EXPECTED_BET_REJECTIONS = ( "Insufficient points", "Market is not open", "only bet on one option", "Cannot decrease bet", ) _next_account_index = 0 def _next_username() -> str: """Round-robins through the pool of pre-created accounts (same scheme as locustfile.py - kept separate so running this file doesn't also spawn PlaceUser).""" global _next_account_index username = f"{USER_PREFIX}{_next_account_index % USER_COUNT}" _next_account_index += 1 return username class MarketUser(HttpUser): wait_time = between(2, 6) def on_start(self): self.username = _next_username() self.markets: list[dict] = [] # market uuid -> {"option": option uuid, "amount": total} self.bets: dict[str, dict] = {} self.points = 0 self._login() self._load_page() 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 _load_page(self): """Full page load: HTML, then what useMarketStore.initializeMarketPage fetches, then one history fetch per card (charts are shown by default).""" self.client.get(f"/{EVENT_PATH}/market", name="/[event]/market") self.client.get("/api/user", name="/api/user") self._fetch_markets() self._fetch_bets() self._fetch_points() for market in self.markets: self._fetch_history(market) def _fetch_markets(self) -> dict[str, int]: """Returns each market's total pot, to tell which charts refetch.""" response = self.client.get(f"{API}/", name="/api/events/[event]/market/") if response.ok: self.markets = response.json() return { market["uuid"]: sum(opt["total_bets"] for opt in market["options"]) for market in self.markets } def _fetch_bets(self): response = self.client.get( f"{API}/user/bets", name="/api/events/[event]/market/user/bets" ) if response.ok: self.bets = { bet["market"]["uuid"]: { "option": bet["option"]["uuid"], "amount": bet["amount"], } for bet in response.json() if bet.get("market") } def _fetch_points(self): response = self.client.get( f"{API}/user/points", name="/api/events/[event]/market/user/points" ) if response.ok: self.points = response.json()["points"] def _fetch_history(self, market: dict): self.client.get( f"{API}/{market['uuid']}/history", name="/api/events/[event]/market/[market]/history", ) @task(5) def place_bet(self): open_markets = [m for m in self.markets if m["status"] == "open" and m["options"]] if not open_markets or self.points < MIN_BET_INCREMENT: return market = random.choice(open_markets) existing = self.bets.get(market["uuid"]) option = existing["option"] if existing else random.choice(market["options"])["uuid"] increment = random.randint(MIN_BET_INCREMENT, min(MAX_BET_INCREMENT, self.points)) amount = (existing["amount"] if existing else 0) + increment with self.client.post( f"{API}/{market['uuid']}/bets", json={"option_uuid": option, "amount": amount}, name="/api/events/[event]/market/[market]/bets", catch_response=True, ) as response: if response.status_code == 400 and any( reason in response.text for reason in EXPECTED_BET_REJECTIONS ): response.success() # MarketCard.placeBet reloads bets + points after every attempt self._fetch_bets() self._fetch_points() @task(3) def refresh(self): """useMarketStore.refreshPage, plus the history refetch the charts do whenever a market's pot changed.""" previous_pots = { market["uuid"]: sum(opt["total_bets"] for opt in market["options"]) for market in self.markets } self.client.get("/api/user", name="/api/user") pots = self._fetch_markets() self._fetch_bets() self._fetch_points() for market in self.markets: if pots.get(market["uuid"]) != previous_pots.get(market["uuid"]): self._fetch_history(market) @task(1) def reload(self): self._load_page()