112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
import unittest
|
|
|
|
from pathlib import Path
|
|
import tempfile, sqlite3
|
|
|
|
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,
|
|
"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 test_records_gameplay_with_wal_and_inferred_moves(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 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",
|
|
"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}},
|
|
)
|
|
await database.record_game_end(self._build_state(turn=2, me_head=(2, 2), enemy_head=(4, 4), include_enemy=False))
|
|
|
|
connection = sqlite3.connect(str(db_path))
|
|
journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0]
|
|
self.assertEqual(str(journal_mode).lower(), "wal")
|
|
|
|
games = connection.execute("SELECT status, winner_you, final_turn FROM games WHERE game_id = ?", ("game-abc",)).fetchone()
|
|
self.assertEqual(games[0], "finished")
|
|
self.assertEqual(games[1], 1)
|
|
self.assertEqual(games[2], 2)
|
|
|
|
turns_count = connection.execute("SELECT COUNT(*) FROM turns WHERE game_id = ?", ("game-abc",)).fetchone()[0]
|
|
self.assertEqual(turns_count, 2)
|
|
|
|
me_inferred = connection.execute("SELECT inferred_move FROM snake_turns WHERE game_id = ? AND turn = ? AND snake_id = ?", ("game-abc", 2, "me")).fetchone()[0]
|
|
enemy_inferred = connection.execute("SELECT inferred_move FROM snake_turns WHERE game_id = ? AND turn = ? AND snake_id = ?", ("game-abc", 2, "enemy")).fetchone()[0]
|
|
self.assertEqual(me_inferred, "up")
|
|
self.assertEqual(enemy_inferred, "left")
|
|
|
|
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.assertIsNotNone(replay)
|
|
replay = replay or {}
|
|
self.assertEqual(replay["game"]["final_turn"], 2)
|
|
self.assertEqual(len(replay["turns"]), 2)
|
|
self.assertEqual(replay["turns"][1]["my_move"], "up")
|
|
self.assertEqual(replay["turns"][1]["my_thinking"]["reason"], "food")
|
|
|
|
connection.close()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|