"""Exercise the real AIServer HTTP boundary in an isolated database/process (no account logs).""" import concurrent.futures import http.client import json from pathlib import Path import socket import subprocess import tempfile import time import uuid def main(): root = Path(__file__).resolve().parents[1] exe = root / "AIServer/build-g1/Release/aiserver.exe" with tempfile.TemporaryDirectory(prefix="g1-http-", dir=root / "Tools/output") as folder: folder = Path(folder) with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] args = [str(exe), "--host", "127.0.0.1", "--port", str(port), "--db", str(folder / "test.db"), "--daemon"] proc = None def start(): nonlocal proc proc = subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) for _ in range(100): if proc.poll() is not None: raise RuntimeError("test server exited") try: if request("GET", "/health")[1].get("gameProtocol") == 1: return except OSError: pass time.sleep(.05) raise RuntimeError("test server did not start") def request(method, path, body=None, token=None): headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = "Bearer " + token conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) try: payload = json.dumps(body) if body is not None else None conn.request(method, path, payload, headers) response = conn.getresponse() return response.status, json.loads(response.read()) finally: conn.close() try: start() assert request("GET", "/game/state")[0] == 401 accounts = [] for name in ("alice", "bob"): status, account = request("POST", "/register", {"username": "g1_" + name, "password": "testing-password-123"}) assert status == 200 accounts.append(account) a, b = accounts status, first = request("GET", "/game/state", token=a["token"]) assert status == 200 and first["state"]["balls"] == 20 command = {"requestId": uuid.uuid4().hex, "action": "shot.begin", "aim": 5, "revision": first["state"]["revision"], "pid": b["pid"]} with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool: results = list(pool.map(lambda _: request("POST", "/game/command", command, a["token"]), range(24))) assert all(status == 200 and value["ok"] and value["state"]["balls"] == 19 for status, value in results) assert request("GET", "/game/state", token=b["token"])[1]["state"]["balls"] == 20 # Replay remains valid after process restart, without trusting a supplied pid. proc.terminate(); proc.wait(timeout=5); start() status, replay = request("POST", "/game/command", command, a["token"]) assert status == 200 and replay["replayed"] and replay["state"]["balls"] == 19 assert replay["shot"]["path"] == results[0][1]["shot"]["path"] changed = dict(command, aim=10) assert request("POST", "/game/command", changed, a["token"])[0] == 409 time.sleep(max(0, replay["shot"]["readyAt"] - time.time()) + .1) finish = {"requestId": uuid.uuid4().hex, "action": "shot.finish", "activityId": command["requestId"], "revision": replay["state"]["revision"], "coins": 999999} status, settled = request("POST", "/game/command", finish, a["token"]) assert status == 200 and settled["ok"] assert settled["state"]["coins"] == replay["shot"]["coins"] assert request("POST", "/game/command", finish, a["token"])[1]["state"] == settled["state"] assert settled['state']['schema'] == 9 and len(settled['state']['mails']) == 2 and len(settled['state']['stages']) == 10 and len(settled['state']['clearedStages']) == 20 assert settled['state']['chests'] == [0, 0, 0] and settled['state']['chestDailyClaimed'] == 0 and len(settled['state']['shards']) == 10 mail_id = settled['state']['mails'][0]['mailId'] mail = {'requestId': uuid.uuid4().hex, 'action': 'mail.claim', 'mailId': mail_id, 'revision': settled['state']['revision'], 'reward': {'coins': 999999}} with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool: receipts = list(pool.map(lambda _: request('POST', '/game/command', mail, a['token']), range(24))) assert all(status == 200 and value['ok'] and value['result']['claimedCount'] == 1 for status, value in receipts) claimed = receipts[0][1] assert claimed['state']['balls'] == settled['state']['balls'] + 5 assert claimed['state']['candy'] == settled['state']['candy'] + 20 assert claimed['state']['coins'] == settled['state']['coins'] bob = request('GET', '/game/state', token=b['token'])[1] assert not bob['state']['mails'][0]['claimed'] and bob['state']['balls'] == 20 proc.terminate(); proc.wait(timeout=5); start() mail_replay = request('POST', '/game/command', mail, a['token'])[1] assert mail_replay['replayed'] and mail_replay['result'] == claimed['result'] assert mail_replay['state']['balls'] == claimed['state']['balls'] mark = {'requestId': uuid.uuid4().hex, 'action': 'mail.read', 'mailId': mail_id, 'revision': bob['state']['revision']} marked = request('POST', '/game/command', mark, b['token'])[1] assert marked['ok'] and marked['state']['mails'][0]['read'] and marked['state']['balls'] == 20 # Daily chest adds a box without awarding contents; opening it awards once under retry. daily = {'requestId': uuid.uuid4().hex, 'action': 'chest.claimDaily', 'revision': claimed['state']['revision']} with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool: boxes = list(pool.map(lambda _: request('POST', '/game/command', daily, a['token']), range(24))) assert all(status == 200 and value['ok'] and value['result']['count'] == 1 for status, value in boxes) boxed = boxes[0][1] assert boxed['state']['chests'] == [1, 0, 0] and boxed['state']['chestDailyClaimed'] == 1 assert boxed['state']['coins'] == claimed['state']['coins'] and boxed['state']['candy'] == claimed['state']['candy'] opened = {'requestId': uuid.uuid4().hex, 'action': 'chest.open', 'chest': 0, 'revision': boxed['state']['revision']} with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool: receipts = list(pool.map(lambda _: request('POST', '/game/command', opened, a['token']), range(24))) assert all(status == 200 and value['ok'] and value['result']['remaining'] == 0 for status, value in receipts) rewarded = receipts[0][1] assert rewarded['state']['chests'] == [0, 0, 0] assert rewarded['state']['coins'] == boxed['state']['coins'] + 60 assert rewarded['state']['candy'] == boxed['state']['candy'] + 30 assert rewarded['state']['balls'] == boxed['state']['balls'] + 2 assert bob['state']['chests'] == [0, 0, 0] and bob['state']['chestDailyClaimed'] == 0 (folder / "maintenance.json").write_text('{"enabled":true,"message":"testing"}', encoding="utf-8") assert request("GET", "/game/state", token=a["token"])[0] == 503 assert request("POST", "/game/command", finish, a["token"])[0] == 503 print("PASS HTTP: bearer auth, account isolation, 24 concurrent shot/mail/chest retries, restart receipts, mail read/claim, trajectory, once-only settlement, maintenance") finally: if proc and proc.poll() is None: proc.terminate(); proc.wait(timeout=5) if __name__ == "__main__": main()