diff --git a/README.md b/README.md index 3b78f26..7a318be 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,14 @@ # Battlesnake Python Starter Project - An official Battlesnake template written in Python. Get started at [play.battlesnake.com](https://play.battlesnake.com). ![Battlesnake Logo](https://media.battlesnake.com/social/StarterSnakeGitHubRepos_Python.png) -This project is a great starting point for anyone wanting to program their first Battlesnake in Python. It can be run locally or easily deployed to a cloud provider of your choosing. See the [Battlesnake API Docs](https://docs.battlesnake.com/api) for more detail. +This project is a great starting point for anyone wanting to program their first Battlesnake in Python. It can be run locally or easily deployed to a cloud provider of your choosing. See the [Battlesnake API Docs](https://docs.battlesnake.com/api) for more detail. ## Technologies Used - This project uses [Python 3](https://www.python.org/) and [Flask](https://flask.palletsprojects.com/). It also comes with an optional [Dockerfile](https://docs.docker.com/engine/reference/builder/) to help with deployment. ## Run Your Battlesnake - Install dependencies using pip ```sh @@ -19,7 +16,6 @@ pip install -r requirements.txt ``` Start your Battlesnake - ```sh python main.py ``` @@ -39,19 +35,16 @@ Open [localhost:8000](http://localhost:8000) in your browser and you should see ``` ## Play a Game Locally - Install the [Battlesnake CLI](https://github.com/BattlesnakeOfficial/rules/tree/main/cli) * You can [download compiled binaries here](https://github.com/BattlesnakeOfficial/rules/releases) * or [install as a go package](https://github.com/BattlesnakeOfficial/rules/tree/main/cli#installation) (requires Go 1.18 or higher) Command to run a local game - ```sh battlesnake play -W 11 -H 11 --name 'Python Starter Project' --url http://localhost:8000 -g solo --browser ``` ## Next Steps - Continue with the [Battlesnake Quickstart Guide](https://docs.battlesnake.com/quickstart) to customize and improve your Battlesnake's behavior. ## Included Competitive Snake @@ -62,13 +55,11 @@ This repo now includes `snakes/BestBattleSnake.py`, a stronger default snake tha - tail access checks for better long-term survival Run it explicitly with: - ```sh SNAKE=BestBattleSnake python main.py ``` Optional duel tuning (when only 2 snakes are alive): - ```sh BATTLE_SNAKE_DUEL_STYLE=balanced python main.py ``` diff --git a/pyproject.toml b/pyproject.toml index b053fa6..312e25e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,7 @@ description = "Add your description here" readme = "README.md" requires-python = ">=3.13" dependencies = [ + "aiofiles>=25.1.0", "aiologger>=0.7.0", "dotenv>=0.9.9", "httpx>=0.28.0", diff --git a/scripts/merge_gameplay_databases.py b/scripts/merge_gameplay_databases.py index 61f2855..2f08c52 100755 --- a/scripts/merge_gameplay_databases.py +++ b/scripts/merge_gameplay_databases.py @@ -117,15 +117,26 @@ def copy_game_snakes(source:sqlite3.Connection, destination:sqlite3.Connection, "SELECT 1 FROM sqlite_master WHERE type='table' AND name='game_snakes'" ).fetchone() if has_table: - cursor = source.execute( - "SELECT game_id, snake_id, snake_name, is_you FROM game_snakes ORDER BY game_id, snake_id" + columns = object_columns(source, "game_snakes") + customizations = ( + "COALESCE(customizations_json, '{}') AS customizations_json" + if "customizations_json" in columns else "'{}' AS customizations_json" ) + cursor = source.execute(f""" + SELECT game_id, snake_id, snake_name, is_you, {customizations} + FROM game_snakes ORDER BY game_id, snake_id + """) else: cursor = source.execute(""" - SELECT game_id, snake_id, MAX(snake_name), MAX(is_you) + SELECT game_id, snake_id, MAX(snake_name), MAX(is_you), + '{}' AS customizations_json FROM snake_turns GROUP BY game_id, snake_id ORDER BY game_id, snake_id """) - sql = "INSERT INTO game_snakes (game_id,snake_id,snake_name,is_you) VALUES (?,?,?,?)" + sql = """ + INSERT INTO game_snakes ( + game_id,snake_id,snake_name,is_you,customizations_json + ) VALUES (?,?,?,?,?) + """ count = 0 while rows := cursor.fetchmany(batch_size): values = [tuple(row) for row in rows if row[0] in allowed] diff --git a/scripts/migrate_gameplay_database.py b/scripts/migrate_gameplay_database.py index 52fc1b9..0cdb553 100644 --- a/scripts/migrate_gameplay_database.py +++ b/scripts/migrate_gameplay_database.py @@ -144,20 +144,27 @@ def copy_game_snakes(source:sqlite3.Connection, destination:sqlite3.Connection, SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'game_snakes' """).fetchone() is not None if has_game_snakes: - cursor = source.execute(""" - SELECT game_id, snake_id, snake_name, is_you + columns = object_columns(source, "game_snakes") + customizations = ( + "COALESCE(customizations_json, '{}') AS customizations_json" + if "customizations_json" in columns else "'{}' AS customizations_json" + ) + cursor = source.execute(f""" + SELECT game_id, snake_id, snake_name, is_you, {customizations} FROM game_snakes ORDER BY game_id, snake_id """) else: cursor = source.execute(""" - SELECT game_id, snake_id, MAX(snake_name), MAX(is_you) + SELECT game_id, snake_id, MAX(snake_name), MAX(is_you), + '{}' AS customizations_json FROM snake_turns GROUP BY game_id, snake_id ORDER BY game_id, snake_id """) sql = """ - INSERT INTO game_snakes (game_id, snake_id, snake_name, is_you) - VALUES (?, ?, ?, ?) + INSERT INTO game_snakes ( + game_id, snake_id, snake_name, is_you, customizations_json + ) VALUES (?, ?, ?, ?, ?) """ count = 0 while rows := cursor.fetchmany(batch_size): diff --git a/server/database/__init__.py b/server/database/__init__.py index f495916..39d2354 100644 --- a/server/database/__init__.py +++ b/server/database/__init__.py @@ -1,12 +1,40 @@ +from typing import TYPE_CHECKING, Any + from .GameplayDatabase import GameplayDatabase from .backend import GameplayBackendBuilder -from .LocalStorage import LocalStorage -from .EdgeDB import EdgeDB +if TYPE_CHECKING: + from .EdgeDB import EdgeDB + from .LocalStorage import LocalStorage + +__all__ = ( + "EdgeDB", + "GameplayBackendBuilder", + "GameplayDatabase", + "LocalStorage", + "StorageLoader", +) + +def __getattr__(name:str): + """Load optional storage backends only when explicitly requested. + + Database maintenance scripts import SQLite backend modules through this + package. Eagerly importing LocalStorage used to make those scripts require + unrelated web-storage dependencies such as aiofiles. + """ + if name == "LocalStorage": + from .LocalStorage import LocalStorage + return LocalStorage + if name == "EdgeDB": + from .EdgeDB import EdgeDB + return EdgeDB + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") class StorageLoader: @classmethod - def build(self, selected_storage:str) -> LocalStorage|EdgeDB: - storage_module = __import__(f"server.database.{selected_storage}", fromlist=[selected_storage]) + def build(cls, selected_storage:str) -> Any: + storage_module = __import__( + f"server.database.{selected_storage}", fromlist=[selected_storage], + ) storage_class = getattr(storage_module, selected_storage) return storage_class diff --git a/snakes/PrismBattleSnake_GPT_5_6_Sol.py b/snakes/PrismBattleSnake_GPT_5_6_Sol.py index 73e7fe5..34e9c50 100644 --- a/snakes/PrismBattleSnake_GPT_5_6_Sol.py +++ b/snakes/PrismBattleSnake_GPT_5_6_Sol.py @@ -1,4 +1,4 @@ -"""PrismBattleSnake_GPT_5_6_Sol v1.0.0 +"""PrismBattleSnake_GPT_5_6_Sol v1.0.1 Built on ApexBattleSnake v1.0.0. All strategic logic is inherited. Performance improvement: all spatial primitives (flood fill, territory, @@ -41,7 +41,7 @@ _DIR_DELTAS = ((0, 1), (0, -1), (-1, 0), (1, 0)) _DIR_NAMES = ("up", "down", "left", "right") class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake): - VERSION = "1.0.0" + VERSION = "1.0.1" def __init__(self) -> None: super().__init__() @@ -83,6 +83,11 @@ class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake): my_snake = game_data.get_my_snake() my_len = my_snake.get("length", len(my_snake["body"])) food_set = {(f["x"], f["y"]) for f in game_data.get_food()} + all_occupied = { + (seg["x"], seg["y"]) + for snake in [my_snake, *other_snakes] + for seg in snake["body"] + } game_type = game_data.get_type() is_constrictor = game_type == "constrictor" w = bb.width @@ -100,7 +105,7 @@ class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake): if not is_constrictor and len(body) >= 2: tail_stacked = (body[-1]["x"] == body[-2]["x"] and body[-1]["y"] == body[-2]["y"]) if not tail_stacked: - can_grow = self._enemy_can_grow_this_turn(snake, food_set) + can_grow = self._enemy_can_grow_this_turn(snake, food_set, all_occupied) if not can_grow: enemy_tail_bits |= 1 << (body[-1]["y"] * w + body[-1]["x"]) @@ -354,17 +359,25 @@ class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake): return -5000.0 # ── Safe next options (enemy-attack aware) ──────────────────────── - # Remove tiles where an enemy of >= our length could head-to-head. - # The danger bitboard was precomputed; filter out tiles blocked by - # current body (enemy can't step there either). - danger_here = self._enemy_attack_danger & ~blocked_bits + # Rebuild danger for the simulated length. The root-turn danger mask is + # stale after eating and includes enemy moves blocked in this future body. + danger_here = 0 + for enemy in other_snakes: + enemy_len = enemy.get("length", len(enemy["body"])) + if enemy_len < body_len: + continue + enemy_head = enemy["head"] + enemy_idx = enemy_head["y"] * w + enemy_head["x"] + danger_here |= bb._neighbor_masks[enemy_idx] + danger_here &= ~blocked_bits safe_nb = nb_free & ~danger_here en_safe = safe_nb.bit_count() if en_safe == 0: return -4000.0 - sc = reachable * 1.9 + liberties * 14.0 + liberties * 11.0 + en_safe * 26.0 + next_opts = liberties + sc = reachable * 1.9 + liberties * 14.0 + next_opts * 11.0 + en_safe * 26.0 if en_safe == 1: sc -= 420.0 return sc diff --git a/snakes/__init__.py b/snakes/__init__.py index 7594166..6d0d97f 100644 --- a/snakes/__init__.py +++ b/snakes/__init__.py @@ -11,7 +11,7 @@ SNAKE_REGISTRY = { "UltimateBattleSnake": "4.5.0", "ApexBattleSnake": "1.0.0", "SupremeBattleSnake_ClaudeOpus4_6": "1.0.0", - "PrismBattleSnake_GPT_5_6_Sol": "1.0.0", + "PrismBattleSnake_GPT_5_6_Sol": "1.0.1", } DEFAULT_SNAKE_CONFIG = { diff --git a/snakes/bitboard.py b/snakes/bitboard.py index 4fba667..8feac3b 100644 --- a/snakes/bitboard.py +++ b/snakes/bitboard.py @@ -127,8 +127,9 @@ class BitBoard: ) -> int: """Simultaneous BFS from *my_idx* and all enemies. - Returns (my_cells − enemy_cells). Cells equidistant from both sides are - counted for neither (contested). + Returns Apex-compatible territory over cells reachable from ``my_idx``: + +1 when we arrive first, -1 when an enemy arrives first, and 0 for ties. + Enemy-only disconnected regions are not counted. """ if not enemy_indices: return 0 @@ -139,48 +140,44 @@ class BitBoard: nlc = self._not_leftcol my_front = 1 << my_idx - my_terr = my_front + my_seen = my_front en_front = 0 for ei in enemy_indices: en_front |= 1 << ei - en_terr = en_front + en_seen = en_front - remaining = free & ~my_terr & ~en_terr + # Each side must expand independently. A cell reached at the same depth is + # unclaimed, but it is not a wall: both sides may route through it later. + # Match Apex semantics by scoring only cells reachable from our head: + # ours when we arrive first, theirs when an enemy arrives first, and zero + # on ties. Enemy-only disconnected regions are intentionally ignored. + score = (my_front & ~en_front).bit_count() + enemy_before = 0 + while my_front: + my_exp = ( + ((my_front & nrc) << 1) + | ((my_front & nlc) >> 1) + | (my_front << w) + | (my_front >> w) + ) & free & ~my_seen + en_exp = ( + ((en_front & nrc) << 1) + | ((en_front & nlc) >> 1) + | (en_front << w) + | (en_front >> w) + ) & free & ~en_seen - while (my_front or en_front) and remaining: - # Expand both sides simultaneously (same BFS depth → ties go to neither) - my_exp = 0 - if my_front: - my_exp = ( - ((my_front & nrc) << 1) - | ((my_front & nlc) >> 1) - | (my_front << w) - | (my_front >> w) - ) & remaining - - en_exp = 0 - if en_front: - en_exp = ( - ((en_front & nrc) << 1) - | ((en_front & nlc) >> 1) - | (en_front << w) - | (en_front >> w) - ) & remaining - - # Contested cells (reached by both at the same depth) → neither claims - contested = my_exp & en_exp - my_exp &= ~contested - en_exp &= ~contested - - my_terr |= my_exp - en_terr |= en_exp - remaining &= ~(my_exp | en_exp | contested) + enemy_before |= en_front + score += (my_exp & ~enemy_before & ~en_exp).bit_count() + score -= (my_exp & enemy_before).bit_count() + my_seen |= my_exp + en_seen |= en_exp my_front = my_exp en_front = en_exp - return my_terr.bit_count() - en_terr.bit_count() + return score # ── Partition sizes (for articulation-point detection) ──────────────────── @@ -324,32 +321,42 @@ class BitBoard: if start_bit & food_bits: return 0, start_idx - frontier = start_bit - seen = frontier - dist = 0 + # Preserve Apex's deterministic up/down/left/right BFS tie-breaking. A + # pure bit frontier finds the right distance but selects the lowest flat + # index when several foods are equally close, which can change contested- + # food scoring and therefore the selected move. + queue = [start_idx] + distances = [0] + seen = start_bit + cursor = 0 w = self.width - nrc = self._not_rightcol - nlc = self._not_leftcol + size = self.size - while frontier: - dist += 1 - expanded = ( - ((frontier & nrc) << 1) - | ((frontier & nlc) >> 1) - | (frontier << w) - | (frontier >> w) - ) & free & ~seen - - if not expanded: - break - - hit = expanded & food_bits - if hit: - # Return the first (lowest-index) food cell found - first_bit = hit & (-hit) - return dist, first_bit.bit_length() - 1 - - seen |= expanded - frontier = expanded + while cursor < len(queue): + cell = queue[cursor] + dist = distances[cursor] + cursor += 1 + x = cell % w + candidates = ( + cell + w, + cell - w, + cell - 1, + cell + 1, + ) + for direction, neighbor in enumerate(candidates): + if neighbor < 0 or neighbor >= size: + continue + if direction == 2 and x == 0: + continue + if direction == 3 and x == w - 1: + continue + bit = 1 << neighbor + if bit & seen or not bit & free: + continue + if bit & food_bits: + return dist + 1, neighbor + seen |= bit + queue.append(neighbor) + distances.append(dist + 1) return None, None diff --git a/snakes/bitboard_duel_search.py b/snakes/bitboard_duel_search.py index 4a25367..f8ef026 100644 --- a/snakes/bitboard_duel_search.py +++ b/snakes/bitboard_duel_search.py @@ -129,8 +129,8 @@ class BitboardDuelSearch: if alpha >= beta: return cached_value, True - my_moves = self._legal_targets(state.my_body, state.enemy_body) - enemy_moves = self._legal_targets(state.enemy_body, state.my_body) + my_moves = self._candidate_targets(state.my_body) + enemy_moves = self._candidate_targets(state.enemy_body) if not my_moves: return self.LOSS - depth, True if not enemy_moves: @@ -219,14 +219,13 @@ class BitboardDuelSearch: ) return child, None - def _legal_targets(self, body: Body, other_body: Body) -> list[int]: - occupied = self._body_bits(body) | self._body_bits(other_body) - if not self._tail_stacked(body): - occupied &= ~(1 << body[-1]) - if not self._tail_stacked(other_body): - occupied &= ~(1 << other_body[-1]) - legal = self.board.neighbors_of(body[0]) & ~occupied & self.board.board_mask - return list(self._iter_bits(legal)) + def _candidate_targets(self, body: Body) -> list[int]: + """Return in-bounds targets; `_advance` resolves simultaneous collisions. + + Delaying occupancy checks until both targets and food growth are known is + essential: whether either tail vacates depends on that snake eating. + """ + return list(self._iter_bits(self.board.neighbors_of(body[0]))) def _ordered_moves(self, moves: list[int], state: DuelState, depth: int, mine: bool) -> list[int]: body = state.my_body if mine else state.enemy_body diff --git a/tests/snakes/test_PrismBattleSnake_GPT_5_6_Sol.py b/tests/snakes/test_PrismBattleSnake_GPT_5_6_Sol.py index 7ca49d1..3763a1c 100644 --- a/tests/snakes/test_PrismBattleSnake_GPT_5_6_Sol.py +++ b/tests/snakes/test_PrismBattleSnake_GPT_5_6_Sol.py @@ -21,20 +21,38 @@ class TestBitBoard(unittest.TestCase): self.assertEqual(board.territory(board.idx(0, 0), [board.idx(4, 0)], 0), 0) + def test_territory_propagates_through_contested_cells(self): + board = BitBoard(5, 3) + blocked = board.set_to_bits({(0, 1), (1, 1), (3, 1), (4, 1)}) + + self.assertEqual(board.territory(board.idx(0, 0), [board.idx(4, 0)], blocked), 0) + + def test_territory_ignores_enemy_only_disconnected_space_like_apex(self): + board = BitBoard(5, 1) + blocked = board.set_to_bits({(2, 0)}) + + self.assertEqual(board.territory(board.idx(0, 0), [board.idx(4, 0)], blocked), 2) + def test_nearest_food_returns_shortest_distance(self): board = BitBoard(5, 5) food = board.set_to_bits({(4, 4), (2, 1)}) self.assertEqual(board.nearest_food(board.idx(0, 0), food, 0), (3, board.idx(2, 1))) + def test_nearest_food_uses_apex_direction_order_for_ties(self): + board = BitBoard(3, 3) + food = board.set_to_bits({(1, 2), (0, 1), (2, 1), (1, 0)}) + + self.assertEqual(board.nearest_food(board.idx(1, 1), food, 0), (1, board.idx(1, 2))) + class TestPrismBattleSnake_GPT_5_6_Sol(unittest.TestCase): def test_api_name_and_version_are_exposed(self): snake = PrismBattleSnake_GPT_5_6_Sol() self.assertEqual(snake.name, "PrismBattleSnake") - self.assertEqual(snake.version, "1.0.0") - self.assertEqual(get_snake_version("PrismBattleSnake_GPT_5_6_Sol"), "1.0.0") + self.assertEqual(snake.version, "1.0.1") + self.assertEqual(get_snake_version("PrismBattleSnake_GPT_5_6_Sol"), "1.0.1") self.assertIsInstance(SnakeBuilder.build("PrismBattleSnake_GPT_5_6_Sol"), PrismBattleSnake_GPT_5_6_Sol) def test_bitboard_primitives_match_apex(self): @@ -82,6 +100,18 @@ class TestPrismBattleSnake_GPT_5_6_Sol(unittest.TestCase): self.assertGreater(value, 0) + def test_duel_search_keeps_tail_blocked_when_its_snake_eats(self): + board = BitBoard(3, 3) + search = BitboardDuelSearch( + board=board, food={(0, 1)}, hazards=set(), hazard_count={}, + hazard_damage=15, deadline=None, + ) + body = (board.idx(0, 0), board.idx(1, 0), board.idx(1, 1)) + + advanced = search._advance_body(body, board.idx(0, 1), ate=True) + + self.assertIn(board.idx(1, 1), advanced) + def test_bitboard_duel_search_reuses_transpositions(self): board = BitBoard(5, 5) search = BitboardDuelSearch( diff --git a/tests/test_DatabasePackageImports.py b/tests/test_DatabasePackageImports.py new file mode 100644 index 0000000..bde59d5 --- /dev/null +++ b/tests/test_DatabasePackageImports.py @@ -0,0 +1,32 @@ +import subprocess +import sys +import unittest +from pathlib import Path + +class TestDatabasePackageImports(unittest.TestCase): + def test_sqlite_backend_import_does_not_require_aiofiles(self): + project_root = Path(__file__).resolve().parents[1] + script = """ +import builtins + +original_import = builtins.__import__ +def reject_aiofiles(name, *args, **kwargs): + if name == 'aiofiles' or name.startswith('aiofiles.'): + raise ModuleNotFoundError("aiofiles intentionally unavailable") + return original_import(name, *args, **kwargs) + +builtins.__import__ = reject_aiofiles +from server.database.backend.SqliteGameplayBackend import SqliteGameplayBackend +assert SqliteGameplayBackend.__name__ == 'SqliteGameplayBackend' +""" + result = subprocess.run( + [sys.executable, "-c", script], + cwd=project_root, + text=True, + capture_output=True, + ) + + self.assertEqual(result.returncode, 0, result.stderr) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_MergeGameplayDatabases.py b/tests/test_MergeGameplayDatabases.py index a17cf58..f7242e5 100644 --- a/tests/test_MergeGameplayDatabases.py +++ b/tests/test_MergeGameplayDatabases.py @@ -27,10 +27,14 @@ class TestMergeGameplayDatabases(unittest.TestCase): 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) VALUES (?,?,?,?)", - (game_id, "me", "PrismBattleSnake", 1), - ) + 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, @@ -69,6 +73,13 @@ class TestMergeGameplayDatabases(unittest.TestCase): ]) 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): diff --git a/tests/test_MigrateGameplayDatabase.py b/tests/test_MigrateGameplayDatabase.py new file mode 100644 index 0000000..76ea139 --- /dev/null +++ b/tests/test_MigrateGameplayDatabase.py @@ -0,0 +1,84 @@ +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from scripts.migrate_gameplay_database import copy_game_snakes +from server.database.backend.SqliteGameplayBackend import SqliteGameplayBackend + +class TestMigrateGameplayDatabase(unittest.TestCase): + def test_copy_game_snakes_preserves_customizations(self): + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source_path = root / "source.sqlite3" + destination_path = root / "destination.sqlite3" + SqliteGameplayBackend(str(source_path)) + SqliteGameplayBackend(str(destination_path)) + + with sqlite3.connect(source_path) as source: + source.execute("PRAGMA foreign_keys = OFF") + source.execute(""" + INSERT INTO game_snakes ( + game_id, snake_id, snake_name, is_you, customizations_json + ) VALUES (?, ?, ?, ?, ?) + """, ( + "game-1", "snake-1", "PrismBattleSnake", 1, + '{"color":"#663399","head":"ferret","tail":"swirl"}', + )) + + source = sqlite3.connect(source_path) + destination = sqlite3.connect(destination_path) + try: + copied = copy_game_snakes( + source, destination, batch_size=10, retained_ids={"game-1"}, + ) + destination.commit() + row = destination.execute(""" + SELECT snake_name, is_you, customizations_json + FROM game_snakes WHERE game_id = ? AND snake_id = ? + """, ("game-1", "snake-1")).fetchone() + finally: + source.close() + destination.close() + + self.assertEqual(copied, 1) + self.assertEqual(row, ( + "PrismBattleSnake", 1, + '{"color":"#663399","head":"ferret","tail":"swirl"}', + )) + + def test_copy_game_snakes_defaults_legacy_schema_to_empty_customizations(self): + source = sqlite3.connect(":memory:") + destination = sqlite3.connect(":memory:") + try: + source.execute(""" + CREATE TABLE game_snakes ( + game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER + ) + """) + source.execute( + "INSERT INTO game_snakes VALUES (?, ?, ?, ?)", + ("game-1", "snake-1", "LegacySnake", 0), + ) + destination.execute(""" + CREATE TABLE game_snakes ( + game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER, + customizations_json TEXT NOT NULL DEFAULT '{}' + ) + """) + + copied = copy_game_snakes( + source, destination, batch_size=10, retained_ids={"game-1"}, + ) + row = destination.execute( + "SELECT customizations_json FROM game_snakes" + ).fetchone() + finally: + source.close() + destination.close() + + self.assertEqual(copied, 1) + self.assertEqual(row, ("{}",)) + +if __name__ == "__main__": + unittest.main() diff --git a/uv.lock b/uv.lock index 616843a..f33ee2e 100644 --- a/uv.lock +++ b/uv.lock @@ -345,6 +345,7 @@ name = "snake-python" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "aiofiles" }, { name = "aiologger" }, { name = "asyncpg" }, { name = "dotenv" }, @@ -356,6 +357,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "aiofiles", specifier = ">=25.1.0" }, { name = "aiologger", specifier = ">=0.7.0" }, { name = "asyncpg", specifier = ">=0.31.0" }, { name = "dotenv", specifier = ">=0.9.9" },