daniel156161
f14d780f29
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.
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
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()
|