[AAASM-5568] 🚨 (python-sdk): Clear 44 SonarCloud code smells

agent_assembly/adapters/openai_agents/patch.py

@@ -510,6 +510,87 @@ def patched_init(self: Any, *args: Any, **kwargs: Any) -> None:
     return None
 
 
+async def _run_async_tool_pre_execution_check(
+    callback_handler: Any,
+    *,
+    tool_name: str,
+    tool_input: Any,
+    agent_id: Any,
+    ctx: Any,
+    enforce: bool,
+) -> tuple[bool, Any, bool]:
+    """Run the pre-execution governance check for a tool invocation.
+
+    Returns ``(return_now, return_value, governance_failed)``. When
+    ``return_now`` is True the caller must immediately return ``return_value``
+    (a deny/blocked result); otherwise the tool call may proceed.
+    """
+    try:
+        decision = await _invoke_async_tool_check(
+            callback_handler,
+            tool_name=tool_name,
+            tool_input=tool_input,
+            agent_id=agent_id,
+            ctx=ctx,
+        )
+        status, reason = _normalize_decision(decision, enforce=enforce)
+        is_pending_flow = False
+        if status == "pending":
+            is_pending_flow = True
+            timeout_seconds = _get_pending_tool_approval_timeout_seconds(callback_handler)
+            final_decision = await _wait_for_async_tool_approval(
+                callback_handler,
+                tool_name=tool_name,
+                timeout_seconds=timeout_seconds,
+                tool_input=tool_input,
+                agent_id=agent_id,
+                ctx=ctx,
+            )
+            status, reason = _normalize_decision(final_decision, enforce=enforce)
+
+        # Fail closed: only an explicit "allow" may proceed. A terminal
+        # "pending" (approval timed out or the resolver returned pending
+        # again) is a non-decision, not a grant — blocking it here stops it
+        # from falling through and running the tool, matching LangChain.
+        if status != "allow":
+            blocked_result = _build_tool_deny_error(
+                tool_name=tool_name,
+                reason=reason,
+                is_pending_rejection=is_pending_flow,
+            )
+            # Guard the audit so its failure cannot re-enter this handler and
+            # downgrade a decided deny to an allow (AAASM-4782).
+            await _record_denied_tool_result(
+                callback_handler,
+                tool_name=tool_name,
+                tool_input=tool_input,
+                result=blocked_result,
+                agent_id=agent_id,
+                ctx=ctx,
+            )
+            return True, blocked_result, False
+    except Exception as error:
+        governance_failed = _is_governance_error(error)
+        if not governance_failed:
+            raise
+        # A governance-layer fault during the pre-execution check. Under
+        # enforce the SDK is a security control: fail closed by denying the
+        # call rather than running the tool ungoverned — matching every other
+        # adapter, which never swallows a governance error into an allow.
+        # Under observe/disabled fall through to run the tool (fail-open by
+        # design), preserving the dry-run/hermetic posture (AAASM-4782).
+        if enforce:
+            deny = _build_tool_deny_error(
+                tool_name=tool_name,
+                reason=_GOVERNANCE_FAULT_DENY_REASON,
+                is_pending_rejection=False,
+            )
+            return True, deny, True
+        return False, None, True
+
+    return False, None, False
+
+
 def _wrap_on_invoke_tool(tool_obj: Any, callback_handler: Any) -> None:
     """Wrap a single tool instance's ``on_invoke_tool`` coroutine with governance."""
     original_invoke = getattr(tool_obj, "on_invoke_tool", None)
@@ -525,67 +606,16 @@ async def governed_invoke(ctx: Any, tool_input: Any) -> Any:
         tool_name = str(getattr(tool_obj, "name", tool_obj.__class__.__name__))
         agent_id = _resolve_agent_id(ctx)
 
-        governance_failed = False
-        try:
-            decision = await _invoke_async_tool_check(
-                callback_handler,
-                tool_name=tool_name,
-                tool_input=tool_input,
-                agent_id=agent_id,
-                ctx=ctx,
-            )
-            status, reason = _normalize_decision(decision, enforce=enforce)
-            is_pending_flow = False
-            if status == "pending":
-                is_pending_flow = True
-                timeout_seconds = _get_pending_tool_approval_timeout_seconds(callback_handler)
-                final_decision = await _wait_for_async_tool_approval(
-                    callback_handler,
-                    tool_name=tool_name,
-                    timeout_seconds=timeout_seconds,
-                    tool_input=tool_input,
-                    agent_id=agent_id,
-                    ctx=ctx,
-                )
-                status, reason = _normalize_decision(final_decision, enforce=enforce)
-
-            # Fail closed: only an explicit "allow" may proceed. A terminal
-            # "pending" (approval timed out or the resolver returned pending
-            # again) is a non-decision, not a grant — blocking it here stops it
-            # from falling through and running the tool, matching LangChain.
-            if status != "allow":
-                blocked_result = _build_tool_deny_error(
-                    tool_name=tool_name,
-                    reason=reason,
-                    is_pending_rejection=is_pending_flow,
-                )
-                # Guard the audit so its failure cannot re-enter this handler and
-                # downgrade a decided deny to an allow (AAASM-4782).
-                await _record_denied_tool_result(
-                    callback_handler,
-                    tool_name=tool_name,
-                    tool_input=tool_input,
-                    result=blocked_result,
-                    agent_id=agent_id,
-                    ctx=ctx,
-                )
-                return blocked_result
-        except Exception as error:
-            governance_failed = _is_governance_error(error)
-            if not governance_failed:
-                raise
-            # A governance-layer fault during the pre-execution check. Under
-            # enforce the SDK is a security control: fail closed by denying the
-            # call rather than running the tool ungoverned — matching every other
-            # adapter, which never swallows a governance error into an allow.
-            # Under observe/disabled fall through to run the tool (fail-open by
-            # design), preserving the dry-run/hermetic posture (AAASM-4782).
-            if enforce:
-                return _build_tool_deny_error(
-                    tool_name=tool_name,
-                    reason=_GOVERNANCE_FAULT_DENY_REASON,
-                    is_pending_rejection=False,
-                )
+        return_now, return_value, governance_failed = await _run_async_tool_pre_execution_check(
+            callback_handler,
+            tool_name=tool_name,
+            tool_input=tool_input,
+            agent_id=agent_id,
+            ctx=ctx,
+            enforce=enforce,
+        )
+        if return_now:
+            return return_value
 
         result = original_invoke(ctx, tool_input)
         if inspect.isawaitable(result):

test/bench/conftest.py

@@ -12,7 +12,7 @@
 MAX_DETECTION_NS = 50_000_000  # <50ms detection overhead (AAASM-47)
 
 
-@pytest.fixture()
+@pytest.fixture
 def mock_gateway_client() -> MagicMock:
     """Return a MagicMock that satisfies GatewayClient interface."""
     client = MagicMock()
@@ -23,7 +23,7 @@ def mock_gateway_client() -> MagicMock:
     return client
 
 
-@pytest.fixture()
+@pytest.fixture
 def noop_interceptor() -> _NoopInterceptor:
     """Return a no-op governance interceptor for benchmarking hooks."""
     return _NoopInterceptor()

test/bench/test_latency_contracts.py

@@ -328,12 +328,12 @@ def test_detection_latency_under_50ms() -> None:
 # ---------------------------------------------------------------------------
 
 
-def test_init_assembly_coldstart_latency() -> None:
+def test_init_assembly_coldstart_latency(monkeypatch: pytest.MonkeyPatch) -> None:
     """Measure init_assembly() cold-start P50/P95/P99."""
     samples: list[int] = []
 
     for _ in range(_ITERATIONS):
-        assembly_mod._ACTIVE_CONTEXT = None
+        monkeypatch.setattr(assembly_mod, "_ACTIVE_CONTEXT", None)
 
         start = time.perf_counter_ns()
         ctx = init_assembly(

test/integration/test_native_core_runtime.py

@@ -148,7 +148,7 @@ def make_audit_entry_payload(index: int, *, worker_id: int = 0) -> str:
     )
 
 
-@pytest.fixture()
+@pytest.fixture
 def native_core() -> Any:
     if os.getenv("AAASM_RUN_NATIVE_CORE_TESTS") != "1":
         pytest.skip("Set AAASM_RUN_NATIVE_CORE_TESTS=1 to run native core runtime tests.")

test/unit/adapters/crewai/test_crewai_spawn_context.py

@@ -65,14 +65,14 @@ def setup_method(self) -> None:
     def teardown_method(self) -> None:
         _revert_task_execute_sync_patch(FakeTask)
 
-    def test_spawn_ctx_set_during_execute_sync(self) -> None:
+    def test_spawn_ctx_set_during_execute_sync(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-x")
         task.execute_sync()
@@ -87,40 +87,40 @@ def test_spawn_ctx_reset_after_execute_sync(self) -> None:
         task.execute_sync()
         assert _SPAWN_CTX.get() is None
 
-    def test_spawn_ctx_reset_on_exception(self) -> None:
+    def test_spawn_ctx_reset_on_exception(self, monkeypatch: pytest.MonkeyPatch) -> None:
         def failing_execute(self: object, *args: object, **kwargs: object) -> str:
             raise RuntimeError("task failed")
 
-        FakeTask.execute_sync = failing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", failing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-z")
 
         with pytest.raises(RuntimeError):
             task.execute_sync()
         assert _SPAWN_CTX.get() is None
 
-    def test_no_spawn_ctx_when_no_agent_id(self) -> None:
+    def test_no_spawn_ctx_when_no_agent_id(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id=None)
         task.execute_sync()
 
         assert captured[0] is None
 
-    def test_team_id_extracted_from_agent_crew(self) -> None:
+    def test_team_id_extracted_from_agent_crew(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-a")
         crew = MagicMock()
@@ -131,14 +131,14 @@ def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
         assert captured[0] is not None
         assert captured[0].team_id == "crew-uuid-123"
 
-    def test_team_id_none_when_no_crew(self) -> None:
+    def test_team_id_none_when_no_crew(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-b")
         task.agent = MagicMock(spec=["id"])
@@ -148,14 +148,14 @@ def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
         assert captured[0] is not None
         assert captured[0].team_id is None
 
-    def test_delegation_reason_from_task_description(self) -> None:
+    def test_delegation_reason_from_task_description(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-c")
         task.description = "Analyze quarterly reports"
@@ -164,14 +164,14 @@ def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
         assert captured[0] is not None
         assert captured[0].delegation_reason == "Analyze quarterly reports"
 
-    def test_delegation_reason_truncated_to_256_chars(self) -> None:
+    def test_delegation_reason_truncated_to_256_chars(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-d")
         task.description = "x" * 300
@@ -180,14 +180,14 @@ def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
         assert captured[0] is not None
         assert len(captured[0].delegation_reason) == 256  # type: ignore[arg-type]
 
-    def test_delegation_reason_none_when_description_empty(self) -> None:
+    def test_delegation_reason_none_when_description_empty(self, monkeypatch: pytest.MonkeyPatch) -> None:
         captured: list[SpawnContext | None] = []
 
         def capturing_execute(self: object, *args: object, **kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeTask.execute_sync = capturing_execute  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeTask, "execute_sync", capturing_execute)
         _apply_task_execute_sync_patch(FakeTask, MagicMock())
         task = FakeTask(agent_id="worker-e")
         task.description = ""
@@ -280,17 +280,16 @@ def teardown_method(self) -> None:
             if hasattr(FakeCrew, attr):
                 delattr(FakeCrew, attr)
 
-    def test_non_hierarchical_kickoff_bypasses_spawn_ctx(self) -> None:
+    def test_non_hierarchical_kickoff_bypasses_spawn_ctx(self, monkeypatch: pytest.MonkeyPatch) -> None:
         from unittest.mock import patch
 
         captured: list[SpawnContext | None] = []
-        original = FakeCrew.kickoff
 
         def capturing_kickoff(self: object, *_args: object, **_kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeCrew.kickoff = capturing_kickoff  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeCrew, "kickoff", capturing_kickoff)
         _apply_crew_kickoff_patch(FakeCrew)
 
         crew = FakeCrew(hierarchical=False)
@@ -301,19 +300,17 @@ def capturing_kickoff(self: object, *_args: object, **_kwargs: object) -> str:
             crew.kickoff()
 
         assert captured[0] is None
-        FakeCrew.kickoff = original  # type: ignore[method-assign]  # reassign fake method to install/restore stub
 
-    def test_hierarchical_kickoff_sets_spawn_ctx(self) -> None:
+    def test_hierarchical_kickoff_sets_spawn_ctx(self, monkeypatch: pytest.MonkeyPatch) -> None:
         from unittest.mock import patch
 
         captured: list[SpawnContext | None] = []
-        original = FakeCrew.kickoff
 
         def capturing_kickoff(self: object, *_args: object, **_kwargs: object) -> str:
             captured.append(_SPAWN_CTX.get())
             return "done"
 
-        FakeCrew.kickoff = capturing_kickoff  # type: ignore[method-assign]  # fake method swap
+        monkeypatch.setattr(FakeCrew, "kickoff", capturing_kickoff)
         _apply_crew_kickoff_patch(FakeCrew)
 
         crew = FakeCrew(hierarchical=True, manager_id="mgr-42", crew_id="crew-99")
@@ -329,7 +326,6 @@ def capturing_kickoff(self: object, *_args: object, **_kwargs: object) -> str:
         assert sc.team_id == "crew-99"
         assert sc.spawned_by_tool == "crewai_kickoff_hierarchical"
         assert sc.depth == 1
-        FakeCrew.kickoff = original  # type: ignore[method-assign]  # reassign fake method to install/restore stub
 
     def test_spawn_ctx_reset_after_hierarchical_kickoff(self) -> None:
         from unittest.mock import patch