"""UIActionScheduler 单元测试。 覆盖: - 向后兼容 enqueue - 优先级排序(CRITICAL 优先于 NORMAL) - drain(空队列 / 有 pending / 超时) - 队列满抛 RATE_LIMITED - 等待超时抛 TIMEOUT - metrics 准确性 - fast-fail 机制(mark_wechat_dead 后非 CRITICAL 抛 WECHAT_NOT_READY) - CRITICAL 任务在 fast-fail 窗口内正常执行 - CRITICAL 任务自动 delay_ms=0 """ from __future__ import annotations import asyncio import os import sys import time import pytest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from woc_bridge.messaging.ui_action_scheduler import UIActionScheduler, Priority from woc_bridge.models import BridgeError pytestmark = pytest.mark.asyncio async def test_backward_compat_enqueue(): """向后兼容 enqueue:以 NORMAL 优先级入队并返回结果。""" scheduler = UIActionScheduler(send_delay_ms=10, max_calls_per_sec=100) await scheduler.start() try: result = await scheduler.enqueue(lambda: asyncio.sleep(0.01, result="ok")) assert result == "ok" metrics = scheduler.get_metrics() assert metrics["executed_total"] == 1 assert metrics["executed_by_priority"]["NORMAL"] == 1 finally: await scheduler.stop() async def test_priority_ordering(): """CRITICAL 任务应先于后入队的 NORMAL 任务开始执行。""" scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) await scheduler.start() try: start_times: list[tuple[str, str]] = [] async def _track(result: str, delay: float = 0.01) -> str: start_times.append(("start", result)) await asyncio.sleep(delay) return result # 第一个 NORMAL 立即开始执行 normal_task = asyncio.create_task( scheduler.enqueue(lambda: _track("normal", 0.1)) ) await asyncio.sleep(0.02) # 等第一个开始 critical_task = asyncio.create_task( scheduler.execute( lambda: _track("critical", 0.01), priority=Priority.CRITICAL ) ) normal2_task = asyncio.create_task( scheduler.execute( lambda: _track("normal2", 0.01), priority=Priority.NORMAL ) ) critical_result = await critical_task normal2_result = await normal2_task normal_result = await normal_task assert critical_result == "critical" assert normal2_result == "normal2" assert normal_result == "normal" # CRITICAL 应在 normal2 之前开始 critical_idx = next(i for i, (_, r) in enumerate(start_times) if r == "critical") normal2_idx = next(i for i, (_, r) in enumerate(start_times) if r == "normal2") assert critical_idx < normal2_idx, ( f"CRITICAL 应在 NORMAL2 之前执行,实际顺序: {start_times}" ) finally: await scheduler.stop() async def test_drain_empty_queue(): """空队列 drain 应立即返回 True。""" scheduler = UIActionScheduler() result = await scheduler.drain(timeout=1.0) assert result is True async def test_drain_with_pending(): """有 pending 任务的队列 drain 应等待任务完成。""" scheduler = UIActionScheduler(send_delay_ms=10) await scheduler.start() try: await scheduler.enqueue(lambda: asyncio.sleep(0.05, result="ok")) result = await scheduler.drain(timeout=2.0) assert result is True await asyncio.sleep(0.01) # 让 worker 完全进入 idle finally: await scheduler.stop() async def test_drain_timeout(): """长任务未完成时 drain 应超时返回 False。""" scheduler = UIActionScheduler(send_delay_ms=0) await scheduler.start() try: # 入队长任务但不 await(避免阻塞测试) asyncio.create_task( scheduler.enqueue(lambda: asyncio.sleep(1.0, result="slow")) ) await asyncio.sleep(0.05) # 让任务开始执行 result = await scheduler.drain(timeout=0.1) assert result is False finally: await scheduler.stop() async def test_queue_full_rate_limited(): """队列满时入队应抛 RATE_LIMITED。""" scheduler = UIActionScheduler(send_delay_ms=1000, max_queue_size=2) await scheduler.start() try: # 第一个任务开始执行(worker 取走后队列腾出空位) asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.5))) await asyncio.sleep(0.05) # 让 worker 取走第一个任务并开始执行 # 再入队 2 个排队任务(队列满:2 个 pending) asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.01))) asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.01))) await asyncio.sleep(0.02) # 确保两个任务都入队 # 第 3 个排队任务应被拒绝(队列已满) with pytest.raises(BridgeError) as exc_info: await scheduler.enqueue(lambda: asyncio.sleep(0.01)) assert exc_info.value.code == "RATE_LIMITED" finally: await scheduler.stop() async def test_wait_timeout(): """等待结果超时应抛 TIMEOUT。""" scheduler = UIActionScheduler(send_delay_ms=1000, max_calls_per_sec=100) await scheduler.start() try: # 第一个任务慢(执行 + 默认延时会让第二个等待很久) asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.5))) await asyncio.sleep(0.02) # 让第一个开始 with pytest.raises(BridgeError) as exc_info: await scheduler.enqueue( lambda: asyncio.sleep(0.01), wait_timeout_ms=100 ) assert exc_info.value.code == "TIMEOUT" finally: await scheduler.stop() async def test_metrics_accuracy(): """metrics 应准确统计各优先级执行次数。""" scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) await scheduler.start() try: async def _ret(v): return v await scheduler.execute(lambda: _ret("a"), priority=Priority.CRITICAL) await scheduler.execute(lambda: _ret("b"), priority=Priority.HIGH) await scheduler.execute(lambda: _ret("c"), priority=Priority.NORMAL) await scheduler.execute(lambda: _ret("d"), priority=Priority.LOW) metrics = scheduler.get_metrics() assert metrics["executed_total"] == 4 assert metrics["executed_by_priority"]["CRITICAL"] == 1 assert metrics["executed_by_priority"]["HIGH"] == 1 assert metrics["executed_by_priority"]["NORMAL"] == 1 assert metrics["executed_by_priority"]["LOW"] == 1 assert metrics["pending_count"] == 0 finally: await scheduler.stop() async def test_fast_fail_wechat_dead(): """mark_wechat_dead 后非 CRITICAL 任务应 fast-fail 抛 WECHAT_NOT_READY。""" scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) await scheduler.start() try: scheduler.mark_wechat_dead(2.0) # 启动 2 秒 fast-fail 窗口 with pytest.raises(BridgeError) as exc_info: await scheduler.execute( lambda: asyncio.sleep(0.01, result="ok"), priority=Priority.NORMAL, ) assert exc_info.value.code == "WECHAT_NOT_READY" metrics = scheduler.get_metrics() assert metrics["fast_fail_total"] == 1 finally: await scheduler.stop() async def test_critical_bypasses_fast_fail(): """CRITICAL 任务在 fast-fail 窗口内应正常执行。""" scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) await scheduler.start() try: scheduler.mark_wechat_dead(2.0) # 启动 fast-fail 窗口 result = await scheduler.execute( lambda: asyncio.sleep(0.01, result="critical_ok"), priority=Priority.CRITICAL, ) assert result == "critical_ok" metrics = scheduler.get_metrics() assert metrics["executed_total"] == 1 assert metrics["executed_by_priority"]["CRITICAL"] == 1 assert metrics["fast_fail_total"] == 0 finally: await scheduler.stop() async def test_critical_auto_delay_zero(): """CRITICAL 任务应自动 delay_ms=0,不受默认 send_delay_ms 影响。""" scheduler = UIActionScheduler(send_delay_ms=5000, max_calls_per_sec=100) await scheduler.start() try: async def _ret(v): return v t0 = time.monotonic() await scheduler.execute(lambda: _ret("first"), priority=Priority.CRITICAL) await scheduler.execute(lambda: _ret("second"), priority=Priority.CRITICAL) elapsed = time.monotonic() - t0 # 若 delay_ms=5000 生效,第二个任务会等 5s。自动 delay=0 后应在 1s 内完成 assert elapsed < 1.5, f"CRITICAL 自动 delay=0 失效,耗时 {elapsed:.2f}s" finally: await scheduler.stop()