/
/
/
1"""Tests for the blocking-IO-in-async check."""
2
3import ast
4
5from scripts.check_blocking_io import _BlockingCallVisitor, main
6
7
8def _blocking_calls(source: str) -> list[str]:
9 """Return the dotted blocking calls the visitor finds inside async scopes in a snippet."""
10 return [call for _, call in _BlockingCallVisitor().collect(ast.parse(source))]
11
12
13def test_flags_blocking_call_in_async_function() -> None:
14 """A module-qualified blocking call directly inside ``async def`` is flagged."""
15 source = "async def f():\n return os.walk(path)\n"
16 assert _blocking_calls(source) == ["os.walk"]
17
18
19def test_flags_requests_and_urllib() -> None:
20 """``requests`` verbs and ``urllib.request.urlopen`` are recognized inside async code."""
21 source = "async def f():\n requests.get(url)\n urllib.request.urlopen(url)\n"
22 assert _blocking_calls(source) == ["requests.get", "urllib.request.urlopen"]
23
24
25def test_ignores_blocking_call_in_sync_function() -> None:
26 """The same call in a plain ``def`` (e.g. a to_thread target) is not flagged."""
27 source = "def f():\n return os.walk(path)\n"
28 assert _blocking_calls(source) == []
29
30
31def test_ignores_sync_function_nested_in_async() -> None:
32 """A blocking call inside a sync helper nested in an async function is not flagged."""
33 source = (
34 "async def outer():\n"
35 " def _worker():\n"
36 " return shutil.rmtree(path)\n"
37 " return await asyncio.to_thread(_worker)\n"
38 )
39 assert _blocking_calls(source) == []
40
41
42def test_ignores_function_reference_passed_to_to_thread() -> None:
43 """Passing a blocking function as a reference (not calling it) is not flagged."""
44 source = "async def f():\n return await asyncio.to_thread(os.listdir, path)\n"
45 assert _blocking_calls(source) == []
46
47
48def test_real_tree_matches_baseline() -> None:
49 """The shipped tree has no blocking calls beyond the grandfathered baseline."""
50 assert main([]) == 0
51