744 lines
41 KiB
Python
Executable File
744 lines
41 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_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), 8)
|
|
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):
|
|
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(self.game.milestone_status(game, game["started_at"])[-1]["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], 8)
|
|
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 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(set(achievements), {"close_call", "full_spectrum", "keeping_us_alive", "red_alert", "into_the_dark", "white_knuckle", "photo_finish", "last_second"})
|
|
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))
|
|
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"]), 8)
|
|
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
|
|
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()
|