"""Persistence and migration regression tests; never use the live game files.""" import copy import importlib.util import json import os import sqlite3 import tempfile import threading import unittest from http.client import HTTPConnection from http.server import ThreadingHTTPServer from pathlib import Path from unittest.mock import patch class PersistenceTests(unittest.TestCase): @classmethod def setUpClass(cls): with tempfile.TemporaryDirectory() as directory: with patch.dict(os.environ, { "BUTTON_DATA_FILE": str(Path(directory) / "data.json"), "BUTTON_DATABASE_FILE": str(Path(directory) / "data.sqlite3"), "BUTTON_MAILING_LIST_FILE": str(Path(directory) / "mailing.csv"), }): spec = importlib.util.spec_from_file_location("button_test_server", Path(__file__).with_name("server.py")) cls.game = importlib.util.module_from_spec(spec) spec.loader.exec_module(cls.game) def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) root = Path(self.directory.name) self.game.DATA_FILE = root / "data.json" self.game.DATABASE_FILE = root / "data.sqlite3" self.game.MAILING_LIST_FILE = root / "mailing.csv" self.game.failed_logins.clear() self.game.subscribed_emails.clear() def sample_state(self): state = self.game.new_game() state["players"] = { "alice": { "name": "Alice '); DROP TABLE players; --", "password_hash": "existing-password-hash", "token_hashes": ["first-session", "second-session"], "created_at": 123, "subscribed": True, }, "bob": { "name": "Bøb", "password_hash": None, "token_hashes": [], "created_at": 456, "subscribed": False, }, } state["presses"] = [ {"player_id": "alice", "at": 100, "remaining_ms": 20, "zone": "white", "points": 32}, {"player_id": "bob", "at": 100, "remaining_ms": 50, "zone": "black", "points": 16}, {"player_id": "alice", "at": 200, "remaining_ms": 90, "zone": "red", "points": 8}, ] return state def write_legacy(self, state): self.game.DATA_FILE.write_text(json.dumps(state), encoding="utf-8") def test_fresh_game_survives_restart_without_creating_json(self): state = self.game.load_state() self.assertEqual(state, self.game.load_state()) self.assertTrue(self.game.DATABASE_FILE.exists()) self.assertFalse(self.game.DATA_FILE.exists()) def test_migration_preserves_all_game_data_and_source(self): expected = self.sample_state() expected["ended"] = True expected["ended_at"] = expected["deadline"] self.write_legacy(expected) original = self.game.DATA_FILE.read_bytes() self.assertEqual(expected, self.game.load_state()) self.assertEqual(expected, self.game.load_state()) self.assertEqual(original, self.game.DATA_FILE.read_bytes()) with self.game.database() as connection: self.assertEqual(connection.execute("PRAGMA integrity_check").fetchone()[0], "ok") self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), []) def test_existing_database_ignores_legacy_even_if_invalid(self): self.write_legacy(self.sample_state()) current = self.game.load_state() current["deadline"] += 12345 current["players"]["alice"]["token_hashes"] = ["new-session"] self.game.save_state(current) self.game.DATA_FILE.write_text("not valid JSON", encoding="utf-8") self.assertEqual(current, self.game.load_state()) def test_single_token_pre_login_players_migrate(self): source = self.sample_state() alice = source["players"]["alice"] del alice["password_hash"], alice["token_hashes"] alice["token_hash"] = "old-session" del source["players"]["bob"]["subscribed"] self.write_legacy(source) loaded = self.game.load_state() self.assertEqual(loaded["players"]["alice"]["token_hashes"], ["old-session"]) reloaded = self.game.load_state() self.assertIsNone(reloaded["players"]["alice"]["password_hash"]) self.assertFalse(reloaded["players"]["bob"]["subscribed"]) self.assertEqual(reloaded["presses"], source["presses"]) def test_empty_database_still_imports_json(self): self.game.DATABASE_FILE.touch() expected = self.sample_state() self.write_legacy(expected) self.assertEqual(expected, self.game.load_state()) def test_invalid_json_fails_without_starting_new_game_and_can_retry(self): self.game.DATA_FILE.write_text("{broken", encoding="utf-8") for _ in range(2): with self.assertRaises(json.JSONDecodeError): self.game.load_state() expected = self.sample_state() self.write_legacy(expected) self.assertEqual(expected, self.game.load_state()) def test_failed_import_rolls_back_and_can_retry(self): invalid = self.sample_state() invalid["presses"][-1]["player_id"] = "missing-player" self.write_legacy(invalid) with self.assertRaises(sqlite3.IntegrityError): self.game.load_state() with self.game.database() as connection: for table in ("game", "players", "sessions", "presses"): self.assertEqual(connection.execute("SELECT COUNT(*) FROM " + table).fetchone()[0], 0) expected = self.sample_state() self.write_legacy(expected) self.assertEqual(expected, self.game.load_state()) def test_failed_save_retains_previous_database_snapshot(self): self.write_legacy(self.sample_state()) previous = self.game.load_state() invalid = copy.deepcopy(previous) invalid["presses"][0]["player_id"] = "missing-player" with self.assertRaises(sqlite3.IntegrityError): self.game.save_state(invalid) self.assertEqual(previous, self.game.load_state()) def test_database_and_source_cannot_be_same_file(self): self.write_legacy(self.sample_state()) original = self.game.DATA_FILE.read_bytes() self.game.DATABASE_FILE = self.game.DATA_FILE with self.assertRaises(ValueError): self.game.load_state() self.assertEqual(original, self.game.DATA_FILE.read_bytes()) def test_nested_database_path_is_created(self): self.game.DATABASE_FILE = Path(self.directory.name) / "nested" / "game.sqlite3" initial = self.game.load_state() self.assertEqual(initial, self.game.load_state()) def test_http_registration_login_press_logout_and_restart(self): self.game.state = self.game.load_state() self.game.token_index = {} httpd = ThreadingHTTPServer(("127.0.0.1", 0), self.game.Handler) thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() self.addCleanup(httpd.server_close) self.addCleanup(thread.join) self.addCleanup(httpd.shutdown) def request(path, data, token=None): connection = HTTPConnection(*httpd.server_address, timeout=5) try: headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = "Bearer " + token connection.request("POST", path, json.dumps(data), headers) response = connection.getresponse() return response.status, json.loads(response.read()) finally: connection.close() status, registered = request("/api/register", {"name": "Alice", "password": "test-password"}) self.assertEqual(status, 201) token = registered["token"] self.assertEqual(request("/api/press", {}, token)[0], 200) self.assertEqual(request("/api/subscribe", {"first_name": "Alice", "email": "alice@example.com"}, token)[0], 200) self.game.state = self.game.load_state() self.game.token_index = {th: pid for pid, p in self.game.state["players"].items() for th in p["token_hashes"]} self.assertEqual(request("/api/press", {}, token)[0], 200) status, logged_in = request("/api/login", {"name": "Alice", "password": "test-password"}) self.assertEqual(status, 200) self.assertEqual(request("/api/logout", {}, token)[0], 200) self.assertEqual(request("/api/press", {}, token)[0], 401) loaded = self.game.load_state() player = next(iter(loaded["players"].values())) self.assertTrue(player["subscribed"]) self.assertEqual(player["token_hashes"], [self.game.hash_token(logged_in["token"])]) self.assertEqual(len(loaded["presses"]), 2) if __name__ == "__main__": unittest.main()