daniel156161
c646392b84
- Preserve snake customizations across database migrations and merges. - Lazily load optional storage backends for SQLite maintenance scripts. - Match Apex territory and nearest-food tie-breaking semantics. - Resolve duel occupancy after simultaneous movement and food growth. - Recompute simulated head-to-head danger after body growth. - Add regression coverage and declare the aiofiles dependency.
105 lines
4.6 KiB
Python
105 lines
4.6 KiB
Python
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
from server.database.backend.SqliteGameplayBackend import SqliteGameplayBackend
|
|
|
|
class TestMergeGameplayDatabases(unittest.TestCase):
|
|
def _create_game(self, path:Path, game_id:str, winner_you:int, cleaned:bool=False) -> None:
|
|
SqliteGameplayBackend(str(path))
|
|
with sqlite3.connect(path) as connection:
|
|
connection.execute("""
|
|
INSERT INTO games (
|
|
game_id, started_at, ended_at, width, height, source, map_name,
|
|
ruleset_name, ruleset_version, your_snake_id, your_snake_name,
|
|
your_snake_type, your_snake_version, game_type, winner_name,
|
|
winner_you, final_turn, status, has_replay, quality_status,
|
|
quality_score, quality_tier, quality_reasons_json
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
""", (
|
|
game_id, "2026-01-01T00:00:00Z", "2026-01-01T00:01:00Z", 11, 11,
|
|
"league", "standard", "standard", "v1", "me", "PrismBattleSnake",
|
|
"PrismBattleSnake_GPT_5_6_Sol", "1.0.0", "duel", "PrismBattleSnake",
|
|
winner_you, 1, "finished", 1, "retained",
|
|
90 if cleaned else None, "high" if cleaned else None,
|
|
'["already_scored"]' if cleaned else None,
|
|
))
|
|
connection.execute("""
|
|
INSERT INTO game_snakes (
|
|
game_id,snake_id,snake_name,is_you,customizations_json
|
|
) VALUES (?,?,?,?,?)
|
|
""", (
|
|
game_id, "me", "PrismBattleSnake", 1,
|
|
'{"color":"#663399","head":"ferret","tail":"swirl"}',
|
|
))
|
|
connection.execute("""
|
|
INSERT INTO turns (
|
|
game_id,turn,observed_at,my_move,my_thinking_json,
|
|
board_state_json,snakes_json,you_json,food_json,hazards_json
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?)
|
|
""", (game_id, 1, "2026-01-01T00:00:01Z", "up", '{"score":1}', '{}', '[]', '{}', '[]', '[]'))
|
|
connection.execute("""
|
|
INSERT INTO snake_turns (
|
|
game_id,turn,snake_id,health,length,head_x,head_y,body_json,is_you,inferred_move
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?)
|
|
""", (game_id, 1, "me", 90, 3, 1, 1, '[{"x":1,"y":1}]', 0, "up"))
|
|
|
|
def test_merges_base_and_delta_and_regenerates_row_ids(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
base = root / "base.sqlite3"
|
|
delta = root / "delta.sqlite3"
|
|
merged = root / "merged.sqlite3"
|
|
self._create_game(base, "base-game", 1, cleaned=True)
|
|
self._create_game(delta, "delta-game", 0)
|
|
|
|
result = subprocess.run([
|
|
sys.executable, "scripts/merge_gameplay_databases.py",
|
|
"--base", str(base), "--delta", str(delta),
|
|
"--destination", str(merged), "--minimum-quality", "medium",
|
|
], cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
with sqlite3.connect(merged) as connection:
|
|
games = connection.execute(
|
|
"SELECT game_id, winner_you, has_replay, quality_tier FROM games ORDER BY game_id"
|
|
).fetchall()
|
|
self.assertEqual(games, [
|
|
("base-game", 1, 1, "high"),
|
|
("delta-game", 0, 1, "medium"),
|
|
])
|
|
self.assertEqual(connection.execute("SELECT COUNT(*) FROM turns").fetchone()[0], 2)
|
|
self.assertEqual(connection.execute("SELECT COUNT(*) FROM snake_turns").fetchone()[0], 2)
|
|
customizations = connection.execute("""
|
|
SELECT game_id, customizations_json FROM game_snakes ORDER BY game_id
|
|
""").fetchall()
|
|
self.assertEqual(customizations, [
|
|
("base-game", '{"color":"#663399","head":"ferret","tail":"swirl"}'),
|
|
("delta-game", '{"color":"#663399","head":"ferret","tail":"swirl"}'),
|
|
])
|
|
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
|
|
|
def test_conflicting_duplicate_aborts_without_destination(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
base = root / "base.sqlite3"
|
|
delta = root / "delta.sqlite3"
|
|
merged = root / "merged.sqlite3"
|
|
self._create_game(base, "same-game", 1, cleaned=True)
|
|
self._create_game(delta, "same-game", 0)
|
|
|
|
result = subprocess.run([
|
|
sys.executable, "scripts/merge_gameplay_databases.py",
|
|
"--base", str(base), "--delta", str(delta), "--destination", str(merged),
|
|
], cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("Conflicting duplicate game_id", result.stderr)
|
|
self.assertFalse(merged.exists())
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|