194 lines
7.0 KiB
Python
194 lines
7.0 KiB
Python
import contextlib
|
|
import random
|
|
import threading
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
|
|
SECRET = "test-secret"
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
settings = Settings(ai_shared_secret=SECRET, ai_move_delay=0, trick_pause=0, ai_decision_timeout=2)
|
|
with TestClient(create_app(settings)) as test_client, contextlib.ExitStack() as sockets:
|
|
test_client.sockets = sockets # websocket sessions are closed before the app stops
|
|
yield test_client
|
|
|
|
|
|
def receive_until(ws, predicate, limit=5000):
|
|
for _ in range(limit):
|
|
message = ws.receive_json()
|
|
if predicate(message):
|
|
return message
|
|
raise AssertionError("expected message not received")
|
|
|
|
|
|
def is_state(message):
|
|
return message["type"] == "state"
|
|
|
|
|
|
def connect(client, token=None):
|
|
ws = client.sockets.enter_context(client.websocket_connect("/api/ws"))
|
|
ws.send_json({"type": "hello", "token": token})
|
|
welcome = receive_until(ws, lambda m: m["type"] == "welcome")
|
|
receive_until(ws, is_state)
|
|
return ws, welcome
|
|
|
|
|
|
def test_health(client):
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
|
|
def test_requires_hello(client):
|
|
with client.websocket_connect("/api/ws") as ws:
|
|
ws.send_json({"type": "create_lobby", "name": "Ann"})
|
|
assert ws.receive_json()["type"] == "error"
|
|
|
|
|
|
def test_invalid_message(client):
|
|
ws, _ = connect(client)
|
|
ws.send_json({"type": "join_lobby", "code": "ABCDEF", "name": ""})
|
|
assert ws.receive_json()["type"] == "error"
|
|
|
|
|
|
def test_lobby_flow_and_full_game(client):
|
|
host, host_welcome = connect(client)
|
|
host.send_json({"type": "create_lobby", "name": "Ann"})
|
|
state = receive_until(host, is_state)
|
|
lobby = state["lobby"]
|
|
code = lobby["code"]
|
|
assert lobby["host_id"] == host_welcome["player_id"]
|
|
assert [p["name"] for p in lobby["players"]] == ["Ann"]
|
|
|
|
guest, _ = connect(client)
|
|
guest.send_json({"type": "join_lobby", "code": code.lower(), "name": "Ann"})
|
|
state = receive_until(guest, lambda m: is_state(m) and m["lobby"])
|
|
assert [p["name"] for p in state["lobby"]["players"]] == ["Ann", "Ann (2)"]
|
|
|
|
# Only the host may start / add AI players.
|
|
guest.send_json({"type": "add_ai"})
|
|
assert receive_until(guest, lambda m: m["type"] == "error")
|
|
|
|
host.send_json({"type": "start_game"})
|
|
assert "needs" in receive_until(host, lambda m: m["type"] == "error")["message"]
|
|
|
|
host.send_json({"type": "add_ai"})
|
|
host.send_json({"type": "set_options", "not_equal": True, "hidden_bids": False})
|
|
host.send_json({"type": "start_game"})
|
|
|
|
rng = random.Random(0)
|
|
finished = {}
|
|
|
|
def play(ws, key):
|
|
while True:
|
|
message = ws.receive_json()
|
|
if not is_state(message) or not message["game"]:
|
|
continue
|
|
obs = message["game"]["observation"]
|
|
if obs["phase"] == "finished":
|
|
finished[key] = obs
|
|
return
|
|
if obs["current_player"] == message["game"]["you"] and obs["legal_actions"]:
|
|
ws.send_json({"type": "action", "action": rng.choice(obs["legal_actions"])})
|
|
|
|
threads = [threading.Thread(target=play, args=(ws, key)) for ws, key in ((host, "host"), (guest, "guest"))]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(timeout=60)
|
|
assert set(finished) == {"host", "guest"}
|
|
result = finished["host"]
|
|
assert len(result["round_history"]) == 20
|
|
assert result["winners"]
|
|
|
|
host.send_json({"type": "return_to_lobby"})
|
|
state = receive_until(host, lambda m: is_state(m) and m["lobby"]["status"] == "waiting")
|
|
assert state["game"] is None
|
|
|
|
|
|
def test_reconnect_with_token(client):
|
|
ws, welcome = connect(client)
|
|
ws.send_json({"type": "create_lobby", "name": "Bob"})
|
|
code = receive_until(ws, is_state)["lobby"]["code"]
|
|
ws.close()
|
|
|
|
ws2 = client.sockets.enter_context(client.websocket_connect("/api/ws"))
|
|
ws2.send_json({"type": "hello", "token": welcome["token"]})
|
|
again = receive_until(ws2, lambda m: m["type"] == "welcome")
|
|
assert again["player_id"] == welcome["player_id"]
|
|
state = receive_until(ws2, is_state)
|
|
assert state["lobby"]["code"] == code
|
|
|
|
|
|
def test_leaving_mid_game_hands_seat_to_ai(client):
|
|
host, _ = connect(client)
|
|
host.send_json({"type": "create_lobby", "name": "Cy"})
|
|
code = receive_until(host, is_state)["lobby"]["code"]
|
|
guest, _ = connect(client)
|
|
guest.send_json({"type": "join_lobby", "code": code, "name": "Di"})
|
|
receive_until(guest, lambda m: is_state(m) and m["lobby"])
|
|
host.send_json({"type": "add_ai"})
|
|
host.send_json({"type": "start_game"})
|
|
receive_until(host, lambda m: is_state(m) and m["game"])
|
|
|
|
host.send_json({"type": "leave_lobby"})
|
|
receive_until(host, lambda m: is_state(m) and m["lobby"] is None)
|
|
state = receive_until(guest, lambda m: is_state(m) and m["lobby"]["host_id"] != m["lobby"]["players"][0]["id"])
|
|
players = state["lobby"]["players"]
|
|
assert players[0]["is_ai"] and players[0]["name"] == "Cy"
|
|
assert state["lobby"]["host_id"] == players[1]["id"]
|
|
|
|
|
|
def test_ai_endpoint_requires_secret(client):
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
with pytest.raises(WebSocketDisconnect):
|
|
with client.websocket_connect("/internal/ai") as ws:
|
|
ws.receive_json()
|
|
|
|
|
|
def test_ai_service_decides(client):
|
|
"""A fake AI service answers decide requests with the first legal action."""
|
|
requests = []
|
|
stop = threading.Event()
|
|
|
|
def fake_ai():
|
|
headers = {"Authorization": f"Bearer {SECRET}"}
|
|
with client.websocket_connect("/internal/ai", headers=headers) as ai:
|
|
ai.send_json({"type": "hello", "agent": "fake"})
|
|
while not stop.is_set():
|
|
message = ai.receive_json()
|
|
if message["type"] == "decide":
|
|
requests.append(message)
|
|
action = message["observation"]["legal_actions"][0]
|
|
ai.send_json({"type": "action", "request_id": message["request_id"], "action": action})
|
|
if len(requests) >= 3:
|
|
return
|
|
|
|
thread = threading.Thread(target=fake_ai, daemon=True)
|
|
thread.start()
|
|
|
|
host, _ = connect(client)
|
|
host.send_json({"type": "create_lobby", "name": "Eve"})
|
|
receive_until(host, lambda m: is_state(m) and m["lobby"] and m["lobby"]["ai_available"])
|
|
host.send_json({"type": "add_ai"})
|
|
host.send_json({"type": "add_ai"})
|
|
host.send_json({"type": "start_game"})
|
|
|
|
rng = random.Random(1)
|
|
while thread.is_alive():
|
|
message = host.receive_json()
|
|
if is_state(message) and message["game"]:
|
|
obs = message["game"]["observation"]
|
|
if obs["current_player"] == message["game"]["you"] and obs["legal_actions"]:
|
|
host.send_json({"type": "action", "action": rng.choice(obs["legal_actions"])})
|
|
stop.set()
|
|
assert len(requests) >= 3
|
|
assert all(r["observation"]["player"] in (1, 2) for r in requests)
|