Add idempotency guards to patch_audit_functions() and start_blocking_io_detection()

aiocop/core/audit_patcher.py

@@ -51,6 +51,8 @@
 
 patched_functions: list[str] = []
 
+_patch_audit_functions_configured = False
+
 
 def _get_target(path: str) -> tuple[Any | None, str | None]:
     """
@@ -81,6 +83,12 @@ def patch_audit_functions() -> None:
 
     Should be called early in application startup, before start_blocking_io_detection().
     """
+    global _patch_audit_functions_configured
+
+    if _patch_audit_functions_configured is True:
+        logger.warning("patch_audit_functions called more than once, ignoring")
+        return
+
     patched_functions.clear()
 
     for func_path in FUNCTIONS_TO_PATCH:
@@ -118,6 +126,8 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
 
     logger.info("Patched functions for audit: %s", patched_functions)
 
+    _patch_audit_functions_configured = True
+
 
 def get_patched_functions() -> list[str]:
     """Return a copy of the list of functions that have been patched."""

aiocop/core/blocking_io.py

@@ -15,6 +15,7 @@
 
 _main_thread_id: int | None = None
 _trace_depth: int = 20
+_start_blocking_io_detection_configured = False
 
 MAX_EVENTS_PER_TASK = 50
 
@@ -92,13 +93,19 @@ def start_blocking_io_detection(trace_depth: int = 20) -> None:
 
     Should be called after patch_audit_functions() and before detect_slow_tasks().
     """
-    global _main_thread_id, _trace_depth
+    global _main_thread_id, _trace_depth, _start_blocking_io_detection_configured
+
+    if _start_blocking_io_detection_configured is True:
+        logger.warning("start_blocking_io_detection called more than once, ignoring")
+        return
+
     _main_thread_id = threading.main_thread().ident
     _trace_depth = trace_depth
 
     try:
         sys.addaudithook(_audit_hook)
         logger.info("Blocking I/O detection hook registered via sys.audit")
+        _start_blocking_io_detection_configured = True
     except Exception as e:
         logger.error("Failed to register blocking I/O detection hook: %s", e)
 

tests/test_aiocop.py

@@ -842,6 +842,49 @@ def test_severity_constants_exported(self) -> None:
         assert aiocop.THRESHOLD_LOW == 1
 
 
+# =============================================================================
+# Setup Idempotency Tests
+# =============================================================================
+
+
+class TestSetupIdempotency:
+    """Test that setup functions are safe to call more than once."""
+
+    def test_patch_audit_functions_twice_is_noop(self, setup_aiocop, caplog) -> None:
+        import os
+
+        # Checking that os.getcwd is not wrapped again by the second call (object identity).
+        before_func = os.getcwd
+
+        with caplog.at_level("WARNING"):
+            aiocop.patch_audit_functions()
+
+        assert os.getcwd is before_func
+        assert "patch_audit_functions called more than once" in caplog.text
+
+    def test_start_blocking_io_detection_twice_is_noop(self, setup_aiocop, caplog) -> None:
+        from aiocop.core import blocking_io
+
+        # Checking that trace_depth is not overwritten by the second call.
+        before_trace_depth = blocking_io._trace_depth
+
+        with caplog.at_level("WARNING"):
+            aiocop.start_blocking_io_detection(trace_depth=before_trace_depth + 999)
+
+        assert blocking_io._trace_depth == before_trace_depth
+        assert "start_blocking_io_detection called more than once" in caplog.text
+
+    def test_detect_slow_tasks_twice_is_noop(self, setup_aiocop, caplog) -> None:
+        # Checking that the threshold is not overwritten by the second call.
+        before_threshold = aiocop.get_slow_task_threshold_ms()
+
+        with caplog.at_level("WARNING"):
+            aiocop.detect_slow_tasks(threshold_ms=before_threshold + 999)
+
+        assert aiocop.get_slow_task_threshold_ms() == before_threshold
+        assert "detect_slow_tasks called more than once" in caplog.text
+
+
 # =============================================================================
 # Raise on Violations Integration Tests
 # =============================================================================

tests/test_uvloop.py

@@ -37,6 +37,12 @@ def reset_aiocop_state():
 
     This is critical for uvloop tests since we need fresh state
     for each test to properly test loop patching.
+
+    Note: _detect_slow_tasks_configured is reset since it gates a fresh event
+    loop object created every test. _patch_audit_functions_configured and
+    _start_blocking_io_detection_configured are NOT reset - they guard
+    permanent, process-wide state with no teardown, so resetting them would
+    re-wrap functions and re-register the audit hook each test.
     """
     # Reset before test
     state._monitoring_active = False