243 lines
10 KiB
Python
243 lines
10 KiB
Python
import asyncio
|
|
import logging
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from app.danmu_queue import CommandHandler, QueueManager, QueueSystem
|
|
|
|
|
|
class IdleDefaultQueueStateTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.manager = QueueManager(self.temp_dir.name, logging.getLogger("idle-default-test"))
|
|
self.manager.state["has_user_finished_once"] = True
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
|
|
def _start_default_with_waiting_admin(self, uid=101):
|
|
self.manager.state.update({
|
|
"default_running": True,
|
|
"current_group": "薄荷",
|
|
"current_group_run_id": "default-run",
|
|
"group_start_time": datetime.now().isoformat(),
|
|
"billing_uid": None,
|
|
})
|
|
self.manager.join_queue(uid)
|
|
|
|
def test_default_completion_preserves_newly_joined_admin(self):
|
|
self._start_default_with_waiting_admin()
|
|
original_window = self.manager.state["admin_window_end"]
|
|
|
|
result = self.manager.group_finished("default-run")
|
|
|
|
self.assertTrue(result["accepted"])
|
|
self.assertTrue(result["was_default_running"])
|
|
self.assertTrue(result["need_default"])
|
|
self.assertEqual(self.manager.state["queue"], [101])
|
|
self.assertEqual(self.manager.state["current_admin_uid"], 101)
|
|
self.assertEqual(self.manager.state["admin_window_end"], original_window)
|
|
self.assertFalse(self.manager.state["default_running"])
|
|
self.assertIsNone(self.manager.state["current_group"])
|
|
|
|
def test_admin_window_can_expire_while_default_group_runs(self):
|
|
self._start_default_with_waiting_admin()
|
|
self.manager.state["admin_window_end"] = time.time() - 1
|
|
|
|
result = self.manager.check_admin_window_timeout()
|
|
|
|
self.assertTrue(result["timeout"])
|
|
self.assertEqual(result["kicked_uid"], 101)
|
|
self.assertEqual(self.manager.state["queue"], [])
|
|
self.assertTrue(self.manager.state["default_running"])
|
|
self.assertEqual(self.manager.state["current_group"], "薄荷")
|
|
|
|
def test_waiting_admin_leave_preserves_default_group(self):
|
|
self._start_default_with_waiting_admin()
|
|
|
|
result = self.manager.leave_queue(101)
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(self.manager.state["default_running"])
|
|
self.assertEqual(self.manager.state["current_group"], "薄荷")
|
|
self.assertEqual(self.manager.state["current_group_run_id"], "default-run")
|
|
|
|
|
|
class IdleDefaultFlowTests(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def _make_system(manager, runner):
|
|
system = QueueSystem.__new__(QueueSystem)
|
|
system.queue_mgr = manager
|
|
system.user_mgr = SimpleNamespace(
|
|
users={"101": {"uname": "tester"}},
|
|
get_points=lambda _uid: 10,
|
|
)
|
|
system.config = SimpleNamespace(default_group="薄荷")
|
|
system.runner = runner
|
|
system.log_monitor = SimpleNamespace(set_current_group=MagicMock())
|
|
system.login_monitor = SimpleNamespace(reset=MagicMock())
|
|
system.stats_store = None
|
|
system.logger = logging.getLogger("idle-default-race-test")
|
|
system.broadcast = AsyncMock()
|
|
system._default_start_lock = asyncio.Lock()
|
|
return system
|
|
|
|
@staticmethod
|
|
def _make_handler(system):
|
|
handler = CommandHandler.__new__(CommandHandler)
|
|
handler.config = SimpleNamespace(admin_uids=[])
|
|
handler.queue_mgr = system.queue_mgr
|
|
handler.user_mgr = system.user_mgr
|
|
handler.runner = system.runner
|
|
handler.log_monitor = system.log_monitor
|
|
handler.login_monitor = system.login_monitor
|
|
handler.logger = system.logger
|
|
handler.stats_store = None
|
|
handler.system = system
|
|
handler.broadcast = AsyncMock()
|
|
return handler
|
|
|
|
async def test_login_serializes_with_default_group_start(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
manager = QueueManager(temp_dir, logging.getLogger("idle-default-login-race"))
|
|
manager.state["has_user_finished_once"] = True
|
|
manager.join_queue(101)
|
|
manager.state.update({
|
|
"default_running": True,
|
|
"current_group": "薄荷",
|
|
"current_group_run_id": "default-run",
|
|
"group_start_time": datetime.now().isoformat(),
|
|
})
|
|
manager._save()
|
|
|
|
kill_started = asyncio.Event()
|
|
allow_kill = asyncio.Event()
|
|
|
|
async def blocked_kill(**_kwargs):
|
|
kill_started.set()
|
|
await allow_kill.wait()
|
|
return True
|
|
|
|
runner = SimpleNamespace(
|
|
kill_bgi=AsyncMock(side_effect=blocked_kill),
|
|
start_groups=AsyncMock(return_value=True),
|
|
)
|
|
system = self._make_system(manager, runner)
|
|
handler = self._make_handler(system)
|
|
|
|
login_task = asyncio.create_task(handler._cmd_login(101, "tester"))
|
|
await kill_started.wait()
|
|
default_task = asyncio.create_task(system._start_default_group())
|
|
await asyncio.sleep(0)
|
|
allow_kill.set()
|
|
await asyncio.gather(login_task, default_task)
|
|
|
|
self.assertEqual(manager.state["login_status"], "logining")
|
|
self.assertFalse(manager.state["default_running"])
|
|
self.assertIsNone(manager.state["current_group"])
|
|
runner.start_groups.assert_awaited_once_with(["扫码上号"])
|
|
|
|
async def test_login_watchdog_runs_even_if_default_state_leaks(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
manager = QueueManager(temp_dir, logging.getLogger("login-watchdog-default-leak"))
|
|
manager.join_queue(101)
|
|
manager.state.update({
|
|
"login_status": "logining",
|
|
"login_started_at": (datetime.now() - timedelta(seconds=300)).isoformat(),
|
|
"default_running": True,
|
|
"current_group": "薄荷",
|
|
})
|
|
manager._save()
|
|
system = self._make_system(manager, SimpleNamespace())
|
|
system._handle_login_timeout = AsyncMock()
|
|
|
|
with patch(
|
|
"app.danmu_queue.asyncio.sleep",
|
|
side_effect=[None, asyncio.CancelledError()],
|
|
):
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await system._check_login_watchdog_loop()
|
|
|
|
system._handle_login_timeout.assert_awaited_once_with(101, "tester")
|
|
|
|
async def test_user_group_completion_starts_default_while_next_admin_waits(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
manager = QueueManager(temp_dir, logging.getLogger("idle-default-flow-test"))
|
|
manager.state.update({
|
|
"queue": [1, 2],
|
|
"current_admin_uid": 1,
|
|
"current_group": "晶蝶",
|
|
"current_group_run_id": "user-run",
|
|
"group_start_time": (datetime.now() - timedelta(seconds=200)).isoformat(),
|
|
"login_status": "logged_in",
|
|
"billing_uid": 1,
|
|
"billing_started_at": datetime.now().isoformat(),
|
|
"billing_last_at": datetime.now().isoformat(),
|
|
"has_user_finished_once": True,
|
|
})
|
|
manager._save()
|
|
|
|
system = QueueSystem.__new__(QueueSystem)
|
|
system.queue_mgr = manager
|
|
system.user_mgr = SimpleNamespace(
|
|
users={"1": {"uname": "first"}, "2": {"uname": "second"}},
|
|
get_points=lambda _uid: 10,
|
|
)
|
|
system.log_monitor = SimpleNamespace(set_current_group=MagicMock())
|
|
system.logger = logging.getLogger("idle-default-flow-test")
|
|
system.broadcast = AsyncMock()
|
|
system._start_default_group = AsyncMock()
|
|
|
|
await system._on_group_finished("晶蝶", "user-run")
|
|
|
|
self.assertEqual(manager.state["queue"], [2])
|
|
self.assertEqual(manager.state["current_admin_uid"], 2)
|
|
self.assertIsNone(manager.state["login_status"])
|
|
system._start_default_group.assert_awaited_once_with()
|
|
|
|
async def test_default_group_can_start_with_waiting_admin(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
manager = QueueManager(temp_dir, logging.getLogger("idle-default-start-test"))
|
|
manager.state["has_user_finished_once"] = True
|
|
manager.join_queue(101)
|
|
original_window = manager.state["admin_window_end"]
|
|
|
|
system = QueueSystem.__new__(QueueSystem)
|
|
system.queue_mgr = manager
|
|
system.config = SimpleNamespace(default_group="薄荷")
|
|
system.user_mgr = SimpleNamespace(
|
|
users={"101": {"uname": "队首昵称"}},
|
|
_is_masked_uname=lambda name: (not name) or ("*" in name),
|
|
resolve_uname=AsyncMock(return_value=""),
|
|
)
|
|
system.runner = SimpleNamespace(
|
|
kill_bgi=AsyncMock(),
|
|
start_groups=AsyncMock(return_value=True),
|
|
)
|
|
system.log_monitor = SimpleNamespace(set_current_group=MagicMock())
|
|
system.stats_store = None
|
|
system.logger = logging.getLogger("idle-default-start-test")
|
|
system.broadcast = AsyncMock()
|
|
system._default_start_lock = asyncio.Lock()
|
|
|
|
await system._start_default_group()
|
|
|
|
self.assertTrue(manager.state["default_running"])
|
|
self.assertEqual(manager.state["current_group"], "薄荷")
|
|
self.assertEqual(manager.state["current_admin_uid"], 101)
|
|
self.assertEqual(manager.state["admin_window_end"], original_window)
|
|
self.assertIsNone(manager.state["billing_uid"])
|
|
system.runner.start_groups.assert_awaited_once_with(["薄荷"])
|
|
# 播报文案应使用昵称而非纯数字 UID
|
|
broadcast_texts = [str(call.args[0]) for call in system.broadcast.await_args_list]
|
|
self.assertTrue(any("队首昵称" in t for t in broadcast_texts))
|
|
self.assertFalse(any("队首101" in t for t in broadcast_texts))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|