add GameplayDatabase database with dashboard

tests/test_GameplayDatabase.py

@@ -0,0 +1,111 @@
+import unittest
+
+from pathlib import Path
+import tempfile, sqlite3
+
+from server.database import GameplayDatabase
+
+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(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()