Chat

test_server.py

@@ -202,6 +202,69 @@ class PersistenceTests(unittest.TestCase):
         self.game.state = self.game.new_game()
         self.assertEqual(get("/api/leaderboard")[1], {"leaderboard": []})
 
+    def test_chat_auth_validation_cooldown_and_persistence(self):
+        self.game.load_state()
+        self.game.state = self.sample_state()
+        self.game.save_state(self.game.state)
+        self.game.token_index = {self.game.hash_token("test-chat-token"): "alice"}
+        httpd = ThreadingHTTPServer(("127.0.0.1", 0), self.game.Handler)
+        thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(httpd.server_close)
+        self.addCleanup(thread.join)
+        self.addCleanup(httpd.shutdown)
+
+        def request(method, data=None, authenticated=False):
+            connection = HTTPConnection(*httpd.server_address, timeout=5)
+            try:
+                headers = {"Content-Type": "application/json"}
+                if authenticated:
+                    headers["Authorization"] = "Bearer test-chat-token"
+                connection.request(method, "/api/chat", json.dumps(data) if data is not None else None, headers)
+                response = connection.getresponse()
+                return response.status, json.loads(response.read())
+            finally:
+                connection.close()
+
+        self.assertEqual(request("GET"), (200, {"messages": []}))
+        self.assertEqual(request("POST", {"message": "Hello"})[0], 401)
+        for message in (None, 42, "", "   ", "x" * 201, "hi\nthere", "bad\x00text"):
+            self.assertEqual(request("POST", {"message": message}, True)[0], 400)
+        with patch.object(self.game, "now_ms", return_value=10000):
+            status, posted = request("POST", {"message": "  <b>Hello</b> '); --  ", "name": "Imposter"}, True)
+            self.assertEqual(status, 201)
+            self.assertEqual(posted["name"], self.game.state["players"]["alice"]["name"])
+            self.assertEqual(posted["message"], "<b>Hello</b> '); --")
+            self.assertEqual(set(posted), {"id", "name", "at", "message"})
+            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_chat_returns_latest_hundred_in_chronological_order(self):
+        self.game.load_state()
+        self.game.state = self.sample_state()
+        self.game.save_state(self.game.state)
+        with self.game.database() as connection:
+            connection.executemany(
+                "INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
+                [("alice", "Alice", index, str(index)) for index in range(105)],
+            )
+        messages = self.game.chat_messages()
+        self.assertEqual(len(messages), 100)
+        self.assertEqual([message["message"] for message in messages], [str(i) for i in range(5, 105)])
+        self.game.load_state()
+        self.assertEqual(messages, self.game.chat_messages())
+
     def test_http_registration_login_press_logout_and_restart(self):
         self.game.state = self.game.load_state()
         self.game.token_index = {}