import unittest from pathlib import Path import tempfile, sqlite3 from contextlib import closing from server.database import GameplayDatabase, GameplayBackendBuilder class TestGameplayDatabase(unittest.IsolatedAsyncioTestCase): def _build_state(self, turn:int, me_head:tuple[int, int], enemy_head:tuple[int, int], include_enemy:bool=True) -> dict: snakes = [ { "id": "me", "name": "Me", "health": 90, "length": 3, "customizations": { "color": "#00ff00", "head": "ferret", "tail": "swirl", }, "head": {"x": me_head[0], "y": me_head[1]}, "body": [ {"x": me_head[0], "y": me_head[1]}, {"x": me_head[0] - 1, "y": me_head[1]}, {"x": me_head[0] - 2, "y": me_head[1]}, ], } ] if include_enemy: snakes.append({ "id": "enemy", "name": "Enemy", "health": 90, "length": 3, "head": {"x": enemy_head[0], "y": enemy_head[1]}, "body": [ {"x": enemy_head[0], "y": enemy_head[1]}, {"x": enemy_head[0], "y": enemy_head[1] + 1}, {"x": enemy_head[0], "y": enemy_head[1] + 2}, ], }) return { "turn": turn, "game": { "id": "game-abc", "source": "league", "map": "standard", "ruleset": {"name": "standard", "version": "v1.0.0"}, }, "board": { "width": 11, "height": 11, "food": [{"x": 2, "y": 2}], "hazards": [], "snakes": snakes, }, "you": snakes[0], } async def _play_game(self, database) -> None: await database.record_game_start(self._build_state(turn=0, me_head=(1, 1), enemy_head=(5, 5))) await database.record_turn( self._build_state(turn=1, me_head=(2, 1), enemy_head=(5, 4)), my_move="right", my_thinking={ "turn": 1, "reason": "safe_space", "health": 90, "length": 3, "scores": {"right": 1.0}, }, ) await database.record_turn( self._build_state(turn=2, me_head=(2, 2), enemy_head=(4, 4)), my_move="up", my_thinking={"turn": 2, "reason": "food", "scores": {"up": 1.4}}, ) async def test_running_game_uses_compact_rows(self): with tempfile.TemporaryDirectory() as temp_dir: db_path = Path(temp_dir) / "gameplay.sqlite3" database = GameplayDatabase(GameplayBackendBuilder.build(db_path=str(db_path), busy_timeout_ms=4000)) await self._play_game(database) with closing(sqlite3.connect(str(db_path))) as connection: self.assertEqual(str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower(), "wal") rows = connection.execute(""" SELECT t.turn, t.my_move, t.thinking_json, t.food FROM game_turns AS t JOIN games AS g ON g.gid = t.gid WHERE g.game_id = ? ORDER BY t.turn """, ("game-abc",)).fetchall() self.assertEqual([row[0] for row in rows], [1, 2]) self.assertEqual([row[1] for row in rows], [4, 1]) # right, up # Derivable thinking fields are stripped; unchanged food is NULL. self.assertNotIn('"turn"', rows[0][2]) self.assertNotIn('"health"', rows[0][2]) self.assertIsNotNone(rows[0][3]) self.assertIsNone(rows[1][3]) states = connection.execute("SELECT COUNT(*) FROM snake_states").fetchone()[0] self.assertEqual(states, 4) self.assertEqual(connection.execute("SELECT COUNT(*) FROM game_replays").fetchone()[0], 0) replay = await database.get_game_replay("game-abc") self.assertEqual(replay["game"]["status"], "running") self.assertEqual(replay["turns"][1]["food"], [{"x": 2, "y": 2}]) async def test_finished_game_is_packed_and_replays_identically(self): with tempfile.TemporaryDirectory() as temp_dir: db_path = Path(temp_dir) / "gameplay.sqlite3" database = GameplayDatabase(GameplayBackendBuilder.build(db_path=str(db_path), busy_timeout_ms=4000)) await self._play_game(database) running_replay = await database.get_game_replay("game-abc") await database.record_game_end(self._build_state(turn=2, me_head=(2, 2), enemy_head=(4, 4), include_enemy=False)) with closing(sqlite3.connect(str(db_path))) as connection: games = connection.execute("SELECT status, winner_you, final_turn FROM games WHERE game_id = ?", ("game-abc",)).fetchone() self.assertEqual(games, ("finished", 1, 2)) self.assertEqual(connection.execute("SELECT COUNT(*) FROM game_turns").fetchone()[0], 0) self.assertEqual(connection.execute("SELECT COUNT(*) FROM snake_states").fetchone()[0], 0) self.assertEqual(connection.execute("SELECT turn_count FROM game_replays").fetchone()[0], 2) identities = connection.execute(""" SELECT snake_id, snake_name, is_you, customizations_json FROM game_snakes WHERE game_id = ? ORDER BY snake_id """, ("game-abc",)).fetchall() self.assertEqual(identities, [ ("enemy", "Enemy", 0, "{}"), ("me", "Me", 1, '{"color":"#00ff00","head":"ferret","tail":"swirl"}'), ]) summary = await database.get_summary() self.assertEqual(summary["finished_games"], 1) self.assertEqual(summary["wins"], 1) replay = await database.get_game_replay("game-abc") self.assertEqual(replay["turns"], running_replay["turns"]) self.assertEqual(replay["game"]["final_turn"], 2) turn = replay["turns"][1] self.assertEqual(turn["my_move"], "up") self.assertEqual(turn["my_thinking"], {"turn": 2, "reason": "food", "scores": {"up": 1.4}}) self.assertEqual(replay["turns"][0]["my_thinking"]["health"], 90) self.assertEqual(turn["food"], [{"x": 2, "y": 2}]) self.assertEqual(turn["you"]["id"], "me") me = next(snake for snake in turn["snakes"] if snake["snake_id"] == "me") enemy = next(snake for snake in turn["snakes"] if snake["snake_id"] == "enemy") self.assertEqual(me["inferred_move"], "up") self.assertEqual(enemy["inferred_move"], "left") self.assertEqual(me["customizations"], {"color": "#00ff00", "head": "ferret", "tail": "swirl"}) self.assertEqual(len(me["body"]), 3) self.assertEqual(turn["snakes"][0]["snake_id"], "me") api_me = next(snake for snake in turn["board"]["snakes"] if snake["id"] == "me") self.assertEqual(api_me["customizations"], me["customizations"]) async def test_turn_after_restart_stores_full_food(self): with tempfile.TemporaryDirectory() as temp_dir: db_path = Path(temp_dir) / "gameplay.sqlite3" first = GameplayDatabase(GameplayBackendBuilder.build(db_path=str(db_path))) await first.record_game_start(self._build_state(turn=0, me_head=(1, 1), enemy_head=(5, 5))) await first.record_turn(self._build_state(turn=1, me_head=(2, 1), enemy_head=(5, 4)), my_move="right") second = GameplayDatabase(GameplayBackendBuilder.build(db_path=str(db_path))) await second.record_turn(self._build_state(turn=2, me_head=(2, 2), enemy_head=(4, 4)), my_move="up") replay = await second.get_game_replay("game-abc") self.assertEqual([turn["food"] for turn in replay["turns"]], [[{"x": 2, "y": 2}]] * 2) async def test_stale_running_game_is_finalized_and_packed(self): with tempfile.TemporaryDirectory() as temp_dir: db_path = Path(temp_dir) / "gameplay.sqlite3" database = GameplayDatabase(GameplayBackendBuilder.build(db_path=str(db_path))) await self._play_game(database) with closing(sqlite3.connect(str(db_path))) as connection, connection: connection.execute("UPDATE games SET started_at = '2020-01-01T00:00:00+00:00'") self.assertEqual(await database.finalize_stale_running_games(600), 1) replay = await database.get_game_replay("game-abc") self.assertEqual(replay["game"]["status"], "finished") self.assertFalse(replay["game"]["winner_you"]) self.assertEqual(len(replay["turns"]), 2) with closing(sqlite3.connect(str(db_path))) as connection: self.assertEqual(connection.execute("SELECT COUNT(*) FROM game_turns").fetchone()[0], 0) if __name__ == "__main__": unittest.main()