[AAASM-4691] ✅ (python-sdk): Fix pytest.raises single-statement smells (S5778)

test/integration/test_native_core_runtime.py

@@ -275,11 +275,12 @@ def test_wired_check_blocks_on_runtime_deny(native_core: Any) -> None:
     try:
         interceptor = RuntimeQueryInterceptor(object(), client, "agent-001")
         handler = AssemblyCallbackHandler(interceptor)
+        run_id = uuid4()
         with pytest.raises(ToolExecutionBlockedError):
             handler.on_tool_start(
                 serialized={"name": "web_search"},
                 input_str="query",
-                run_id=uuid4(),
+                run_id=run_id,
                 tool_name="web_search",
                 args={"q": "x"},
             )

test/integration/test_spawn_lineage_integration.py

@@ -178,9 +178,16 @@ def test_exception_in_scope_still_resets_ctx() -> None:
     """_SPAWN_CTX is reset even when an exception is raised inside the scope."""
     ctx = SpawnContext(parent_agent_id="root", depth=0, spawned_by_tool=None)
 
-    with pytest.raises(RuntimeError, match="intentional"), spawn_context_scope(ctx):
-        assert _SPAWN_CTX.get() is not None
-        raise RuntimeError("intentional")
+    # The RuntimeError must propagate *through* spawn_context_scope.__exit__ so the
+    # test exercises the reset-on-exception path; the helper keeps that propagation
+    # while leaving pytest.raises wrapping a single call.
+    def _raise_inside_scope() -> None:
+        with spawn_context_scope(ctx):
+            assert _SPAWN_CTX.get() is not None
+            raise RuntimeError("intentional")
+
+    with pytest.raises(RuntimeError, match="intentional"):
+        _raise_inside_scope()
 
     assert _SPAWN_CTX.get() is None
 

test/unit/adapters/langchain/test_callback_handler_async.py

@@ -35,12 +35,13 @@ async def on_llm_end(self, **kwargs: object) -> None:
 @pytest.mark.asyncio
 async def test_aon_tool_start_raises_when_governance_denies() -> None:
     handler = AssemblyCallbackHandler(AsyncInterceptor())
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError):
         await handler.aon_tool_start(
             serialized={"name": "web_search"},
             input_str="query",
-            run_id=uuid4(),
+            run_id=run_id,
             decision={"status": "deny", "reason": "blocked"},
         )
 
@@ -108,12 +109,13 @@ class _EnforcingAsyncInterceptor(AsyncInterceptor):
 @pytest.mark.parametrize("decision", [None, "maybe", 12345, {"status": "garbage"}, {}])
 async def test_aon_tool_start_denies_unknown_under_enforce(decision: object) -> None:
     handler = AssemblyCallbackHandler(_EnforcingAsyncInterceptor())
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError):
         await handler.aon_tool_start(
             serialized={"name": "web_search"},
             input_str="query",
-            run_id=uuid4(),
+            run_id=run_id,
             decision=decision,
         )
 

test/unit/adapters/langchain/test_callback_handler_sync.py

@@ -38,12 +38,13 @@ def on_llm_end(self, **kwargs: object) -> None:
 
 def test_on_tool_start_raises_when_governance_denies() -> None:
     handler = AssemblyCallbackHandler(SyncInterceptor())
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError):
         handler.on_tool_start(
             serialized={"name": "web_search"},
             input_str="query",
-            run_id=uuid4(),
+            run_id=run_id,
             decision={"status": "deny", "reason": "blocked"},
         )
 
@@ -79,12 +80,13 @@ def test_on_tool_start_waits_for_pending_approval() -> None:
 
 def test_on_tool_start_blocks_when_pending_never_approved() -> None:
     handler = AssemblyCallbackHandler(SyncInterceptor())
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError):
         handler.on_tool_start(
             serialized={"name": "calendar_write"},
             input_str="create event",
-            run_id=uuid4(),
+            run_id=run_id,
             decision={"status": "pending"},
             approval_decision={"status": "deny"},
         )
@@ -145,12 +147,13 @@ class _EnforcingInterceptor(SyncInterceptor):
 )
 def test_unknown_decision_denies_under_enforce(decision: object) -> None:
     handler = AssemblyCallbackHandler(_EnforcingInterceptor())
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError):
         handler.on_tool_start(
             serialized={"name": "web_search"},
             input_str="query",
-            run_id=uuid4(),
+            run_id=run_id,
             decision=decision,
         )
 

test/unit/adapters/langchain/test_langgraph_patch.py

@@ -77,8 +77,9 @@ def on_graph_node_start(self, **kwargs: object) -> None:
             del kwargs
             raise TypeError("internal callback failure")
 
+    recorder = InternalTypeErrorRecorder()
     with pytest.raises(TypeError, match="internal callback failure"):
-        langgraph_patch._invoke_pre_node_hook(InternalTypeErrorRecorder(), "n4", {"state": 4})
+        langgraph_patch._invoke_pre_node_hook(recorder, "n4", {"state": 4})
 
 
 @pytest.mark.asyncio

test/unit/adapters/mcp/test_patch.py

@@ -277,8 +277,9 @@ async def check_tool_start(self, **kwargs: object) -> object:
     patcher = mcp_patch.MCPClientPatch(EnforcingUnknownInterceptor(), process_agent_id="agent-9")
     assert patcher.apply() is True
 
+    session = FakeClientSession()
     with pytest.raises(MCPToolBlockedError):
-        await FakeClientSession().call_tool("some_tool", {"q": "x"})
+        await session.call_tool("some_tool", {"q": "x"})
 
 
 @pytest.mark.asyncio

test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_patch.py

@@ -130,8 +130,9 @@ async def test_deny_blocks_tool_and_prevents_side_effect(monkeypatch: pytest.Mon
     patcher = maf_patch.MicrosoftAgentFrameworkPatch(_DenyInterceptor())
     assert patcher.apply() is True
 
+    tool = FakeTool()
     with pytest.raises(PolicyViolationError, match="blocked by governance policy"):
-        await FakeTool().invoke(arguments={"city": "Seattle"})
+        await tool.invoke(arguments={"city": "Seattle"})
 
     # The wrapped function must NOT have run — a no-op patch would let it through.
     assert side_effects == []
@@ -147,8 +148,9 @@ async def test_pending_then_deny_rejects_at_approval(monkeypatch: pytest.MonkeyP
     patcher = maf_patch.MicrosoftAgentFrameworkPatch(_PendingThenDenyInterceptor())
     assert patcher.apply() is True
 
+    tool = FakeTool()
     with pytest.raises(PolicyViolationError, match="rejected during approval"):
-        await FakeTool().invoke(arguments={"x": 1})
+        await tool.invoke(arguments={"x": 1})
 
     assert side_effects == []
 
@@ -170,8 +172,9 @@ def check_tool_start(self, **kwargs: Any) -> dict[str, str]:
     patcher = maf_patch.MicrosoftAgentFrameworkPatch(_MalformedInterceptor())
     assert patcher.apply() is True
 
+    tool = FakeTool()
     with pytest.raises(PolicyViolationError):
-        await FakeTool().invoke(arguments={"x": 1})
+        await tool.invoke(arguments={"x": 1})
     assert side_effects == []
 
     patcher.revert()

test/unit/adapters/openai_agents/test_patch.py

@@ -300,8 +300,9 @@ async def check_tool_start(self, **kwargs: object) -> dict[str, str]:
     assert patcher.apply() is True
 
     tool = function_tool_cls(name="fail_closed_tool")
+    ctx = SimpleNamespace(agent_id="agent-fail-closed")
     with pytest.raises(RuntimeError, match="gateway unavailable"):
-        await tool.on_invoke_tool(SimpleNamespace(agent_id="agent-fail-closed"), '{"x": 2}')
+        await tool.on_invoke_tool(ctx, '{"x": 2}')
 
 
 @pytest.mark.asyncio
@@ -349,8 +350,9 @@ async def check_tool_start(self, **kwargs: object) -> dict[str, str]:
     assert patcher.apply() is True
 
     tool = function_tool_cls(name="strict_tool")
+    ctx = SimpleNamespace(agent_id="agent-strict")
     with pytest.raises(ValueError, match="unexpected"):
-        await tool.on_invoke_tool(SimpleNamespace(agent_id="agent-strict"), '{"x": 3}')
+        await tool.on_invoke_tool(ctx, '{"x": 3}')
 
 
 @pytest.mark.asyncio

test/unit/adapters/openai_agents/test_runner_spawn_patch.py

@@ -96,8 +96,9 @@ async def failing_run(agent: object, *, input: object, **kwargs: object) -> str:
         FakeRunner.run = classmethod(failing_run)  # type: ignore[arg-type,assignment,method-assign]
         _apply_runner_run_patch(FakeRunner, "process-agent-001")
 
+        agent = MagicMock()
         with pytest.raises(RuntimeError):
-            await FakeRunner.run(MagicMock(), input="x")
+            await FakeRunner.run(agent, input="x")
         assert _SPAWN_CTX.get() is None
 
     def test_idempotent_apply(self) -> None:
@@ -232,8 +233,9 @@ async def failing_invoke(ctx: Any, input_json: Any) -> str:
         from agent_assembly.adapters.openai_agents.patch import _wrap_on_invoke_handoff
 
         _wrap_on_invoke_handoff(h, "process-agent-001")
+        ctx = MagicMock()
         with pytest.raises(RuntimeError, match="handoff failed"):
-            await h.on_invoke_handoff(MagicMock(), "{}")
+            await h.on_invoke_handoff(ctx, "{}")
         assert _SPAWN_CTX.get() is None
 
     def test_idempotent_apply(self) -> None:

test/unit/adapters/pydantic_ai/test_pydantic_ai_patch.py

@@ -152,9 +152,10 @@ async def check_tool_start(self, **kwargs: object) -> dict[str, str]:
 
     tool = FakeTool()
     ctx = SimpleNamespace(deps=SimpleNamespace(assembly_agent_id="agent-a"), run_id="run-1")
+    args = _ArgsModel({"topic": "finance"})
 
     with pytest.raises(PolicyViolationError, match="blocked by governance policy: blocked for safety"):
-        await tool._run(ctx, _ArgsModel({"topic": "finance"}))
+        await tool._run(ctx, args)
 
 
 @pytest.mark.asyncio
@@ -214,9 +215,10 @@ async def wait_for_tool_approval(self, **kwargs: object) -> dict[str, str]:
 
     tool = FakeTool()
     ctx = SimpleNamespace(deps=SimpleNamespace(assembly_agent_id="agent-c"), run_id="run-3")
+    args = _ArgsModel({"q": "secret"})
 
     with pytest.raises(PolicyViolationError, match="rejected during approval: approval rejected"):
-        await tool._run(ctx, _ArgsModel({"q": "secret"}))
+        await tool._run(ctx, args)
 
 
 @pytest.mark.asyncio

test/unit/adapters/pydantic_ai/test_pydantic_ai_spawn_patch.py

@@ -120,8 +120,9 @@ async def failing_run(self: object, *args: object, **kwargs: object) -> str:
         FakeAgent.run = failing_run  # type: ignore[method-assign]  # reassign fake method to install/restore stub
         _apply_agent_run_patch(FakeAgent, "pydantic-parent")
 
+        agent = FakeAgent()
         with pytest.raises(RuntimeError):
-            await FakeAgent().run("x")
+            await agent.run("x")
         assert _SPAWN_CTX.get() is None
 
     def test_spawn_ctx_reset_on_exception_sync(self) -> None:
@@ -131,8 +132,9 @@ def failing_run_sync(self: object, *args: object, **kwargs: object) -> str:
         FakeAgent.run_sync = failing_run_sync  # type: ignore[method-assign]  # fake method swap
         _apply_agent_run_patch(FakeAgent, "pydantic-parent")
 
+        agent = FakeAgent()
         with pytest.raises(RuntimeError):
-            FakeAgent().run_sync("x")
+            agent.run_sync("x")
         assert _SPAWN_CTX.get() is None
 
     @pytest.mark.asyncio
@@ -287,8 +289,10 @@ async def failing_run(self: object, ctx: object, args: object, **kw: object) ->
         FakeTool._run = failing_run  # type: ignore[assignment,method-assign]  # fake method swap
         _apply_tool_run_patch(FakeTool, _FakeAllowHandler())
 
+        tool = FakeTool()
+        ctx = _FakeCtx()
         with pytest.raises(RuntimeError):
-            await FakeTool()._run(_FakeCtx(), {})
+            await tool._run(ctx, {})
         assert _SPAWN_CTX.get() is None
 
     @pytest.mark.asyncio
@@ -304,8 +308,10 @@ async def should_not_be_called(self: object, ctx: object, args: object, **kw: ob
 
         from agent_assembly.exceptions import PolicyViolationError
 
+        tool = FakeTool()
+        ctx = _FakeCtx()
         with pytest.raises(PolicyViolationError):
-            await FakeTool()._run(_FakeCtx(), {})
+            await tool._run(ctx, {})
 
         assert called == []
         assert _SPAWN_CTX.get() is None

test/unit/adapters/test_base.py

@@ -103,8 +103,10 @@ def unregister_hooks(self) -> None:
 
 
 def test_register_raises_validation_error_for_invalid_contract() -> None:
+    adapter = InvalidRegistrationAdapter()
+    contract = object()
     with pytest.raises(AdapterValidationError):
-        InvalidRegistrationAdapter().register(object())
+        adapter.register(contract)
 
 
 class EmptyVersionsAdapter(FrameworkAdapter):
@@ -122,8 +124,9 @@ def unregister_hooks(self) -> None:
 
 
 def test_validate_registration_rejects_empty_supported_versions() -> None:
+    adapter = EmptyVersionsAdapter()
     with pytest.raises(AdapterValidationError, match="supported versions must not be empty"):
-        EmptyVersionsAdapter().validate_registration()
+        adapter.validate_registration()
 
 
 class BlankVersionRangeAdapter(FrameworkAdapter):
@@ -141,8 +144,9 @@ def unregister_hooks(self) -> None:
 
 
 def test_validate_registration_rejects_blank_version_range() -> None:
+    adapter = BlankVersionRangeAdapter()
     with pytest.raises(AdapterValidationError, match="version ranges must be non-empty"):
-        BlankVersionRangeAdapter().validate_registration()
+        adapter.validate_registration()
 
 
 class AgentIdAwareAdapter(FrameworkAdapter):

test/unit/cli/test_loader.py

@@ -65,8 +65,9 @@ def test_file_with_no_adapter_raises(self, tmp_path: object) -> None:
         assert isinstance(tmp_path, pathlib.Path)
         empty_file = tmp_path / "empty.py"
         empty_file.write_text("x = 1\n")
+        path = str(empty_file)
         with pytest.raises(ValueError, match="No FrameworkAdapter subclass"):
-            load_adapter_class_from_path(str(empty_file))
+            load_adapter_class_from_path(path)
 
 
 class TestLoadAdapterClass:

test/unit/core/test_runtime_interceptor.py

@@ -118,12 +118,13 @@ def test_callback_handler_blocks_on_runtime_deny() -> None:
     """End-to-end: a DENY runtime drives on_tool_start to raise."""
     interceptor = RuntimeQueryInterceptor(_FakeGatewayClient(), _FakeRuntimeClient("deny", reason="nope"), "agent-001")
     handler = AssemblyCallbackHandler(interceptor)
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError, match="nope"):
         handler.on_tool_start(
             serialized={"name": "web_search"},
             input_str="query",
-            run_id=uuid4(),
+            run_id=run_id,
             tool_name="web_search",
             args={"q": "x"},
         )
@@ -428,12 +429,13 @@ def query_policy(self, *_args: Any, **_kwargs: Any) -> dict[str, str]:
 
     interceptor = RuntimeQueryInterceptor(_FakeGatewayClient(), _Raising(), "agent-001", enforce=True)
     handler = AssemblyCallbackHandler(interceptor)
+    run_id = uuid4()
 
     with pytest.raises(ToolExecutionBlockedError):
         handler.on_tool_start(
             serialized={"name": "web_search"},
             input_str="query",
-            run_id=uuid4(),
+            run_id=run_id,
             tool_name="web_search",
             args={"q": "x"},
         )