diff --git a/cloud/tests/test_coverage85_misc.py b/cloud/tests/test_coverage85_misc.py index 1d2c08c..3ed4078 100644 --- a/cloud/tests/test_coverage85_misc.py +++ b/cloud/tests/test_coverage85_misc.py @@ -9,7 +9,6 @@ playwright 对象),ensure_* 只写临时 codex_home,绝不启动 import json import os -import queue import sqlite3 import tempfile import time @@ -437,6 +436,28 @@ class AutoscalerProcTest(unittest.TestCase): # ---------------------------------------------------------- playwright +class _SimpleQueue: + """queue.Queue 的最小替身(只给 screencast sink 用)。 + + 刻意不用 stdlib queue:cloud/tests/test_queue_sandbox.py + 靠 cloud/queue.py 遮蔽 stdlib queue(它把 cloud/ 插进 + sys.path 再 from queue import);discover 按字母顺序先 + 导入本文件时若已缓存 stdlib queue,那个模块就 ImportError。 + 本步实测:import queue 一行导致 cloud 套件 47 项未跑。 + """ + + def __init__(self): + self._items = [] + + def put(self, item): + self._items.append(item) + + def get_nowait(self): + if not self._items: + raise RuntimeError("empty") + return self._items.pop(0) + + class _FakeMouse: def __init__(self): self.moves = [] @@ -695,7 +716,7 @@ class PlaywrightBrowserTest(unittest.TestCase): def test_screencast_sink_and_pump(self): b = self._browser() self.assertEqual(b.pump_screencast(), "") - sink = queue.Queue() + sink = _SimpleQueue() b.set_frame_sink(sink) b.navigate("https://example.com") # navigate 后 sink 已接上 CDP(_start_screencast 在