File size: 1,596 Bytes
d9d9e44
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
"""SPEC-4 §3: assert_sync_context 行为验证。

注: SPEC §4.2 原稿用 @pytest.mark.asyncio,但仓库未引入 pytest-asyncio。
为避免新增依赖,asyncio loop 内的测试改用 asyncio.run() 直接驱动协程,
行为等价,在 loop 内调用 assert_sync_context() 仍会触发 RuntimeError。
"""

from __future__ import annotations

import asyncio
import re
import threading

import pytest

from autoteam._playwright_guard import assert_sync_context


def test_passes_in_plain_thread():
    """普通线程无 loop, guard 必须放行"""
    holder: dict = {}

    def worker():
        try:
            assert_sync_context()
            holder["ok"] = True
        except Exception as e:  # noqa: BLE001
            holder["err"] = e

    t = threading.Thread(target=worker, name="test-plain")
    t.start()
    t.join(timeout=2.0)
    assert holder.get("ok") is True
    assert "err" not in holder


def test_blocks_in_asyncio_loop():
    """asyncio loop 内调用必须抛 RuntimeError"""
    async def runner():
        assert_sync_context()

    with pytest.raises(RuntimeError, match=r"\[PlaywrightGuard\].*asyncio loop"):
        asyncio.run(runner())


def test_message_contains_thread_and_loop_id():
    """异常消息应携带 thread 名 + loop id 便于诊断"""
    captured: dict = {}

    async def runner():
        try:
            assert_sync_context()
        except RuntimeError as e:
            captured["msg"] = str(e)

    asyncio.run(runner())
    msg = captured.get("msg", "")
    assert "thread=" in msg
    assert re.search(r"loop_id=0x[0-9a-f]+", msg)