From 7b9a0b315bf92e0654d76846d281bed6e52def1f Mon Sep 17 00:00:00 2001 From: xxxigm Date: Sun, 21 Jun 2026 14:55:00 +0700 Subject: [PATCH] test(mcp): cover 'unknown method' ping keepalive fallback (#50028) Two regression tests for the agentmemory reconnect-loop: - _is_method_not_found_error matches the plain 'Unknown method: ping' phrasing (no structural -32601 code). - _keepalive_probe latches _ping_unsupported and falls back to list_tools when send_ping raises 'Unknown method: ping', instead of propagating (which would reconnect-loop). --- tests/tools/test_mcp_capability_gating.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/tools/test_mcp_capability_gating.py b/tests/tools/test_mcp_capability_gating.py index 551af1340d7..95fddb11093 100644 --- a/tests/tools/test_mcp_capability_gating.py +++ b/tests/tools/test_mcp_capability_gating.py @@ -254,6 +254,12 @@ class TestMethodNotFoundDetection: from tools.mcp_tool import _is_method_not_found_error assert _is_method_not_found_error(Exception("Method not found")) is True + def test_unknown_method_phrasing_is_match(self): + # agentmemory's MCP server surfaces method-not-found as a plain + # "Unknown method: ping" string with no structural -32601 code (#50028). + from tools.mcp_tool import _is_method_not_found_error + assert _is_method_not_found_error(Exception("Unknown method: ping")) is True + def test_unrelated_exception_is_not_match(self): from tools.mcp_tool import _is_method_not_found_error assert _is_method_not_found_error(TimeoutError()) is False @@ -295,6 +301,23 @@ class TestKeepaliveProbeFallback: task.session.list_tools.assert_awaited_once() assert task._ping_unsupported is True + async def test_falls_back_on_unknown_method_string(self): + """Regression for #50028: a server that surfaces method-not-found as a + plain "Unknown method: ping" string (no structural -32601 code) must + still latch the fallback and use list_tools, NOT reconnect-loop.""" + task = MCPServerTask("test") + task.initialize_result = _caps(tools=SimpleNamespace()) + task.session = SimpleNamespace( + send_ping=AsyncMock(side_effect=Exception("Unknown method: ping")), + list_tools=AsyncMock(return_value=SimpleNamespace(tools=[])), + ) + + await task._keepalive_probe() + + task.session.send_ping.assert_awaited_once() + task.session.list_tools.assert_awaited_once() + assert task._ping_unsupported is True + async def test_latch_skips_ping_on_subsequent_cycles(self): task = MCPServerTask("test") task.initialize_result = _caps(tools=SimpleNamespace())