From 1b4e7a5ee0cd807906f06eb44a27aa22c9980857 Mon Sep 17 00:00:00 2001 From: Vishesh 'ironeagle' Bangotra Date: Wed, 26 Nov 2025 14:51:10 +0530 Subject: [PATCH] feat(test): add comprehensive TicTacToe gameplay scenario flows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Introduce multiple async test paths for simulation-based validation: • happy_path (P1 top-row win) • p2_wins_diagonal • draw_game (full board, no winner) • illegal_occupied_cell • illegal_out_of_turn • illegal_out_of_bounds • midgame_disconnect • abandoned_lobby (no opponent joins) • spam_moves (anti-flood behavior) • random_game (stochastic stress playthrough) - Add TEST_SCENARIOS registry for automated execution - Improve coverage of server-side match logic, validation, and cleanup - Enables CI-driven load, rule enforcement, and termination testing --- game_flow.py | 157 +++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 147 insertions(+), 10 deletions(-) diff --git a/game_flow.py b/game_flow.py index 95c054d..06c754c 100644 --- a/game_flow.py +++ b/game_flow.py @@ -1,3 +1,4 @@ +import random import asyncio import base64 import json @@ -123,6 +124,149 @@ class PlayerWebSocketHandler(object): print(f"[{self.label}] Connection closed") +async def happy_path( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + # Play moves + await p1.send_move(match_id, 0, 0) + await asyncio.sleep(0.3) + await p2.send_move(match_id, 1, 1) + await asyncio.sleep(0.3) + await p1.send_move(match_id, 0, 1) + await asyncio.sleep(0.3) + await p2.send_move(match_id, 2, 2) + await asyncio.sleep(0.3) + await p1.send_move(match_id, 0, 2) + + +async def p2_wins_diagonal( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + await p1.send_move(match_id, 0, 0) + await asyncio.sleep(0.3) + await p2.send_move(match_id, 0, 2) + await asyncio.sleep(0.3) + await p1.send_move(match_id, 1, 0) + await asyncio.sleep(0.3) + await p2.send_move(match_id, 1, 1) + await asyncio.sleep(0.3) + await p1.send_move(match_id, 2, 1) + await asyncio.sleep(0.3) + await p2.send_move(match_id, 2, 0) # P2 wins + + +async def draw_game( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + moves = [ + (p1, 0, 0), (p2, 0, 1), + (p1, 0, 2), (p2, 1, 0), + (p1, 1, 2), (p2, 1, 1), + (p1, 2, 1), (p2, 2, 2), + (p1, 2, 0), + ] + for player, r, c in moves: + await player.send_move(match_id, r, c) + await asyncio.sleep(0.25) + + +async def illegal_occupied_cell( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + await p1.send_move(match_id, 0, 0) + await asyncio.sleep(0.3) + + # P2 tries same spot → should trigger server rejection + await p2.send_move(match_id, 0, 0) + + +async def illegal_out_of_turn( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + await p1.send_move(match_id, 2, 2) + await asyncio.sleep(0.3) + + # P1 tries again before P2 + await p1.send_move(match_id, 1, 1) + + +async def illegal_out_of_bounds( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + await p1.send_move(match_id, 3, 3) # Invalid indices + + +async def midgame_disconnect( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + await p1.send_move(match_id, 0, 0) + await asyncio.sleep(0.3) + await p2.close() # simulate rage quit + + +async def abandoned_lobby( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + # No p2.join_match + await asyncio.sleep(5) # test timeout, cleanup, match state + + +async def spam_moves( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + for _ in range(10): + await p1.send_move(match_id, 0, 0) + await asyncio.sleep(0.05) + + +async def random_game( + match_id: str, + p1: PlayerWebSocketHandler, + p2: PlayerWebSocketHandler +): + board = {(r, c) for r in range(3) for c in range(3)} + players = [p1, p2] + + for i in range(9): + player = players[i % 2] + print(f"[{player.label}] Playing move...") + r, c = random.choice(list(board)) + board.remove((r, c)) + await player.send_move(match_id, r, c) + await asyncio.sleep(0.25) + + +TEST_SCENARIOS = [ + happy_path, + p2_wins_diagonal, + draw_game, + illegal_occupied_cell, + illegal_out_of_turn, + illegal_out_of_bounds, + midgame_disconnect, + abandoned_lobby, + spam_moves, + random_game, +] + async def main(): # Create players p1 = PlayerWebSocketHandler("player_one_123456", "P1") @@ -142,16 +286,9 @@ async def main(): await asyncio.sleep(1) - # Play moves - await p1.send_move(match_id, 0, 0) - await asyncio.sleep(0.3) - await p2.send_move(match_id, 1, 1) - await asyncio.sleep(0.3) - await p1.send_move(match_id, 0, 1) - await asyncio.sleep(0.3) - await p2.send_move(match_id, 2, 2) - await asyncio.sleep(0.3) - await p1.send_move(match_id, 0, 2) + for test_scenario in TEST_SCENARIOS: + print(f'running {test_scenario.__name__!r}') + await test_scenario(match_id, p1, p2) await asyncio.sleep(2)