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).
This commit is contained in:
xxxigm 2026-06-21 14:55:00 +07:00 committed by kshitij
parent 472c068159
commit 7b9a0b315b

View file

@ -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())