no reason to ever enable twice

This commit is contained in:
J. Nick Koston
2024-06-12 16:31:06 -05:00
parent ed6cac4e4f
commit 4c6c45d8e0
2 changed files with 12 additions and 1 deletions

View File

@@ -160,7 +160,7 @@ _BLOCKED_CALLS = BlockedCalls(set())
def enable() -> None: def enable() -> None:
"""Enable the detection of blocking calls in the event loop.""" """Enable the detection of blocking calls in the event loop."""
if _BLOCKED_CALLS.calls: if _BLOCKED_CALLS.calls:
return raise RuntimeError("Blocking call detection is already enabled")
loop_thread_id = threading.get_ident() loop_thread_id = threading.get_ident()
for blocking_call in BLOCKING_CALLS: for blocking_call in BLOCKING_CALLS:

View File

@@ -229,6 +229,17 @@ async def test_protect_open(caplog: pytest.LogCaptureFixture) -> None:
assert "Detected blocking call to open with args" in caplog.text assert "Detected blocking call to open with args" in caplog.text
async def test_enable_multiple_times(caplog: pytest.LogCaptureFixture) -> None:
"""Test trying to enable multiple times."""
with patch.object(block_async_io, "_IN_TESTS", False):
block_async_io.enable()
with pytest.raises(
RuntimeError, match="Blocking call detection is already enabled"
):
block_async_io.enable()
@pytest.mark.parametrize( @pytest.mark.parametrize(
"path", "path",
[ [