504 lines
27 KiB
Python
Executable File
504 lines
27 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()
|
|
self.game.presence.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 presence_client(self):
|
|
self.game.load_state()
|
|
self.game.state = self.sample_state()
|
|
self.game.state["players"]["alice"]["token_hashes"] = [self.game.hash_token("alice-token"), self.game.hash_token("alice-second")]
|
|
self.game.state["players"]["bob"]["token_hashes"] = [self.game.hash_token("bob-token")]
|
|
self.game.save_state(self.game.state)
|
|
self.game.token_index = {
|
|
th: pid for pid, player in self.game.state["players"].items() for th in player["token_hashes"]
|
|
}
|
|
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="GET", data=None, token=None, path="/api/presence"):
|
|
connection = HTTPConnection(*httpd.server_address, timeout=5)
|
|
try:
|
|
headers = {"Content-Type": "application/json"}
|
|
if token:
|
|
headers["Authorization"] = "Bearer " + token
|
|
connection.request(method, path, json.dumps(data) if data is not None else None, headers)
|
|
response = connection.getresponse()
|
|
return response.status, json.loads(response.read())
|
|
finally:
|
|
connection.close()
|
|
|
|
return request
|
|
|
|
def test_presence_auth_validation_and_default_privacy(self):
|
|
request = self.presence_client()
|
|
self.assertEqual(request(), (200, {"online_count": 0, "chat_players": []}))
|
|
update = {"tab_id": "a" * 32, "active": True, "in_chat": True}
|
|
self.assertEqual(request("POST", update)[0], 401)
|
|
self.assertEqual(request("POST", update, "not-a-token")[0], 401)
|
|
for invalid in ({}, {**update, "active": 1}, {**update, "in_chat": "true"},
|
|
{**update, "tab_id": "a" * 33}, {**update, "tab_id": "<script>"},
|
|
{**update, "tab_id": None}):
|
|
self.assertEqual(request("POST", invalid, "alice-token")[0], 400)
|
|
self.assertEqual(request()[1], {"online_count": 0, "chat_players": []})
|
|
self.assertNotIn("presence", self.game.load_state())
|
|
|
|
def test_presence_deduplicates_tabs_sessions_and_updates_chat(self):
|
|
request = self.presence_client()
|
|
update = {"tab_id": "a" * 32, "active": True, "in_chat": False}
|
|
self.assertEqual(request("POST", update, "alice-token")[1], {"online_count": 1, "chat_players": []})
|
|
request("POST", {**update, "tab_id": "b" * 32, "in_chat": True}, "alice-token")
|
|
request("POST", {**update, "in_chat": True}, "alice-second")
|
|
request("POST", {**update, "in_chat": True, "name": "Imposter"}, "bob-token")
|
|
expected = {"online_count": 2, "chat_players": [self.game.state["players"]["alice"]["name"], "Bøb"]}
|
|
self.assertEqual(request()[1], expected)
|
|
request("POST", {**update, "tab_id": "b" * 32, "active": False}, "alice-token")
|
|
self.assertEqual(request()[1], expected) # other session still viewing chat
|
|
request("POST", update, "alice-second")
|
|
self.assertEqual(request()[1], {"online_count": 2, "chat_players": ["Bøb"]})
|
|
request("POST", {**update, "active": False}, "alice-token")
|
|
self.assertEqual(request()[1]["online_count"], 2) # second session still online
|
|
request("POST", {**update, "active": False}, "alice-second")
|
|
self.assertEqual(request()[1], {"online_count": 1, "chat_players": ["Bøb"]})
|
|
|
|
def test_presence_expires_and_revoked_sessions_disappear(self):
|
|
request = self.presence_client()
|
|
update = {"tab_id": "a" * 32, "active": True, "in_chat": True}
|
|
with patch.object(self.game.time, "monotonic", return_value=100):
|
|
request("POST", update, "alice-token")
|
|
with patch.object(self.game.time, "monotonic", return_value=144):
|
|
self.assertEqual(request()[1]["online_count"], 1)
|
|
request("POST", update, "alice-token") # heartbeat renews TTL
|
|
with patch.object(self.game.time, "monotonic", return_value=188):
|
|
self.assertEqual(request()[1]["online_count"], 1)
|
|
with patch.object(self.game.time, "monotonic", return_value=189):
|
|
self.assertEqual(request()[1], {"online_count": 0, "chat_players": []})
|
|
request("POST", update, "alice-token")
|
|
self.game.token_index.pop(self.game.hash_token("alice-token"))
|
|
self.assertEqual(request()[1]["online_count"], 0)
|
|
self.assertEqual(self.game.presence, {})
|
|
|
|
def test_presence_bounded_tabs_and_logout_cleanup(self):
|
|
request = self.presence_client()
|
|
for index in range(self.game.MAX_PRESENCE_TABS_PER_SESSION):
|
|
update = {"tab_id": f"{index:032x}", "active": True, "in_chat": True}
|
|
self.assertEqual(request("POST", update, "alice-token")[0], 200)
|
|
update = {"tab_id": "f" * 32, "active": True}
|
|
self.assertEqual(request("POST", update, "alice-token")[0], 429)
|
|
self.assertEqual(request("POST", {**update, "tab_id": "0" * 32}, "alice-token")[0], 200)
|
|
self.assertEqual(request("POST", {}, "alice-token", "/api/logout")[0], 200)
|
|
self.assertEqual(request()[1], {"online_count": 0, "chat_players": []})
|
|
self.assertEqual(self.game.presence, {})
|
|
|
|
def test_personal_bests_use_only_own_presses_and_existing_history(self):
|
|
self.game.state = self.sample_state()
|
|
summary = self.game.player_summary("alice")
|
|
self.assertEqual(summary["personal_bests"]["closest_call"], {
|
|
"at": 100, "remaining_ms": 20, "zone": "white", "points": 32,
|
|
})
|
|
self.assertEqual(summary["personal_bests"]["highest_scoring_press"]["points"], 32)
|
|
bob = self.game.player_summary("bob")
|
|
self.assertEqual(bob["personal_bests"]["closest_call"]["remaining_ms"], 50)
|
|
self.assertEqual(bob["personal_bests"]["highest_scoring_press"]["points"], 16)
|
|
self.assertEqual(bob["ranking"], {
|
|
"position": 2, "total_players": 2, "tied_for_lead": False,
|
|
"rival": {"name": summary["name"], "total_points": 40, "points_to_tie": 24, "points_to_pass": 25},
|
|
})
|
|
self.assertIsNone(summary["ranking"]["rival"])
|
|
self.assertFalse(summary["ranking"]["tied_for_lead"])
|
|
self.game.load_state()
|
|
self.game.save_state(self.game.state)
|
|
self.game.state = self.game.load_state()
|
|
self.assertEqual(self.game.player_summary("alice")["personal_bests"], summary["personal_bests"])
|
|
|
|
def test_personal_records_empty_zero_points_and_deterministic_ties(self):
|
|
self.game.state = self.game.new_game()
|
|
self.game.state["players"] = {"solo": {"name": "Solo"}}
|
|
summary = self.game.player_summary("solo")
|
|
self.assertEqual(summary["personal_bests"], {"closest_call": None, "highest_scoring_press": None})
|
|
self.assertEqual(summary["ranking"], {"position": 1, "total_players": 1, "tied_for_lead": False, "rival": None})
|
|
self.game.state["presses"] = [
|
|
{"player_id": "solo", "at": at, "remaining_ms": remaining, "zone": "green", "points": 0}
|
|
for at, remaining in [(100, 6000), (300, 5000), (200, 5000)]
|
|
]
|
|
records = self.game.player_summary("solo")["personal_bests"]
|
|
self.assertEqual(records["highest_scoring_press"]["points"], 0)
|
|
self.assertEqual(records["highest_scoring_press"]["at"], 200)
|
|
self.assertEqual(records["closest_call"]["at"], 200)
|
|
|
|
def test_rivals_follow_alphabetical_ties_and_shared_lead(self):
|
|
self.game.state = self.game.new_game()
|
|
self.game.state["players"] = {"bob": {"name": "Bob"}, "alice": {"name": "alice"}, "charlie": {"name": "Charlie"}}
|
|
alice = self.game.player_summary("alice")["ranking"]
|
|
bob = self.game.player_summary("bob")["ranking"]
|
|
charlie = self.game.player_summary("charlie")["ranking"]
|
|
self.assertEqual(alice["position"], 1)
|
|
self.assertTrue(alice["tied_for_lead"])
|
|
self.assertIsNone(alice["rival"])
|
|
self.assertEqual(bob["position"], 2)
|
|
self.assertEqual(bob["rival"], {"name": "alice", "total_points": 0, "points_to_tie": 0, "points_to_pass": 1})
|
|
self.assertEqual(charlie["rival"]["name"], "Bob")
|
|
self.game.state["players"]["bob"]["name"] = "alice" # legacy duplicate names remain distinct IDs
|
|
self.assertNotEqual(self.game.player_summary("bob")["ranking"]["position"], self.game.player_summary("alice")["ranking"]["position"])
|
|
|
|
def test_rival_changes_when_another_player_overtakes(self):
|
|
self.game.state = self.sample_state()
|
|
self.game.state["presses"].append({"player_id": "bob", "at": 300, "remaining_ms": 1, "zone": "white", "points": 32})
|
|
summary = self.game.player_summary("alice")
|
|
self.assertEqual(summary["ranking"]["rival"], {"name": "Bøb", "total_points": 48, "points_to_tie": 8, "points_to_pass": 9})
|
|
self.assertEqual(summary["personal_bests"]["highest_scoring_press"]["points"], 32)
|
|
public = self.game.public_leaderboard()
|
|
for row in public:
|
|
self.assertEqual(set(row), {"name", "total_points", "press_count"})
|
|
|
|
def test_personal_records_endpoint_requires_authentication(self):
|
|
request = self.presence_client()
|
|
self.assertEqual(request(path="/api/me")[0], 401)
|
|
status, summary = request(token="bob-token", path="/api/me")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(summary["personal_bests"]["highest_scoring_press"]["points"], 16)
|
|
self.assertEqual(set(summary["ranking"]["rival"]), {"name", "total_points", "points_to_tie", "points_to_pass"})
|
|
|
|
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()
|