344 lines
17 KiB
Python
Executable File
344 lines
17 KiB
Python
Executable File
"""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_public_leaderboard_scores_ties_and_zero_press_players(self):
|
|
self.game.state = self.sample_state()
|
|
self.game.state["players"]["charlie"] = {"name": "Charlie"}
|
|
self.game.state["players"]["aaron"] = {"name": "aaron"}
|
|
self.game.state["presses"].append({
|
|
"player_id": "aaron", "at": 300, "remaining_ms": 10, "zone": "white", "points": 40,
|
|
})
|
|
self.assertEqual(self.game.public_leaderboard(), [
|
|
{"name": "aaron", "total_points": 40, "press_count": 1},
|
|
{"name": self.game.state["players"]["alice"]["name"], "total_points": 40, "press_count": 2},
|
|
{"name": "Bøb", "total_points": 16, "press_count": 1},
|
|
{"name": "Charlie", "total_points": 0, "press_count": 0},
|
|
])
|
|
|
|
def test_leaderboard_http_is_public_before_and_after_game_ends(self):
|
|
self.game.state = self.sample_state()
|
|
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 get(path):
|
|
connection = HTTPConnection(*httpd.server_address, timeout=5)
|
|
try:
|
|
connection.request("GET", path)
|
|
response = connection.getresponse()
|
|
return response.status, json.loads(response.read()), response.getheader("Cache-Control")
|
|
finally:
|
|
connection.close()
|
|
|
|
for ended in (False, True):
|
|
self.game.state["ended"] = ended
|
|
status, payload, cache = get("/api/leaderboard")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(cache, "no-store")
|
|
self.assertEqual(set(payload), {"leaderboard"})
|
|
self.assertEqual(payload["leaderboard"], self.game.public_leaderboard())
|
|
for player in payload["leaderboard"]:
|
|
self.assertEqual(set(player), {"name", "total_points", "press_count"})
|
|
self.assertEqual(get("/api/stats")[0], 200 if ended else 403)
|
|
self.game.state = self.game.new_game()
|
|
self.assertEqual(get("/api/leaderboard")[1], {"leaderboard": []})
|
|
|
|
def test_chat_auth_validation_cooldown_and_persistence(self):
|
|
self.game.load_state()
|
|
self.game.state = self.sample_state()
|
|
self.game.save_state(self.game.state)
|
|
self.game.token_index = {self.game.hash_token("test-chat-token"): "alice"}
|
|
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(method, data=None, authenticated=False):
|
|
connection = HTTPConnection(*httpd.server_address, timeout=5)
|
|
try:
|
|
headers = {"Content-Type": "application/json"}
|
|
if authenticated:
|
|
headers["Authorization"] = "Bearer test-chat-token"
|
|
connection.request(method, "/api/chat", json.dumps(data) if data is not None else None, headers)
|
|
response = connection.getresponse()
|
|
return response.status, json.loads(response.read())
|
|
finally:
|
|
connection.close()
|
|
|
|
self.assertEqual(request("GET"), (200, {"messages": []}))
|
|
self.assertEqual(request("POST", {"message": "Hello"})[0], 401)
|
|
for message in (None, 42, "", " ", "x" * 201, "hi\nthere", "bad\x00text"):
|
|
self.assertEqual(request("POST", {"message": message}, True)[0], 400)
|
|
with patch.object(self.game, "now_ms", return_value=10000):
|
|
status, posted = request("POST", {"message": " <b>Hello</b> '); -- ", "name": "Imposter"}, True)
|
|
self.assertEqual(status, 201)
|
|
self.assertEqual(posted["name"], self.game.state["players"]["alice"]["name"])
|
|
self.assertEqual(posted["message"], "<b>Hello</b> '); --")
|
|
self.assertEqual(set(posted), {"id", "name", "at", "message", "kind"})
|
|
self.assertEqual(request("POST", {"message": "Too soon"}, True)[0], 429)
|
|
# Snapshot writes delete/reinsert players; deferred chat FKs must survive.
|
|
self.game.save_state(self.game.state)
|
|
self.game.state = self.game.load_state()
|
|
self.assertEqual(request("POST", {"message": "Still too soon"}, True)[0], 429)
|
|
self.assertEqual(request("GET")[1], {"messages": [posted]})
|
|
self.game.state["ended"] = True
|
|
with patch.object(self.game, "now_ms", return_value=12000):
|
|
self.assertEqual(request("POST", {"message": "x" * 200}, True)[0], 201)
|
|
self.game.token_index = {}
|
|
self.assertEqual(request("POST", {"message": "Revoked"}, True)[0], 401)
|
|
with self.game.database() as connection:
|
|
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
|
|
|
def test_existing_chat_schema_upgrades_without_losing_messages(self):
|
|
self.game.load_state()
|
|
self.game.state = self.sample_state()
|
|
self.game.save_state(self.game.state)
|
|
with self.game.database() as connection:
|
|
connection.execute("DROP TABLE chat_messages")
|
|
connection.execute("""CREATE TABLE chat_messages (
|
|
id INTEGER PRIMARY KEY,
|
|
player_id TEXT NOT NULL REFERENCES players(id) DEFERRABLE INITIALLY DEFERRED,
|
|
name TEXT NOT NULL, at INTEGER NOT NULL, message TEXT NOT NULL
|
|
)""")
|
|
connection.execute(
|
|
"INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
|
|
("alice", "Alice", 123, "Existing message"),
|
|
)
|
|
self.game.load_state()
|
|
self.assertEqual(self.game.chat_messages(), [
|
|
{"id": 1, "name": "Alice", "at": 123, "message": "Existing message", "kind": "message"},
|
|
])
|
|
|
|
def test_chat_returns_latest_hundred_in_chronological_order(self):
|
|
self.game.load_state()
|
|
self.game.state = self.sample_state()
|
|
self.game.save_state(self.game.state)
|
|
with self.game.database() as connection:
|
|
connection.executemany(
|
|
"INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
|
|
[("alice", "Alice", index, str(index)) for index in range(105)],
|
|
)
|
|
messages = self.game.chat_messages()
|
|
self.assertEqual(len(messages), 100)
|
|
self.assertEqual([message["message"] for message in messages], [str(i) for i in range(5, 105)])
|
|
self.game.load_state()
|
|
self.assertEqual(messages, self.game.chat_messages())
|
|
|
|
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", {})[0], 401)
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
t = self.game.now_ms()
|
|
self.game.state["deadline"] = t + 1234
|
|
with patch.object(self.game, "now_ms", return_value=t):
|
|
self.assertEqual(request("/api/press", {}, token)[0], 200)
|
|
announcements = self.game.chat_messages()
|
|
self.assertEqual(len(announcements), 1)
|
|
self.assertEqual(announcements[0]["kind"], "press")
|
|
self.assertEqual(announcements[0]["name"], "The Button")
|
|
self.assertEqual(announcements[0]["message"], "Alice pushed the button with 00:00:01.234 remaining and received 32 points.")
|
|
self.assertEqual(request("/api/chat", {"message": "That was close!"}, token)[0], 201)
|
|
self.assertEqual(request("/api/subscribe", {"first_name": "Alice", "email": "alice@example.com"}, token)[0], 200)
|
|
self.game.state = self.game.load_state()
|
|
self.assertEqual(self.game.chat_messages()[0], announcements[0])
|
|
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)
|
|
self.assertEqual(len([m for m in self.game.chat_messages() if m["kind"] == "press"]), 2)
|
|
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()
|