""" Challenger M3 Empirical Stress Test: /api/map Endpoint and Biome Pipeline. Validates: 1. All 30 biome integer queries (biome=1..30) return HTTP 200 with byte[3] == biome_code. 2. All 30 biome style string queries (biome=STY_01..STY_30) return HTTP 200 with expected byte[3]. 3. Legacy biome name queries return HTTP 200. 4. Canonical zone resolution with and without biome override. 5. Boundary, invalid, malformed, and adversarial inputs (fuzzing). 6. Multi-threaded concurrent request performance and thread safety. """ from __future__ import annotations import concurrent.futures import json import socketserver import struct import threading import urllib.parse import urllib.request import urllib.error import pytest from typing import Generator, Dict, Any, List from server.world.map_style_catalog import list_styles, get_map_style from server.world.map_biome_catalog import MAP_BIOMES from tools.serve_webapp import FreeExileHTTPRequestHandler @pytest.fixture(scope="module") def api_server() -> Generator[str, None, None]: class ThreadingServer(socketserver.ThreadingMixIn, socketserver.TCPServer): daemon_threads = True allow_reuse_address = True server = ThreadingServer(("127.0.0.1", 0), FreeExileHTTPRequestHandler) port = server.server_address[1] t = threading.Thread(target=server.serve_forever, daemon=True) t.start() base_url = f"http://127.0.0.1:{port}" yield base_url server.shutdown() server.server_close() class TestApiMapEndpointEmpirical: """Empirical verification of /api/map across all 30 biomes and edge cases.""" def test_all_30_integer_biomes(self, api_server: str) -> None: """Every biome code from 1 to 30 must yield 200 and exact byte[3] match.""" for code in range(1, 31): url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed={code * 11}&biome={code}" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200, f"Expected 200 for biome={code}, got {resp.status}" assert resp.headers.get("Content-Type") == "application/octet-stream" body = resp.read() assert len(body) > 16, f"Body too small for biome={code}" assert body[:2] == b"FE", f"Magic header mismatch for biome={code}" assert body[2] == 1, f"Version mismatch for biome={code}" assert body[3] == code, f"Byte 3 mismatch: expected {code}, got {body[3]}" def test_all_30_style_id_strings(self, api_server: str) -> None: """Querying by STY_ style ID string must return 200 and correct byte[3].""" styles = list_styles() assert len(styles) == 30 for style in styles: url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed=999&biome={style.style_id}" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200, f"Failed for style {style.style_id}" body = resp.read() assert body[:2] == b"FE" assert body[3] == style.biome_code, ( f"Style {style.style_id} expected byte 3 to be {style.biome_code}, got {body[3]}" ) def test_legacy_biome_names(self, api_server: str) -> None: """Legacy biome names (codes 1-5) must return 200 and match codes 1-5.""" legacy = { "BLEACHED_BONE_CANYON": 1, "SAVAGE_MANGROVE_SWAMP": 2, "CRIMSON_BLOOD_FOREST": 3, "OUTCAST_MINE_SHAFTS": 4, "CORRUPTED_FIEND_RUINS": 5, } for b_name, b_code in legacy.items(): url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed=123&biome={b_name}" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200 body = resp.read() assert body[3] == b_code, f"Legacy biome {b_name} expected {b_code}, got {body[3]}" def test_all_canonical_zones_default_biomes(self, api_server: str) -> None: """Every canonical zone without ?biome= parameter returns valid binary map.""" canonical_zones = [ ("zone_tang_kiem_nhai", 1), ("zone_ancient_sword_barrow", 6), ("zone_boundless_sandstorm", 14), ("zone_blood_scale_ruins", 3), ("zone_five_elements_altar", 23), ("zone_abyssal_ice_pond", 11), ("zone_infinite_blood_rift", 27), ("zone_purgatory_lava_cavern", 20), ("zone_boundless_celestial_palace", 30), ] for zid, expected_code in canonical_zones: url = f"{api_server}/api/map?zone_id={zid}&seed=42" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200 body = resp.read() assert body[:2] == b"FE" assert body[3] == expected_code, f"Zone {zid} expected canonical biome {expected_code}, got {body[3]}" def test_out_of_range_integer_biomes_behavior(self, api_server: str) -> None: """Inspect server behavior on out-of-range integer biomes.""" # Code 0, 31, 999: WildernessMapGenerator falls back to BLEACHED_BONE_CANYON (code 1) for bad_code in [0, 31, 999, -5]: url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed=42&biome={bad_code}" req = urllib.request.Request(url) try: with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200 body = resp.read() assert body[:2] == b"FE" # When out of range, get_biome_by_code returns None -> falls back to code 1 assert body[3] == 1, f"Expected fallback to code 1 for bad code {bad_code}, got {body[3]}" except urllib.error.HTTPError as e: # Document if server returns error assert e.code in (400, 500) def test_unknown_string_biome_returns_500_error_json(self, api_server: str) -> None: """Unknown string biome produces 500 JSON response rather than crashing server.""" url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed=42&biome=INVALID_NONEXISTENT_BIOME" req = urllib.request.Request(url) with pytest.raises(urllib.error.HTTPError) as exc_info: urllib.request.urlopen(req, timeout=5) err = exc_info.value assert err.code == 500 data = json.loads(err.read().decode("utf-8")) assert "error" in data assert "Unknown biome_id" in data["error"] def test_malformed_seed_parameter(self, api_server: str) -> None: """Non-integer seed returns 500 JSON response with error explanation.""" url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed=not_a_number" req = urllib.request.Request(url) with pytest.raises(urllib.error.HTTPError) as exc_info: urllib.request.urlopen(req, timeout=5) err = exc_info.value assert err.code == 500 data = json.loads(err.read().decode("utf-8")) assert "error" in data def test_xss_and_path_traversal_payloads(self, api_server: str) -> None: """Verify adversarial payloads do not cause code injection or server crashes.""" adversarial_payloads = [ "", "../../../../etc/passwd", "SELECT * FROM map_styles", "%00%ff%fe", ] for payload in adversarial_payloads: encoded = urllib.parse.quote(payload) url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed=42&biome={encoded}" req = urllib.request.Request(url) try: with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200 except urllib.error.HTTPError as e: assert e.code in (400, 500) data = json.loads(e.read().decode("utf-8")) assert "error" in data def test_concurrent_requests_throughput(self, api_server: str) -> None: """50 concurrent requests across all 30 biomes execute with 100% success.""" def fetch_biome(code: int) -> int: url = f"{api_server}/api/map?zone_id=zone_tang_kiem_nhai&seed={code * 7}&biome={((code - 1) % 30) + 1}" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=10) as resp: assert resp.status == 200 b = resp.read() return b[3] with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: futures = [executor.submit(fetch_biome, i) for i in range(1, 51)] results = [f.result() for f in concurrent.futures.as_completed(futures)] assert len(results) == 50 assert all(1 <= r <= 30 for r in results)