test(mcp): cover v2 reconnect, transport wiring, and session security

- Split test_invoke_hook_reconnects_on_session_terminated: the session-terminated
  success path (reconnect then retry succeeds) was dead code shadowed by a second
  plugin setup. Restore it as a real assertion and re-add the separate
  no-reconnect-on-other-errors test that had been merged away.
- Assert streamable_http_client is called with a pre-built http_client instance
  and terminate_on_close=True (the v2 API change and DELETE-on-close behavior
  that replaced the removed __terminate_http_session).
- Assert UDS servers populate _transport_security with DNS rebinding protection,
  non-UDS servers leave it unset, and run_streamable_http_async forwards both
  transport_security and the real bind host to streamable_http_app().

Signed-off-by: Frederico Araujo <frederico.araujo@ibm.com>

Author: araujof

tests/unit/cpex/framework/external/mcp/server/test_runtime_coverage.py

@@ -49,6 +49,48 @@ def test_uds_sets_transport_security(self, tmp_path):
         config = MCPServerConfig(host="127.0.0.1", port=8000, uds=uds_path)
         server = runtime.SSLCapableMCPServer(server_config=config, name="UDSTest")
         assert server.server_config.uds == uds_path
+        # UDS servers get DNS rebinding protection auto-configured in __init__.
+        assert server._transport_security is not None
+        assert server._transport_security.enable_dns_rebinding_protection is True
+
+    def test_non_uds_leaves_transport_security_unset(self):
+        config = MCPServerConfig(host="0.0.0.0", port=8000)
+        server = runtime.SSLCapableMCPServer(server_config=config, name="NonUDSTest")
+        # Non-UDS servers rely on the SDK/host handling, not the UDS auto-config.
+        assert server._transport_security is None
+
+    @pytest.mark.asyncio
+    async def test_run_streamable_http_async_forwards_transport_security(self, tmp_path, monkeypatch):
+        """run_streamable_http_async must forward _transport_security to streamable_http_app()."""
+        uds_path = str(tmp_path / "plugin.sock")
+        config = MCPServerConfig(host="127.0.0.1", port=8000, uds=uds_path)
+        server = runtime.SSLCapableMCPServer(server_config=config, name="ForwardTest")
+
+        captured = {}
+
+        def capture_app(**kwargs):
+            captured.update(kwargs)
+            return SimpleNamespace(routes=[])
+
+        server.streamable_http_app = capture_app
+        monkeypatch.setattr(runtime.SSLCapableMCPServer, "_get_ssl_config", lambda self: {})
+
+        class DummyServer:
+            def __init__(self, config):
+                self.config = config
+
+            async def serve(self):
+                pass
+
+        monkeypatch.setattr(runtime.uvicorn, "Config", lambda **kwargs: SimpleNamespace(**kwargs))
+        monkeypatch.setattr(runtime.uvicorn, "Server", lambda config: DummyServer(config))
+
+        await runtime.SSLCapableMCPServer.run_streamable_http_async(server)
+
+        assert captured["transport_security"] is server._transport_security
+        assert captured["transport_security"].enable_dns_rebinding_protection is True
+        # Real bind host is forwarded so the SDK does not force a localhost-only allowlist.
+        assert captured["host"] == "127.0.0.1"
 
     def test_ssl_config_partial_tls_warns(self, tmp_path, caplog):
         """TLS present but no keyfile/certfile returns empty dict + warning."""

tests/unit/cpex/framework/external/mcp/test_client_coverage.py

@@ -237,6 +237,41 @@ def mock_streamable(*args, **kwargs):
             with pytest.raises(PluginError, match="connection failed after 3 attempts"):
                 await plugin._ExternalPlugin__connect_to_http_server("http://localhost:9999/mcp")
 
+    @pytest.mark.asyncio
+    async def test_streamable_client_called_with_prebuilt_client_and_terminate_on_close(self):
+        """v2 SDK contract: streamable_http_client receives a pre-built httpx client and
+        terminate_on_close=True (replacing the removed manual __terminate_http_session)."""
+        plugin = _make_plugin()
+
+        class OkCtx:
+            async def __aenter__(self):
+                return AsyncMock(), AsyncMock()
+
+            async def __aexit__(self, *args):
+                return False
+
+        mock_session = AsyncMock()
+        mock_session.initialize = AsyncMock()
+        list_tools_result = MagicMock()
+        list_tools_result.tools = []
+        mock_session.list_tools = AsyncMock(return_value=list_tools_result)
+
+        prebuilt_client = AsyncMock(spec=httpx.AsyncClient)
+
+        with (
+            patch("cpex.framework.external.mcp.client.streamable_http_client", return_value=OkCtx()) as mock_streamable,
+            patch("cpex.framework.external.mcp.client.ClientSession", return_value=mock_session),
+            patch("cpex.framework.external.mcp.client.httpx.AsyncClient", return_value=prebuilt_client),
+        ):
+            plugin._exit_stack = AsyncExitStack()
+            await plugin._ExternalPlugin__connect_to_http_server("http://localhost:9999/mcp")
+
+        mock_streamable.assert_called_once()
+        _, kwargs = mock_streamable.call_args
+        assert kwargs["terminate_on_close"] is True
+        # A pre-built AsyncClient instance is passed, not a factory callable (v2 API change).
+        assert kwargs["http_client"] is prebuilt_client
+
 
 # ===========================================================================
 # Shutdown

tests/unit/cpex/framework/external/mcp/test_client_reconnect.py

@@ -248,6 +248,21 @@ async def mock_call_tool(*args, **kwargs):
             from mcp_types import CallToolResult, TextContent
 
             return CallToolResult(content=[TextContent(type="text", text='{"result": {"name": "test", "args": {}}}')])
+
+        mock_session.call_tool = mock_call_tool
+
+        with patch("cpex.framework.external.mcp.client.get_hook_registry") as mock_registry:
+            mock_registry.return_value.get_result_type.return_value = ToolPreInvokePayload
+            with patch.object(plugin, "_reconnect_session", new_callable=AsyncMock) as mock_reconnect:
+                payload = ToolPreInvokePayload(name="test", args={})
+                result = await plugin.invoke_hook("tool_pre_invoke", payload, mock_plugin_context)
+                mock_reconnect.assert_called_once()
+                assert result is not None
+                # The call is retried after reconnect: first raises, second succeeds.
+                assert call_count == 2
+
+    @pytest.mark.asyncio
+    async def test_invoke_hook_no_reconnect_on_other_plugin_errors(self, mock_http_plugin_config, mock_plugin_context):
         plugin = ExternalPlugin(mock_http_plugin_config)
         mock_session = AsyncMock()
         plugin._session = mock_session