1238 lines
73 KiB
Python
Executable File
1238 lines
73 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 datetime import datetime, timezone
|
|
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,
|
|
"is_admin": False, "is_banned": False, "is_deleted": False,
|
|
},
|
|
"bob": {
|
|
"name": "Bøb",
|
|
"password_hash": None,
|
|
"token_hashes": [],
|
|
"created_at": 456,
|
|
"subscribed": False,
|
|
"is_admin": False, "is_banned": False, "is_deleted": 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 admin_client(self):
|
|
request = self.presence_client()
|
|
self.game.state["players"]["alice"]["is_admin"] = True
|
|
self.game.save_state(self.game.state)
|
|
return request
|
|
|
|
def test_admin_endpoints_deny_guests_and_regular_players(self):
|
|
request = self.admin_client()
|
|
for path in ("/api/admin",):
|
|
self.assertEqual(request(path=path)[0], 401)
|
|
self.assertEqual(request(token="bob-token", path=path)[0], 403)
|
|
for action in ("set_admin", "set_banned", "delete_player", "delete_message"):
|
|
payload = {"action": action, "target_id": "alice", "value": True, "at": 1}
|
|
self.assertEqual(request("POST", payload, path="/api/admin/action")[0], 401)
|
|
self.assertEqual(request("POST", payload, "bob-token", "/api/admin/action")[0], 403)
|
|
self.assertTrue(self.game.state["players"]["alice"]["is_admin"])
|
|
|
|
def test_admin_overview_does_not_expose_credentials_and_role_is_private(self):
|
|
request = self.admin_client()
|
|
status, dashboard = request(token="alice-token", path="/api/admin")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(dashboard["actor_id"], "alice")
|
|
self.assertEqual(len(dashboard["players"]), 2)
|
|
raw = json.dumps(dashboard)
|
|
for secret in ("password_hash", "token_hash", "existing-password-hash", "first-session"):
|
|
self.assertNotIn(secret, raw)
|
|
self.assertTrue(request(token="alice-token", path="/api/me")[1]["is_admin"])
|
|
for row in self.game.public_leaderboard() + self.game.global_stats()["leaderboard"]:
|
|
self.assertNotIn("is_admin", row)
|
|
|
|
def test_admin_role_changes_revoke_sessions_and_survive_save_reload(self):
|
|
request = self.admin_client()
|
|
payload = {"action": "set_admin", "target_id": "bob", "value": True}
|
|
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
|
|
self.assertTrue(self.game.state["players"]["bob"]["is_admin"])
|
|
self.assertEqual(request(token="bob-token", path="/api/admin")[0], 401)
|
|
self.assertEqual(self.game.state["players"]["bob"]["token_hashes"], [])
|
|
self.game.save_state(self.game.state)
|
|
self.game.state = self.game.load_state()
|
|
self.assertTrue(self.game.state["players"]["bob"]["is_admin"])
|
|
# Simulate a new authenticated session, then demote it from a different admin.
|
|
self.game.token_index[self.game.hash_token("bob-new")] = "bob"
|
|
self.game.state["players"]["bob"]["token_hashes"] = [self.game.hash_token("bob-new")]
|
|
self.assertEqual(request(token="bob-new", path="/api/admin")[0], 200)
|
|
self.assertEqual(request("POST", {**payload, "value": False}, "alice-token", "/api/admin/action")[0], 200)
|
|
self.assertEqual(request(token="bob-new", path="/api/admin")[0], 401)
|
|
|
|
def test_ban_revokes_all_access_and_unban_requires_new_login(self):
|
|
request = self.admin_client()
|
|
self.game.state["players"]["bob"]["password_hash"] = self.game.hash_password("test-password")
|
|
self.game.save_state(self.game.state)
|
|
update = {"tab_id": "a" * 32, "active": True, "in_chat": True}
|
|
self.assertEqual(request("POST", update, "bob-token")[0], 200)
|
|
payload = {"action": "set_banned", "target_id": "bob", "value": True}
|
|
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
|
|
for path, method, data in (("/api/me", "GET", None), ("/api/press", "POST", {}),
|
|
("/api/chat", "POST", {"message": "Banned"}), ("/api/presence", "POST", update),
|
|
("/api/subscribe", "POST", {})):
|
|
self.assertEqual(request(method, data, "bob-token", path)[0], 401)
|
|
self.assertEqual(request()[1]["online_count"], 0)
|
|
self.assertEqual(request("POST", {"name": "Bøb", "password": "test-password"}, path="/api/login")[0], 401)
|
|
self.assertEqual(request("POST", {"action": "set_admin", "target_id": "bob", "value": True}, "alice-token", "/api/admin/action")[0], 409)
|
|
self.assertEqual(request("POST", {**payload, "value": False}, "alice-token", "/api/admin/action")[0], 200)
|
|
self.assertFalse(self.game.state["players"]["bob"]["is_banned"])
|
|
self.assertEqual(request(token="bob-token", path="/api/me")[0], 401)
|
|
self.assertEqual(request("POST", {"name": "Bøb", "password": "test-password"}, path="/api/login")[0], 200)
|
|
|
|
def test_admin_self_protection_and_last_active_admin(self):
|
|
request = self.admin_client()
|
|
for action, value in (("set_admin", False), ("set_banned", True), ("delete_player", None)):
|
|
payload = {"action": action, "target_id": "alice", "value": value}
|
|
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 409)
|
|
self.assertEqual(sum(self.game.active_admin(p) for p in self.game.state["players"].values()), 1)
|
|
self.assertEqual(request("POST", {"action": "set_admin", "target_id": "bob", "value": "true"}, "alice-token", "/api/admin/action")[0], 400)
|
|
self.assertEqual(request("POST", {"action": "delete_player", "target_id": "missing"}, "alice-token", "/api/admin/action")[0], 404)
|
|
self.assertEqual(request("POST", {"action": "execute_sql", "target_id": "bob"}, "alice-token", "/api/admin/action")[0], 400)
|
|
|
|
def test_admin_deletes_chat_without_id_reuse_or_cooldown_bypass(self):
|
|
request = self.admin_client()
|
|
with self.game.database() as connection:
|
|
at = self.game.now_ms()
|
|
cursor = connection.execute("INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
|
|
("bob", "Bøb", at, "Remove this"))
|
|
mid = cursor.lastrowid
|
|
payload = {"action": "delete_message", "target_id": str(mid), "at": at}
|
|
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
|
|
self.assertFalse(any(m["id"] == mid for m in self.game.chat_messages()))
|
|
self.assertEqual(request("POST", {"message": "Too soon"}, "bob-token", "/api/chat")[0], 429)
|
|
with self.game.database() as connection:
|
|
self.assertEqual(connection.execute("SELECT message FROM chat_messages WHERE id = ?", (mid,)).fetchone()[0], "[Message removed]")
|
|
self.game.load_state()
|
|
self.assertFalse(any(m["id"] == mid for m in self.game.chat_messages()))
|
|
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 404)
|
|
|
|
def test_admin_can_remove_system_milestone_without_reannouncing_it(self):
|
|
request = self.admin_client()
|
|
at = self.game.now_ms()
|
|
with self.game.database() as connection:
|
|
connection.execute("UPDATE community_milestones SET reached_at = ?, announced_at = ?, message = ? WHERE id = 'ten_players'", (at, at, "Milestone to remove"))
|
|
payload = {"action": "delete_message", "target_id": "milestone:ten_players", "at": at}
|
|
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
|
|
self.game.check_ended()
|
|
self.assertFalse(any(m["message"] == "Milestone to remove" for m in self.game.chat_messages()))
|
|
with self.game.database() as connection:
|
|
self.assertIsNotNone(connection.execute("SELECT reached_at FROM community_milestones WHERE id = 'ten_players'").fetchone()[0])
|
|
boundary = self.game.current_season(connection)["ends_at"]
|
|
self.game.check_ended(boundary)
|
|
self.assertFalse(any(m["at"] == at and m["kind"] == "milestone" for m in self.game.chat_messages()))
|
|
|
|
def test_admin_account_deletion_anonymizes_without_losing_game_history(self):
|
|
request = self.admin_client()
|
|
before_presses = copy.deepcopy(self.game.state["presses"])
|
|
with self.game.database() as connection:
|
|
season = self.game.current_season(connection)
|
|
self.game.archive_season(connection, season, self.game.state, self.game.state["deadline"], "countdown")
|
|
connection.execute("INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)", ("bob", "Bøb", 123, "My message"))
|
|
self.assertEqual(request("POST", {"action": "delete_player", "target_id": "bob"}, "alice-token", "/api/admin/action")[0], 200)
|
|
player = self.game.state["players"]["bob"]
|
|
self.assertTrue(player["is_deleted"])
|
|
self.assertEqual(player["name"], "Deleted player")
|
|
self.assertIsNone(player["password_hash"])
|
|
self.assertEqual(player["token_hashes"], [])
|
|
self.assertIsNone(self.game.find_player_by_name("Bøb"))
|
|
self.assertEqual(self.game.state["presses"], before_presses)
|
|
self.assertEqual(request(token="bob-token", path="/api/me")[0], 401)
|
|
self.assertNotIn("bob", [p["id"] for p in request(token="alice-token", path="/api/admin")[1]["players"]])
|
|
self.assertNotIn("Bøb", json.dumps(self.game.seasonal_leaderboard(), ensure_ascii=False))
|
|
self.assertTrue(any(m["message"] == "[Account deleted]" for m in self.game.chat_messages()))
|
|
self.game.state = self.game.load_state()
|
|
self.assertTrue(self.game.state["players"]["bob"]["is_deleted"])
|
|
with self.game.database() as connection:
|
|
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
|
self.assertEqual(connection.execute("SELECT action, actor_id FROM admin_audit").fetchone()["action"], "delete_player")
|
|
|
|
def test_admin_action_rollback_does_not_revoke_or_mutate(self):
|
|
request = self.admin_client()
|
|
before = copy.deepcopy(self.game.state)
|
|
with self.game.database() as connection:
|
|
connection.execute("""CREATE TRIGGER fail_admin_audit BEFORE INSERT ON admin_audit
|
|
BEGIN SELECT RAISE(ABORT, 'test rollback'); END""")
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.game.admin_action("alice", {"action": "set_banned", "target_id": "bob", "value": True})
|
|
self.assertEqual(self.game.state, before)
|
|
self.assertIn(self.game.hash_token("bob-token"), self.game.token_index)
|
|
self.assertEqual(request(token="bob-token", path="/api/me")[0], 200)
|
|
with self.game.database() as connection:
|
|
connection.execute("DROP TRIGGER fail_admin_audit")
|
|
self.assertEqual(self.game.load_state(), before)
|
|
|
|
def test_public_registration_cannot_grant_admin_or_bypass_ban_flags(self):
|
|
request = self.admin_client()
|
|
data = {"name": "Newcomer", "password": "test-password", "is_admin": True, "is_banned": False}
|
|
status, result = request("POST", data, path="/api/register")
|
|
self.assertEqual(status, 201)
|
|
pid = self.game.find_player_by_name("Newcomer")
|
|
self.assertFalse(self.game.state["players"][pid].get("is_admin", False))
|
|
self.assertEqual(request(token=result["token"], path="/api/admin")[0], 403)
|
|
|
|
def test_admin_schema_upgrade_and_database_provisioning(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("PRAGMA foreign_keys = OFF")
|
|
connection.execute("""CREATE TABLE old_players (
|
|
id TEXT PRIMARY KEY, name TEXT NOT NULL, password_hash TEXT,
|
|
created_at INTEGER NOT NULL, subscribed INTEGER NOT NULL
|
|
)""")
|
|
connection.execute("INSERT INTO old_players SELECT id, name, password_hash, created_at, subscribed FROM players")
|
|
connection.execute("DROP TABLE players")
|
|
connection.execute("ALTER TABLE old_players RENAME TO players")
|
|
loaded = self.game.load_state()
|
|
self.assertFalse(any(p["is_admin"] for p in loaded["players"].values()))
|
|
with self.game.database() as connection:
|
|
connection.execute("UPDATE players SET is_admin = 1 WHERE id = ?", ("alice",))
|
|
self.game.state = self.game.load_state()
|
|
self.assertTrue(self.game.state["players"]["alice"]["is_admin"])
|
|
self.game.save_state(self.game.state)
|
|
self.assertTrue(self.game.load_state()["players"]["alice"]["is_admin"])
|
|
with self.game.database() as connection:
|
|
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
|
|
|
def test_legacy_json_cannot_provision_administrators(self):
|
|
legacy = self.sample_state()
|
|
legacy["players"]["alice"].update(is_admin=True, is_banned=True, is_deleted=True)
|
|
self.write_legacy(legacy)
|
|
self.game.state = self.game.load_state()
|
|
for field in ("is_admin", "is_banned", "is_deleted"):
|
|
self.assertFalse(self.game.state["players"]["alice"][field])
|
|
self.assertFalse(self.game.load_state()["players"]["alice"][field])
|
|
|
|
def test_admin_malformed_request_and_cross_site_mutation(self):
|
|
request = self.admin_client()
|
|
self.assertEqual(request("POST", {}, "alice-token", "/api/admin/unknown")[0], 404)
|
|
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)
|
|
headers = {"Authorization": "Bearer alice-token", "Content-Type": "application/json"}
|
|
for body, extra, expected in (("{broken", {}, 400), ("[]", {}, 400),
|
|
(json.dumps({"action": "set_banned", "target_id": "bob", "value": True}), {"Sec-Fetch-Site": "cross-site"}, 403)):
|
|
connection = HTTPConnection(*httpd.server_address, timeout=5)
|
|
try:
|
|
connection.request("POST", "/api/admin/action", body, {**headers, **extra})
|
|
response = connection.getresponse()
|
|
self.assertEqual(response.status, expected)
|
|
response.read()
|
|
finally:
|
|
connection.close()
|
|
self.assertFalse(self.game.state["players"]["bob"]["is_banned"])
|
|
|
|
def season_fixture(self):
|
|
at = int(datetime(2026, 10, 21, tzinfo=timezone.utc).timestamp() * 1000)
|
|
with patch.object(self.game, "now_ms", return_value=at):
|
|
self.game.state = self.game.load_state()
|
|
return at
|
|
|
|
def test_astronomical_calendar_known_usno_dates_and_exact_midnight(self):
|
|
# USNO API: https://aa.usno.navy.mil/api/seasons?year=2026 (also 2027).
|
|
for year, dates in ((2026, ("03-20", "06-21", "09-23", "12-21")),
|
|
(2027, ("03-20", "06-21", "09-23", "12-22"))):
|
|
boundaries = self.game.season_boundaries(year)
|
|
self.assertEqual([b["id"] for b in boundaries], [f"{year}-{date}" for date in dates])
|
|
for boundary in boundaries:
|
|
self.assertEqual(boundary["starts_at"] % 86400000, 0)
|
|
self.assertEqual(self.game.season_for_time(boundary["starts_at"])["id"], boundary["id"])
|
|
self.assertNotEqual(self.game.season_for_time(boundary["starts_at"] - 1)["id"], boundary["id"])
|
|
january = int(datetime(2027, 1, 1, tzinfo=timezone.utc).timestamp() * 1000)
|
|
self.assertEqual(self.game.season_for_time(january)["id"], "2026-12-21")
|
|
with self.assertRaises(ValueError):
|
|
self.game.season_for_time(0)
|
|
|
|
def test_season_timeout_archives_winner_and_waits_until_boundary(self):
|
|
at = self.season_fixture()
|
|
state = self.sample_state()
|
|
state.update(started_at=at, deadline=at + 10000, ended=False, ended_at=None)
|
|
self.game.state = state
|
|
self.game.save_state(state)
|
|
with patch.object(self.game, "now_ms", return_value=at + 10000):
|
|
self.game.check_ended()
|
|
result = self.game.seasonal_leaderboard()
|
|
self.assertTrue(self.game.state["ended"])
|
|
archive = result["seasons"][0]
|
|
self.assertEqual(archive["duration_ms"], 10000)
|
|
self.assertEqual(archive["winners"][0]["total_points"], 40)
|
|
self.assertEqual(archive["reason"], "countdown")
|
|
with patch.object(self.game, "now_ms", return_value=at + 20000):
|
|
self.game.check_ended()
|
|
self.game.state = self.game.load_state()
|
|
self.game.check_ended()
|
|
self.assertEqual(self.game.seasonal_leaderboard()["seasons"], result["seasons"])
|
|
self.assertTrue(self.game.state["ended"])
|
|
|
|
def test_season_boundary_resets_scores_keeps_accounts_chat_and_archives_history(self):
|
|
self.season_fixture()
|
|
with self.game.database() as connection:
|
|
season = self.game.current_season(connection)
|
|
boundary = season["ends_at"]
|
|
self.game.state = self.sample_state()
|
|
self.game.state.update(started_at=boundary - 10000, deadline=boundary + 5000)
|
|
with patch.object(self.game, "now_ms", return_value=boundary - 100):
|
|
self.game.save_state(self.game.state)
|
|
players = copy.deepcopy(self.game.state["players"])
|
|
original_presses = copy.deepcopy(self.game.state["presses"])
|
|
with self.game.database() as connection:
|
|
connection.execute("INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
|
|
("alice", "Alice", boundary - 100, "See you next season"))
|
|
with patch.object(self.game, "now_ms", return_value=boundary - 1):
|
|
self.game.check_ended()
|
|
self.assertEqual(self.game.state["deadline"], boundary)
|
|
with patch.object(self.game, "now_ms", return_value=boundary):
|
|
self.game.check_ended()
|
|
self.game.check_ended()
|
|
self.assertFalse(self.game.state["ended"])
|
|
self.assertEqual(self.game.state["started_at"], boundary)
|
|
self.assertEqual(self.game.state["deadline"], boundary + self.game.DURATION_MS)
|
|
self.assertEqual(self.game.state["players"], players)
|
|
self.assertEqual(self.game.state["presses"], [])
|
|
self.assertEqual(self.game.player_summary("alice")["total_points"], 0)
|
|
self.assertTrue(any(m["message"] == "See you next season" for m in self.game.chat_messages()))
|
|
archive = self.game.seasonal_leaderboard()["seasons"][0]
|
|
self.assertEqual(archive["id"], season["id"])
|
|
self.assertEqual(archive["reason"], "boundary")
|
|
self.assertEqual(archive["duration_ms"], 10000)
|
|
with self.game.database() as connection:
|
|
row = connection.execute("SELECT presses FROM season_results WHERE id = ?", (season["id"],)).fetchone()
|
|
self.assertEqual(json.loads(row["presses"]), original_presses)
|
|
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
|
|
|
def test_season_tied_winners_include_zero_score_participants_not_idle_accounts(self):
|
|
self.season_fixture()
|
|
self.game.state = self.sample_state()
|
|
self.game.state["players"]["idle"] = dict(self.game.state["players"]["bob"], name="Idle", token_hashes=[])
|
|
for press in self.game.state["presses"]:
|
|
press["points"] = 0
|
|
with self.game.database() as connection:
|
|
season = self.game.current_season(connection)
|
|
self.game.archive_season(connection, season, self.game.state, self.game.state["deadline"], "countdown")
|
|
winners = self.game.seasonal_leaderboard()["seasons"][0]["winners"]
|
|
self.assertEqual({w["name"] for w in winners}, {self.game.state["players"]["alice"]["name"], "Bøb"})
|
|
self.assertTrue(all(w["total_points"] == 0 for w in winners))
|
|
|
|
def test_seasons_catch_up_downtime_without_fabricating_winners(self):
|
|
self.season_fixture()
|
|
at = int(datetime(2027, 7, 1, tzinfo=timezone.utc).timestamp() * 1000)
|
|
with patch.object(self.game, "now_ms", return_value=at):
|
|
self.game.check_ended()
|
|
self.game.state = self.game.load_state()
|
|
self.game.check_ended()
|
|
result = self.game.seasonal_leaderboard()
|
|
self.assertEqual(result["current"]["id"], "2027-06-21")
|
|
self.assertTrue(result["current"]["ended"])
|
|
self.assertEqual(len(result["seasons"]), 4)
|
|
self.assertTrue(all(row["winners"] == [] for row in result["seasons"]))
|
|
|
|
def test_season_rollover_failure_does_not_change_memory_or_database(self):
|
|
self.season_fixture()
|
|
before = copy.deepcopy(self.game.state)
|
|
with self.game.database() as connection:
|
|
boundary = self.game.current_season(connection)["ends_at"]
|
|
connection.execute("""CREATE TRIGGER reject_archive BEFORE INSERT ON season_results
|
|
BEGIN SELECT RAISE(ABORT, 'test rollback'); END""")
|
|
with patch.object(self.game, "now_ms", return_value=boundary):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.game.check_ended()
|
|
self.assertEqual(self.game.state, before)
|
|
with self.game.database() as connection:
|
|
self.assertEqual(self.game.current_season(connection)["ends_at"], boundary)
|
|
self.assertEqual(connection.execute("SELECT COUNT(*) FROM season_results").fetchone()[0], 0)
|
|
connection.execute("DROP TRIGGER reject_archive")
|
|
self.assertEqual(self.game.load_state(), before)
|
|
|
|
def test_season_upgrade_adopts_existing_game_without_losing_data(self):
|
|
source = self.sample_state()
|
|
self.write_legacy(source)
|
|
self.assertEqual(self.game.load_state(), source)
|
|
with self.game.database() as connection:
|
|
original = self.game.current_season(connection)
|
|
connection.execute("DROP TABLE season_clock")
|
|
connection.execute("DROP TABLE season_results")
|
|
connection.execute("DROP TABLE season_milestone_messages")
|
|
self.assertEqual(self.game.load_state(), source)
|
|
with self.game.database() as connection:
|
|
self.assertEqual(self.game.current_season(connection)["id"], original["id"])
|
|
self.assertEqual(self.game.seasonal_leaderboard()["seasons"], [])
|
|
|
|
def test_seasons_api_is_public_and_contains_no_private_credentials(self):
|
|
request = self.presence_client()
|
|
status, result = request(path="/api/seasons")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(set(result), {"current", "seasons"})
|
|
for player in result["current"]["leaderboard"]:
|
|
self.assertEqual(set(player), {"name", "total_points", "press_count"})
|
|
self.assertNotIn("password_hash", json.dumps(result))
|
|
self.assertNotIn("token_hashes", json.dumps(result))
|
|
|
|
def test_press_at_season_boundary_scores_only_in_new_season(self):
|
|
request = self.presence_client()
|
|
with self.game.database() as connection:
|
|
boundary = self.game.current_season(connection)["ends_at"]
|
|
self.game.state["deadline"] = boundary + 1000
|
|
before = copy.deepcopy(self.game.state["presses"])
|
|
with patch.object(self.game, "now_ms", return_value=boundary):
|
|
status, result = request("POST", {}, "alice-token", "/api/press")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(result["press"]["points"], 0)
|
|
self.assertEqual(len(self.game.state["presses"]), 1)
|
|
self.assertEqual(self.game.state["presses"][0]["at"], boundary)
|
|
self.assertEqual(self.game.state["players"]["alice"]["token_hashes"][0], self.game.hash_token("alice-token"))
|
|
with self.game.database() as connection:
|
|
archive = connection.execute("SELECT presses FROM season_results ORDER BY starts_at LIMIT 1").fetchone()
|
|
self.assertEqual(json.loads(archive["presses"]), before)
|
|
|
|
def test_finished_season_rejects_presses_and_keeps_archive_immutable(self):
|
|
request = self.presence_client()
|
|
deadline = self.game.state["deadline"]
|
|
with patch.object(self.game, "now_ms", return_value=deadline):
|
|
self.assertEqual(request("POST", {}, "alice-token", "/api/press")[0], 409)
|
|
first = request(path="/api/seasons")[1]
|
|
self.assertEqual(request("POST", {}, "alice-token", "/api/press")[0], 409)
|
|
self.assertEqual(request(path="/api/seasons")[1], first)
|
|
self.assertEqual(first["seasons"][0]["winners"][0]["total_points"], 40)
|
|
self.assertEqual(len(self.game.state["presses"]), 3)
|
|
|
|
def test_season_reset_preserves_prior_milestone_announcements(self):
|
|
self.season_fixture()
|
|
with self.game.database() as connection:
|
|
season = self.game.current_season(connection)
|
|
connection.execute("UPDATE community_milestones SET reached_at = ?, announced_at = ?, message = ? WHERE id = 'ten_players'",
|
|
(self.game.state["started_at"], self.game.state["started_at"], "Historic milestone"))
|
|
self.game.state["deadline"] = season["ends_at"] + 100
|
|
with patch.object(self.game, "now_ms", return_value=season["ends_at"]):
|
|
self.game.check_ended()
|
|
messages = [message for message in self.game.chat_messages() if message["message"] == "Historic milestone"]
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertIn(season["id"], messages[0]["id"])
|
|
with self.game.database() as connection:
|
|
self.assertIsNone(connection.execute("SELECT reached_at FROM community_milestones WHERE id = 'ten_players'").fetchone()[0])
|
|
|
|
def test_season_watcher_keeps_running_after_countdown_ends(self):
|
|
at = self.season_fixture()
|
|
self.game.state.update(ended=True, ended_at=at, deadline=at)
|
|
with patch.object(self.game, "now_ms", return_value=at):
|
|
with patch.object(self.game.time, "sleep", side_effect=RuntimeError("stop watcher")) as sleep:
|
|
with self.assertRaisesRegex(RuntimeError, "stop watcher"):
|
|
self.game.end_watcher()
|
|
sleep.assert_called_once()
|
|
self.assertGreater(sleep.call_args.args[0], 0)
|
|
self.assertLessEqual(sleep.call_args.args[0], 60)
|
|
|
|
def test_season_roman_numerals(self):
|
|
for number, expected in ((1, "I"), (2, "II"), (3, "III"), (4, "IV"), (9, "IX"),
|
|
(14, "XIV"), (40, "XL"), (49, "XLIX"), (99, "XCIX"), (400, "CD")):
|
|
self.assertEqual(self.game.roman_numeral(number), expected)
|
|
with self.assertRaises(ValueError):
|
|
self.game.roman_numeral(0)
|
|
|
|
def test_season_numbering_stable_on_rollover_and_restart(self):
|
|
self.season_fixture()
|
|
with self.game.database() as connection:
|
|
first = self.game.current_season(connection)
|
|
self.assertEqual((first["number"], first["name"]), (1, "Season I"))
|
|
with patch.object(self.game, "now_ms", return_value=first["ends_at"]):
|
|
self.game.check_ended()
|
|
self.game.state = self.game.load_state()
|
|
result = self.game.seasonal_leaderboard()
|
|
self.assertEqual((result["current"]["number"], result["current"]["name"]), (2, "Season II"))
|
|
self.assertEqual(result["current"]["starts_at"], first["ends_at"])
|
|
self.assertEqual((result["seasons"][0]["number"], result["seasons"][0]["name"]), (1, "Season I"))
|
|
with self.game.database() as connection:
|
|
self.assertEqual(self.game.current_season(connection)["name"], "Season II")
|
|
|
|
def test_season_numbering_upgrade_starts_with_earliest_archive(self):
|
|
self.season_fixture()
|
|
with self.game.database() as connection:
|
|
earlier = self.game.season_for_time(self.game.current_season(connection)["starts_at"] - 1)
|
|
self.game.archive_season(connection, earlier, self.game.state, self.game.state["deadline"], "countdown")
|
|
connection.execute("DROP TABLE season_numbering")
|
|
self.game.state = self.game.load_state()
|
|
result = self.game.seasonal_leaderboard()
|
|
self.assertEqual(result["seasons"][0]["name"], "Season I")
|
|
self.assertEqual(result["current"]["name"], "Season II")
|
|
with self.game.database() as connection:
|
|
origin = connection.execute("SELECT origin_at FROM season_numbering").fetchone()[0]
|
|
self.assertEqual(origin, earlier["starts_at"])
|
|
self.game.load_state()
|
|
self.assertEqual(self.game.seasonal_leaderboard()["current"]["name"], "Season II")
|
|
|
|
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_eight_milestones_match_achievement_count_and_start_unreached(self):
|
|
self.game.state = self.game.load_state()
|
|
goals = self.game.milestone_status(self.game.state, self.game.now_ms())
|
|
self.assertEqual(len(goals), 12)
|
|
self.assertEqual(len(goals), len(self.game.player_achievements("nobody")))
|
|
self.assertTrue(all(goal["progress"] < goal["target"] for goal in goals))
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
|
|
def test_milestone_thresholds_for_counts_and_precision(self):
|
|
game = self.game.new_game()
|
|
press = {"player_id": "p", "at": game["started_at"], "remaining_ms": 10000, "zone": "green", "points": 0}
|
|
for count in (9, 10, 49, 50, 51):
|
|
game["players"] = {str(i): {} for i in range(count)}
|
|
goals = {g["id"]: g for g in self.game.milestone_status(game, game["started_at"])}
|
|
self.assertEqual(goals["ten_players"]["progress"] == goals["ten_players"]["target"], count >= 10)
|
|
self.assertEqual(goals["fifty_players"]["progress"] == goals["fifty_players"]["target"], count >= 50)
|
|
for count in (99, 100, 999, 1000):
|
|
game["presses"] = [press] * count
|
|
goals = {g["id"]: g for g in self.game.milestone_status(game, game["started_at"])}
|
|
self.assertEqual(goals["hundred_presses"]["progress"] == 100, count >= 100)
|
|
self.assertEqual(goals["thousand_presses"]["progress"] == 1000, count >= 1000)
|
|
for remaining, qualifies in ((10000, False), (9999, True), (0, False), (-1, False)):
|
|
game["presses"] = [dict(press, remaining_ms=remaining)]
|
|
self.assertEqual(next(g for g in self.game.milestone_status(game, game["started_at"]) if g["id"] == "first_photo_finish")["progress"], int(qualifies))
|
|
|
|
def test_survival_milestones_boundaries_and_game_end_cap(self):
|
|
game = self.game.new_game()
|
|
day = 86400000
|
|
start = game["started_at"]
|
|
game["deadline"] = start + 40 * day
|
|
for days in (1, 7, 30):
|
|
before = self.game.milestone_status(game, start + days * day - 1)
|
|
after = self.game.milestone_status(game, start + days * day)
|
|
index = (1, 7, 30).index(days)
|
|
self.assertLess(before[index]["progress"], before[index]["target"])
|
|
self.assertEqual(after[index]["progress"], after[index]["target"])
|
|
game.update(ended=True, ended_at=start + day // 2)
|
|
self.assertTrue(all(g["progress"] < g["target"] for g in self.game.milestone_status(game, start + 100 * day)[:3]))
|
|
game.update(ended=False, ended_at=None, deadline=start + day // 2)
|
|
self.assertTrue(all(g["progress"] < g["target"] for g in self.game.milestone_status(game, start + 100 * day)[:3]))
|
|
|
|
def test_historical_milestones_are_silent_on_upgrade(self):
|
|
game = self.sample_state()
|
|
game["players"].update({str(i): dict(game["players"]["bob"], token_hashes=[]) for i in range(50)})
|
|
game["presses"] *= 334
|
|
game["started_at"] = self.game.now_ms() - 31 * 86400000
|
|
self.write_legacy(game)
|
|
self.game.state = self.game.load_state()
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
with self.game.database() as connection:
|
|
self.assertEqual(connection.execute("SELECT COUNT(*) FROM community_milestones WHERE reached_at IS NOT NULL").fetchone()[0], 10)
|
|
self.game.load_state()
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
# Existing SQLite database without the new table follows the same baseline.
|
|
with self.game.database() as connection:
|
|
connection.execute("DROP TABLE community_milestones")
|
|
self.game.load_state()
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
|
|
def test_new_milestone_announces_once_survives_restart_and_has_no_player_fk(self):
|
|
self.game.state = self.game.load_state()
|
|
start = self.game.state["started_at"]
|
|
self.game.state["deadline"] = start + 2 * 86400000
|
|
with patch.object(self.game, "now_ms", return_value=start + 86400000):
|
|
self.game.save_state(self.game.state)
|
|
self.game.check_ended()
|
|
self.game.state = self.game.load_state()
|
|
messages = self.game.chat_messages()
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["id"], "milestone:day_one")
|
|
self.assertEqual(messages[0]["kind"], "milestone")
|
|
self.assertEqual(messages[0]["name"], "The Button")
|
|
self.assertEqual(self.game.state["players"], {})
|
|
with self.game.database() as connection:
|
|
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
|
|
|
def test_milestone_failure_rolls_back_game_and_unlock(self):
|
|
self.game.state = self.game.load_state()
|
|
before = copy.deepcopy(self.game.state)
|
|
changed = copy.deepcopy(before)
|
|
changed["started_at"] -= 86400000
|
|
with self.game.database() as connection:
|
|
connection.execute("""CREATE TRIGGER reject_milestone BEFORE UPDATE ON community_milestones
|
|
BEGIN SELECT RAISE(ABORT, 'test rollback'); END""")
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.game.save_state(changed)
|
|
with self.game.database() as connection:
|
|
connection.execute("DROP TRIGGER reject_milestone")
|
|
self.assertEqual(self.game.load_state(), before)
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
|
|
def test_milestones_share_chat_limit_and_do_not_affect_player_cooldown(self):
|
|
self.write_legacy(self.sample_state())
|
|
self.game.state = self.game.load_state()
|
|
at = self.game.now_ms()
|
|
with self.game.database() as connection:
|
|
connection.executemany(
|
|
"INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
|
|
[("alice", "Alice", at - 1000 + i, str(i)) for i in range(105)],
|
|
)
|
|
connection.execute(
|
|
"UPDATE community_milestones SET reached_at = ?, announced_at = ?, message = ? WHERE id = 'ten_players'",
|
|
(at, at, "Milestone: Ten players!"),
|
|
)
|
|
messages = self.game.chat_messages()
|
|
self.assertEqual(len(messages), 100)
|
|
self.assertEqual(messages[-1]["id"], "milestone:ten_players")
|
|
self.assertEqual(messages[0]["message"], "6")
|
|
with self.game.database() as connection:
|
|
latest_player = connection.execute(
|
|
"SELECT at FROM chat_messages WHERE player_id = ? AND kind = 'message' ORDER BY id DESC LIMIT 1", ("alice",),
|
|
).fetchone()
|
|
self.assertEqual(latest_player["at"], at - 896)
|
|
|
|
def test_archived_season_awards_and_shared_ranks(self):
|
|
self.game.state = self.game.load_state()
|
|
scores = {"a": 32, "b": 32, "c": 16, "d": 8}
|
|
for number in (1, 2):
|
|
presses = [{"player_id": pid, "points": points} for pid, points in scores.items()]
|
|
with self.game.database() as connection:
|
|
connection.execute("""INSERT INTO season_results (id,name,starts_at,ends_at,played_from,finished_at,reason,leaderboard,presses)
|
|
VALUES (?,?,?,?,?,?,?,?,?)""", (str(number), "Same name", number*1000, number*1000+100, number*1000, number*1000+50, "countdown", "[]", json.dumps(presses)))
|
|
badges = {a["id"]: a for a in self.game.seasonal_achievements("a")}
|
|
self.assertTrue(badges["season_champion"]["unlocked"])
|
|
self.assertEqual(badges["season_veteran"]["progress"], number)
|
|
self.assertEqual(badges["season_veteran"]["unlocked"], number == 2)
|
|
self.assertTrue(self.game.seasonal_achievements("b")[-1]["unlocked"])
|
|
self.assertTrue(self.game.seasonal_achievements("c")[-2]["unlocked"])
|
|
self.assertFalse(self.game.seasonal_achievements("d")[-2]["unlocked"])
|
|
self.assertFalse(any(a["unlocked"] for a in self.game.seasonal_achievements("idle")))
|
|
self.game.load_state()
|
|
self.assertEqual(self.game.seasonal_achievements("a")[0]["unlocked_at"], 1050)
|
|
|
|
def test_new_events_upgrade_silently_and_announce_future_threshold_once(self):
|
|
self.game.state = self.game.load_state()
|
|
at = self.game.now_ms()
|
|
presses = [{"player_id": str(i), "points": 100, "zone": self.game.ZONES[i % 6]["name"], "remaining_ms": 60000, "at": at} for i in range(10)]
|
|
self.game.state["presses"] = presses
|
|
goals = {g["id"]: g for g in self.game.milestone_status(self.game.state, at)}
|
|
self.assertEqual(goals["ten_contenders"]["progress"], 10)
|
|
self.assertEqual(goals["shared_spectrum"]["progress"], 6)
|
|
self.assertEqual(goals["thousand_points"]["progress"], 1000)
|
|
with self.game.database() as connection:
|
|
connection.executemany("DELETE FROM community_milestones WHERE id = ?", [(key,) for key in ("shared_spectrum", "ten_contenders", "thousand_points", "ten_thousand_points")])
|
|
self.game.sync_milestones(connection, self.game.state, at)
|
|
self.assertEqual(self.game.chat_messages(), [])
|
|
self.game.state["presses"] = presses * 10
|
|
with self.game.database() as connection:
|
|
self.game.sync_milestones(connection, self.game.state, at+1)
|
|
self.game.sync_milestones(connection, self.game.state, at+2)
|
|
messages = [m for m in self.game.chat_messages() if m["id"] == "milestone:ten_thousand_points"]
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["at"], at+1)
|
|
|
|
def achievement_game(self):
|
|
self.game.state = self.sample_state()
|
|
self.game.state.update(duration_ms=120000, started_at=1000, presses=[])
|
|
|
|
def achievement_map(self, pid="alice"):
|
|
return {a["id"]: a for a in self.game.player_achievements(pid)}
|
|
|
|
def test_achievements_start_locked_without_presses(self):
|
|
self.achievement_game()
|
|
achievements = self.achievement_map()
|
|
self.assertEqual(len(achievements), 12)
|
|
for achievement in achievements.values():
|
|
self.assertFalse(achievement["unlocked"])
|
|
self.assertIsNone(achievement["unlocked_at"])
|
|
self.assertEqual(achievement["progress"], 0)
|
|
self.assertEqual(achievements["full_spectrum"]["target"], 6)
|
|
self.assertEqual(achievements["keeping_us_alive"]["target"], 60000)
|
|
|
|
def test_close_call_strict_boundary_and_first_unlock(self):
|
|
self.achievement_game()
|
|
for remaining in (60000, 0, -1):
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 2000, "remaining_ms": remaining, "zone": "white", "points": 32})
|
|
self.game.state["presses"].append({"player_id": "bob", "at": 3000, "remaining_ms": 1, "zone": "white", "points": 32})
|
|
self.assertFalse(self.achievement_map()["close_call"]["unlocked"])
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 4000, "remaining_ms": 59999, "zone": "white", "points": 32})
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 5000, "remaining_ms": 1, "zone": "white", "points": 32})
|
|
close = self.achievement_map()["close_call"]
|
|
self.assertTrue(close["unlocked"])
|
|
self.assertEqual(close["unlocked_at"], 4000)
|
|
self.assertEqual(close["progress"], 1)
|
|
|
|
def test_full_spectrum_counts_distinct_personal_zones_only(self):
|
|
self.achievement_game()
|
|
for index, zone in enumerate(self.game.ZONES[:-1]):
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 2000 + index, "remaining_ms": 1000, "zone": zone["name"], "points": zone["points"]})
|
|
repeated = dict(self.game.state["presses"][0], at=3000)
|
|
self.game.state["presses"].extend([repeated] * 10)
|
|
self.game.state["presses"].append({"player_id": "bob", "at": 4000, "remaining_ms": 1000, "zone": "white", "points": 32})
|
|
spectrum = self.achievement_map()["full_spectrum"]
|
|
self.assertEqual(spectrum["progress"], 5)
|
|
self.assertFalse(spectrum["unlocked"])
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 5000, "remaining_ms": 1000, "zone": "white", "points": 32})
|
|
spectrum = self.achievement_map()["full_spectrum"]
|
|
self.assertTrue(spectrum["unlocked"])
|
|
self.assertEqual(spectrum["unlocked_at"], 5000)
|
|
self.assertEqual(spectrum["progress"], 6)
|
|
|
|
def test_quiet_achievement_uses_global_gaps_and_inclusive_boundary(self):
|
|
self.achievement_game()
|
|
self.game.state["presses"] = [
|
|
{"player_id": "bob", "at": 31000, "remaining_ms": 90000, "zone": "green", "points": 0},
|
|
{"player_id": "alice", "at": 61000, "remaining_ms": 90000, "zone": "green", "points": 0},
|
|
]
|
|
quiet = self.achievement_map()["keeping_us_alive"]
|
|
self.assertFalse(quiet["unlocked"])
|
|
self.assertEqual(quiet["progress"], 30000)
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 120999, "remaining_ms": 60001, "zone": "green", "points": 0})
|
|
self.assertFalse(self.achievement_map()["keeping_us_alive"]["unlocked"])
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 180999, "remaining_ms": 60000, "zone": "blue", "points": 2})
|
|
quiet = self.achievement_map()["keeping_us_alive"]
|
|
self.assertTrue(quiet["unlocked"])
|
|
self.assertEqual(quiet["unlocked_at"], 180999)
|
|
self.assertEqual(quiet["progress"], quiet["target"])
|
|
|
|
def test_quiet_achievement_counts_first_press_from_game_start(self):
|
|
self.achievement_game()
|
|
self.game.state["presses"] = [{"player_id": "alice", "at": 61000, "remaining_ms": 60000, "zone": "blue", "points": 2}]
|
|
self.assertEqual(self.achievement_map()["keeping_us_alive"]["unlocked_at"], 61000)
|
|
self.game.state["duration_ms"] = 86400000
|
|
self.assertEqual(self.achievement_map()["keeping_us_alive"]["target"], 43200000)
|
|
self.assertFalse(self.achievement_map()["keeping_us_alive"]["unlocked"])
|
|
|
|
def test_achievements_credit_legacy_history_and_survive_restart(self):
|
|
self.achievement_game()
|
|
for index, zone in enumerate(self.game.ZONES):
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 61000 + index * 1000, "remaining_ms": 999, "zone": zone["name"], "points": zone["points"]})
|
|
expected = self.game.player_achievements("alice")
|
|
self.assertTrue(all(a["unlocked"] for a in expected if a.get("scope") != "lifetime"))
|
|
self.write_legacy(self.game.state)
|
|
self.game.state = self.game.load_state()
|
|
self.assertEqual(self.game.player_achievements("alice"), expected)
|
|
self.game.save_state(self.game.state)
|
|
self.game.state = self.game.load_state()
|
|
self.assertEqual(self.game.player_achievements("alice"), expected)
|
|
self.assertFalse(any(a["unlocked"] for a in self.game.player_achievements("bob")))
|
|
|
|
def test_achievements_are_private_and_do_not_change_scores(self):
|
|
request = self.presence_client()
|
|
self.assertEqual(request(path="/api/me")[0], 401)
|
|
before = copy.deepcopy(self.game.state)
|
|
status, summary = request(token="alice-token", path="/api/me")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(len(summary["achievements"]), 12)
|
|
self.assertEqual(summary["total_points"], 40)
|
|
self.assertEqual(self.game.state, before)
|
|
self.assertNotIn("achievements", self.game.load_state())
|
|
for row in self.game.public_leaderboard() + self.game.global_stats()["leaderboard"]:
|
|
self.assertNotIn("achievements", row)
|
|
|
|
def test_precision_badges_strict_boundaries_and_first_unlock(self):
|
|
for achievement_id, threshold in (("photo_finish", 10000), ("last_second", 1000)):
|
|
with self.subTest(achievement=achievement_id):
|
|
self.achievement_game()
|
|
for at, remaining in enumerate((threshold, threshold + 1, 0, -1), start=2000):
|
|
self.game.state["presses"].append({"player_id": "alice", "at": at, "remaining_ms": remaining, "zone": "green", "points": 0})
|
|
self.game.state["presses"].append({"player_id": "bob", "at": 3000, "remaining_ms": 1, "zone": "white", "points": 32})
|
|
self.assertFalse(self.achievement_map()[achievement_id]["unlocked"])
|
|
for at in (4000, 5000):
|
|
self.game.state["presses"].append({"player_id": "alice", "at": at, "remaining_ms": threshold - 1, "zone": "white", "points": 32})
|
|
badge = self.achievement_map()[achievement_id]
|
|
self.assertTrue(badge["unlocked"])
|
|
self.assertEqual(badge["unlocked_at"], 4000)
|
|
self.assertEqual((badge["progress"], badge["target"]), (1, 1))
|
|
|
|
def test_zone_badges_require_own_press_in_exact_zone(self):
|
|
for achievement_id, zone in (("red_alert", "red"), ("into_the_dark", "black"), ("white_knuckle", "white")):
|
|
with self.subTest(achievement=achievement_id):
|
|
self.achievement_game()
|
|
self.game.state["presses"] = [
|
|
{"player_id": "bob", "at": 2000, "remaining_ms": 1, "zone": zone, "points": 32},
|
|
{"player_id": "alice", "at": 3000, "remaining_ms": 1, "zone": "green", "points": 0},
|
|
]
|
|
self.assertFalse(self.achievement_map()[achievement_id]["unlocked"])
|
|
self.game.state["presses"].append({"player_id": "alice", "at": 4000, "remaining_ms": 10000, "zone": zone, "points": 8})
|
|
self.assertEqual(self.achievement_map()[achievement_id]["unlocked_at"], 4000)
|
|
|
|
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.load_state()
|
|
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
|
|
self.game.state["ended_at"] = self.game.state["deadline"] if ended else None
|
|
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.write_legacy(self.sample_state())
|
|
self.game.state = self.game.load_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.write_legacy(self.sample_state())
|
|
self.game.state = self.game.load_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.write_legacy(self.sample_state())
|
|
self.game.state = self.game.load_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), 2)
|
|
self.assertEqual(announcements[1]["kind"], "milestone")
|
|
self.assertIn("Our first photo finish", announcements[1]["message"])
|
|
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()
|