feat: add live replays and PostgreSQL maintenance tools
Build and Push Docker Container / build-and-push (push) Successful in 7m53s
Build and Push Docker Container / build-and-push (push) Successful in 7m53s
- Stream compact live replay updates across local and clustered dashboards. - Render responsive snake bodies as SVG paths with aligned custom icons. - Add cache-busted assets, replay fallback routes, and live-follow playback. - Support PostgreSQL benchmark sampling and idempotent SQLite migration. - Add dry-run cleanup for old low-quality PostgreSQL replay payloads. - Reward safe perimeter lanes and bump Prism to version 1.5.0. - Add backend, migration, dashboard, and perimeter regression coverage.
This commit is contained in:
@@ -0,0 +1,59 @@
|
||||
import unittest
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from server.services.dashboard_query import DashboardQueryService
|
||||
|
||||
class _Logger:
|
||||
def warning(self, message):
|
||||
return message
|
||||
|
||||
class TestDashboardQueryService(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self):
|
||||
self.database = AsyncMock()
|
||||
self.hub = AsyncMock()
|
||||
self.service = DashboardQueryService(
|
||||
gameplay_database=self.database,
|
||||
ws_hub=self.hub,
|
||||
logger=_Logger(),
|
||||
dashboard_running_game_stale_sec=600,
|
||||
)
|
||||
|
||||
async def test_pushes_compact_latest_turn_event(self):
|
||||
self.database.get_game_replay.return_value = {
|
||||
"game": {"game_id": "game-1", "status": "running"},
|
||||
"turns": [{"turn": 1}, {"turn": 2, "my_move": "up"}],
|
||||
}
|
||||
|
||||
await self.service.push_dashboard_game_replay_update("game-1")
|
||||
|
||||
payload = self.hub.broadcast_payload.await_args.args[0]
|
||||
self.assertEqual(payload["type"], "dashboard_game_replay_update")
|
||||
self.assertEqual(payload["game_id"], "game-1")
|
||||
self.assertEqual(payload["turn"], {"turn": 2, "my_move": "up"})
|
||||
self.assertNotIn("turns", payload)
|
||||
|
||||
async def test_publishes_game_id_for_cluster_live_updates(self):
|
||||
self.database.get_game_replay.return_value = {
|
||||
"game": {"game_id": "game-1", "status": "running"},
|
||||
"turns": [{"turn": 3}],
|
||||
}
|
||||
publish = AsyncMock()
|
||||
self.service.set_publish_notice(publish)
|
||||
|
||||
await self.service.push_dashboard_game_replay_update("game-1")
|
||||
|
||||
publish.assert_awaited_once_with("game_turn", "game-1")
|
||||
|
||||
async def test_remote_turn_notice_loads_and_broadcasts_turn(self):
|
||||
self.database.get_game_replay.return_value = {
|
||||
"game": {"game_id": "game-1", "status": "running"},
|
||||
"turns": [{"turn": 4}],
|
||||
}
|
||||
|
||||
await self.service.on_dashboard_games_update_notice("game_turn", "game-1")
|
||||
|
||||
payload = self.hub.broadcast_payload.await_args.args[0]
|
||||
self.assertEqual(payload["turn"]["turn"], 4)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user