import asyncio
import types
import logging
import pytest
from domin8 import server
from mcp.types import TextContent
@pytest.mark.asyncio
async def test_call_tool_fallback_handler_raises(caplog, capsys):
caplog.set_level(logging.ERROR, logger='domin8')
# Use a dotted name to trigger normalization to request_change
name = "domin8.request_change"
arguments = {"summary": "Update README", "diff": "--- a/README.md\n+++ b/README.md\n@@ -1 +1 @@\n-old\n+new\n", "actor_id": "agent"}
# Monkeypatch the actual handler to raise
import importlib
module = importlib.import_module("domin8.tools.request_change")
orig = getattr(module, "handle_request_change")
async def _raise(arguments):
raise RuntimeError("handler failure")
setattr(module, "handle_request_change", _raise)
try:
with pytest.raises(Exception):
await server.call_tool(name, arguments)
finally:
# restore to avoid leaking into other tests
setattr(module, "handle_request_change", orig)
@pytest.mark.asyncio
async def test_call_tool_unknown_tool_raises():
with pytest.raises(ValueError):
await server.call_tool('i_do_not_exist', {})
@pytest.mark.asyncio
async def test_main_populate_tool_cache_exception(monkeypatch):
# Make list_tools raise to exercise the logger.exception branch in main
async def _raise():
raise RuntimeError('no tools')
monkeypatch.setattr(server, 'list_tools', _raise)
# Ensure stdio_server and app.run are no-ops to let main finish quickly
class DummyCtx:
async def __aenter__(self):
return (object(), object())
async def __aexit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(server, 'stdio_server', lambda: DummyCtx())
async def _run(read, write, init_opts):
return None
monkeypatch.setattr(server.app, 'run', _run)
# Should not raise even if list_tools raises
await server.main()
@pytest.mark.asyncio
async def test_main_populate_tool_cache_and_cached_definition_raises(monkeypatch):
# Return a tool and make _get_cached_tool_definition raise to hit the except branch in main
async def _list():
from types import SimpleNamespace
return [SimpleNamespace(name='request_change')]
async def _get_cached(name):
raise RuntimeError('cache fail')
monkeypatch.setattr(server, 'list_tools', _list)
monkeypatch.setattr(server.app, '_get_cached_tool_definition', _get_cached)
class DummyCtx:
async def __aenter__(self):
return (object(), object())
async def __aexit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(server, 'stdio_server', lambda: DummyCtx())
async def _run(read, write, init_opts):
return None
monkeypatch.setattr(server.app, 'run', _run)
await server.main()
@pytest.mark.asyncio
async def test_main_stdio_setup_failure(monkeypatch):
# Make stdio_server raise on context manager entry to trigger the except branch
class BadCtx:
async def __aenter__(self):
raise RuntimeError('stdio broken')
async def __aexit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(server, 'stdio_server', lambda: BadCtx())
# app.run should not be called; run main and ensure it returns without raising
await server.main()
@pytest.mark.asyncio
async def test_create_initialization_options_with_tools_attach_failure(caplog, monkeypatch):
caplog.set_level(logging.ERROR)
# Make app.create_initialization_options return an object whose capabilities.tools
# raises on setattr to exercise the exception branch
class BadTools:
def __setattr__(self, name, value):
raise RuntimeError("setattr failed")
class BadInitOpts:
def __init__(self):
class Cap:
pass
self.capabilities = types.SimpleNamespace()
self.capabilities.tools = BadTools()
monkeypatch.setattr(server.app, "create_initialization_options", lambda: BadInitOpts())
# Ensure list_tools returns at least one tool to populate
async def _fake_list_tools():
return []
monkeypatch.setattr(server, "list_tools", _fake_list_tools)
init_opts = await server.create_initialization_options_with_tools()
# Should still return without raising; ensure it doesn't raise and returns an init options object
assert init_opts is not None
@pytest.mark.asyncio
async def test_main_installs_wrapper_and_handle_message_logging(monkeypatch, caplog):
caplog.set_level(logging.DEBUG, logger='domin8')
# Replace app._handle_message with a dummy that records a call
called = {}
async def original_dummy(message, session, lifespan_context, raise_exceptions=False):
called['ok'] = True
return "original-result"
monkeypatch.setattr(server.app, "_handle_message", original_dummy)
# Replace stdio_server with an async context manager that yields simple streams
class DummyCtx:
async def __aenter__(self):
return (object(), object())
async def __aexit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(server, "stdio_server", lambda: DummyCtx())
# Patch app.run to be a no-op async function
async def _run(read, write, init_opts):
return None
monkeypatch.setattr(server.app, "run", _run)
# Run main() to install wrapper
await server.main()
# After main, app._handle_message should now be the wrapper
assert server.app._handle_message is not original_dummy
# Create a message that has no 'method' and causes the fallback branch
class BadMessage:
def __getattr__(self, name):
raise RuntimeError("boom")
# Call the wrapper; it should call the original dummy and log "Incoming message (no method attribute)"
res = await server.app._handle_message(BadMessage(), None, None)
assert called.get('ok')
assert res == "original-result"
