| from fastapi.testclient import TestClient | |
| from app import Card, app, rooms | |
| def setup_function(): | |
| rooms.clear() | |
| def create_room(client, name="Host", total=3): | |
| res = client.post( | |
| "/api/rooms", | |
| json={"name": name, "totalPlayers": total, "botDifficulty": "hard", "botJumpIn": True}, | |
| ) | |
| assert res.status_code == 200 | |
| return res.json() | |
| def join_room(client, code, name): | |
| res = client.post(f"/api/rooms/{code}/join", json={"name": name}) | |
| assert res.status_code == 200 | |
| return res.json() | |
| def ws_url(creds): | |
| return f"/ws/{creds['roomCode']}?playerId={creds['playerId']}&token={creds['token']}" | |
| def read_until_snapshot(ws): | |
| while True: | |
| msg = ws.receive_json() | |
| if msg["type"] == "snapshot": | |
| return msg["state"] | |
| def read_until_snapshot_matching(ws, predicate): | |
| while True: | |
| state = read_until_snapshot(ws) | |
| if predicate(state): | |
| return state | |
| def read_until_event(ws, event_type): | |
| while True: | |
| msg = ws.receive_json() | |
| if msg["type"] == "event" and msg["event"]["type"] == event_type: | |
| return msg["event"] | |
| def test_room_create_join_add_bot_and_privacy(): | |
| with TestClient(app) as client: | |
| host = create_room(client, total=4) | |
| p2 = join_room(client, host["roomCode"], "P2") | |
| with client.websocket_connect(ws_url(host)) as ws1, client.websocket_connect(ws_url(p2)) as ws2: | |
| s1 = read_until_snapshot(ws1) | |
| s2 = read_until_snapshot(ws2) | |
| assert s1["roomCode"] == host["roomCode"] | |
| assert s2["roomCode"] == host["roomCode"] | |
| ws1.send_json({"type": "add_bot", "payload": {}}) | |
| s1 = read_until_snapshot_matching(ws1, lambda state: any(p["isBot"] for p in state["players"])) | |
| assert any(p["isBot"] for p in s1["players"]) | |
| ws1.send_json({"type": "add_bot", "payload": {}}) | |
| s1 = read_until_snapshot_matching(ws1, lambda state: len(state["players"]) == 4) | |
| assert len(s1["players"]) == 4 | |
| ws1.send_json({"type": "start_game", "payload": {}}) | |
| s1 = read_until_snapshot_matching(ws1, lambda state: state["phase"] == "playing") | |
| s2 = read_until_snapshot_matching(ws2, lambda state: state["phase"] == "playing") | |
| assert s1["phase"] == "playing" | |
| my1 = next(p for p in s1["players"] if p["id"] == s1["myPlayerId"]) | |
| other1 = next(p for p in s1["players"] if p["id"] == s2["myPlayerId"]) | |
| assert "hand" in my1 | |
| assert "hand" not in other1 | |
| assert other1["handCount"] == 7 | |
| def test_non_host_cannot_start_or_change_options(): | |
| with TestClient(app) as client: | |
| host = create_room(client, total=3) | |
| p2 = join_room(client, host["roomCode"], "P2") | |
| join_room(client, host["roomCode"], "P3") | |
| with client.websocket_connect(ws_url(p2)) as ws: | |
| read_until_snapshot(ws) | |
| ws.send_json({"type": "start_game", "payload": {}}) | |
| msg = ws.receive_json() | |
| assert msg["type"] == "error" | |
| assert msg["code"] == "not_host" | |
| def test_room_full_and_started_errors(): | |
| with TestClient(app) as client: | |
| host = create_room(client, total=3) | |
| join_room(client, host["roomCode"], "P2") | |
| join_room(client, host["roomCode"], "P3") | |
| full = client.post(f"/api/rooms/{host['roomCode']}/join", json={"name": "P4"}) | |
| assert full.status_code == 409 | |
| p2 = { | |
| "roomCode": host["roomCode"], | |
| "playerId": rooms[host["roomCode"]].players[1].id, | |
| "token": rooms[host["roomCode"]].players[1].token, | |
| } | |
| p3 = { | |
| "roomCode": host["roomCode"], | |
| "playerId": rooms[host["roomCode"]].players[2].id, | |
| "token": rooms[host["roomCode"]].players[2].token, | |
| } | |
| with ( | |
| client.websocket_connect(ws_url(host)) as ws, | |
| client.websocket_connect(ws_url(p2)), | |
| client.websocket_connect(ws_url(p3)), | |
| ): | |
| read_until_snapshot(ws) | |
| ws.send_json({"type": "start_game", "payload": {}}) | |
| read_until_snapshot_matching(ws, lambda state: state["phase"] == "playing") | |
| late = client.post(f"/api/rooms/{host['roomCode']}/join", json={"name": "Late"}) | |
| assert late.status_code == 409 | |
| def test_rename_broadcasts_and_reconnect_restores_playing_seat(): | |
| with TestClient(app) as client: | |
| host = create_room(client, total=3) | |
| p2 = join_room(client, host["roomCode"], "P2") | |
| p3 = join_room(client, host["roomCode"], "P3") | |
| with ( | |
| client.websocket_connect(ws_url(host)) as ws_host, | |
| client.websocket_connect(ws_url(p2)) as ws_p2, | |
| client.websocket_connect(ws_url(p3)), | |
| ): | |
| read_until_snapshot(ws_host) | |
| read_until_snapshot(ws_p2) | |
| ws_host.send_json({"type": "rename", "payload": {"name": "Renamed Host"}}) | |
| host_view = read_until_snapshot_matching( | |
| ws_host, | |
| lambda state: next(p for p in state["players"] if p["id"] == host["playerId"])["name"] | |
| == "Renamed Host", | |
| ) | |
| p2_view = read_until_snapshot_matching( | |
| ws_p2, | |
| lambda state: next(p for p in state["players"] if p["id"] == host["playerId"])["name"] | |
| == "Renamed Host", | |
| ) | |
| assert host_view["players"][0]["name"] == "Renamed Host" | |
| assert any(p["name"] == "Renamed Host" for p in p2_view["players"]) | |
| ws_host.send_json({"type": "start_game", "payload": {}}) | |
| playing = read_until_snapshot_matching(ws_host, lambda state: state["phase"] == "playing") | |
| host_hand_before = [ | |
| card["id"] | |
| for card in next(p for p in playing["players"] if p["id"] == host["playerId"])["hand"] | |
| ] | |
| with client.websocket_connect(ws_url(host)) as ws_reconnected: | |
| restored = read_until_snapshot_matching( | |
| ws_reconnected, | |
| lambda state: state["phase"] == "playing" | |
| and next(p for p in state["players"] if p["id"] == host["playerId"])["name"] | |
| == "Renamed Host", | |
| ) | |
| host_after = next(p for p in restored["players"] if p["id"] == host["playerId"]) | |
| assert [card["id"] for card in host_after["hand"]] == host_hand_before | |
| def test_bad_room_and_bad_token_errors(): | |
| with TestClient(app) as client: | |
| missing = client.post("/api/rooms/ZZZZZ/join", json={"name": "Nope"}) | |
| assert missing.status_code == 404 | |
| host = create_room(client, total=3) | |
| bad = dict(host) | |
| bad["token"] = "wrong" | |
| with client.websocket_connect(ws_url(bad)) as ws: | |
| msg = ws.receive_json() | |
| assert msg["type"] == "error" | |
| assert msg["code"] == "bad_token" | |
| def test_room_code_deep_links_return_frontend(): | |
| with TestClient(app) as client: | |
| upper = client.get("/ABCDE") | |
| lower = client.get("/abcde") | |
| invalid = client.get("/TOOLONG") | |
| assert upper.status_code == 200 | |
| assert lower.status_code == 200 | |
| assert "UNO Multiplayer" in upper.text | |
| assert invalid.status_code == 404 | |
| created = client.post( | |
| "/api/rooms", | |
| json={"name": "Host", "totalPlayers": 3, "botDifficulty": "hard", "botJumpIn": True}, | |
| ) | |
| assert created.status_code == 200 | |
| def test_auto_draw_broadcasts_and_keeps_other_hands_private(): | |
| with TestClient(app) as client: | |
| host = create_room(client, total=3) | |
| p2 = join_room(client, host["roomCode"], "P2") | |
| p3 = join_room(client, host["roomCode"], "P3") | |
| with ( | |
| client.websocket_connect(ws_url(host)) as ws_host, | |
| client.websocket_connect(ws_url(p2)) as ws_p2, | |
| client.websocket_connect(ws_url(p3)), | |
| ): | |
| read_until_snapshot(ws_host) | |
| read_until_snapshot(ws_p2) | |
| ws_host.send_json({"type": "start_game", "payload": {}}) | |
| read_until_snapshot_matching(ws_host, lambda state: state["phase"] == "playing") | |
| room = rooms[host["roomCode"]] | |
| room.current_player_id = host["playerId"] | |
| room.current_color = "red" | |
| room.pending_draw = 0 | |
| room.discard_pile = [Card("auto_top", "red", "5")] | |
| host_player = next(player for player in room.players if player.id == host["playerId"]) | |
| host_player.hand = [Card("host_bad", "green", "8")] | |
| room.deck = [Card("auto_drawn", "blue", "9")] | |
| room.version += 1 | |
| ws_host.send_json({"type": "rename", "payload": {"name": "Auto Host"}}) | |
| host_event = read_until_event(ws_host, "cards_drawn") | |
| p2_event = read_until_event(ws_p2, "cards_drawn") | |
| assert host_event["auto"] is True | |
| assert host_event["reason"] == "no_playable" | |
| assert host_event["cards"] == [{"id": "auto_drawn", "color": "blue", "value": "9"}] | |
| assert "cards" not in p2_event | |
| assert p2_event["count"] == 1 | |
| host_state = read_until_snapshot_matching(ws_host, lambda state: state["currentPlayerId"] != host["playerId"]) | |
| host_view = next(player for player in host_state["players"] if player["id"] == host["playerId"]) | |
| p2_view_of_host = next(player for player in host_state["players"] if player["id"] == p2["playerId"]) | |
| assert len(host_view["hand"]) == 2 | |
| assert "hand" not in p2_view_of_host | |