Gaming/the-button
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Seasons

Browse Source
This commit is contained in:
eric
2026-10-21 14:02:50 -04:00
parent 69d49073f0
commit 613647cb50
6 changed files with 580 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_server.py
+239
View File
@@ -11,6 +11,7 @@ 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
@@ -66,6 +67,243 @@ class PersistenceTests(unittest.TestCase):
def write_legacy(self, state):
self.game.DATA_FILE.write_text(json.dumps(state), encoding="utf-8")
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()
@@ -592,6 +830,7 @@ class PersistenceTests(unittest.TestCase):
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")