Bear MCP Server

# test_server.py # # Copyright (c) 2025 Junpei Kawamoto # # This software is released under the MIT License. # # http://opensource.org/licenses/mit-license.php # type: ignore import json import os import random import webbrowser from pathlib import Path from typing import Generator, Tuple, AsyncGenerator, Any from unittest.mock import patch, MagicMock from urllib.parse import urlparse, parse_qs, urlencode, quote import pytest from mcp.server import FastMCP from mcp.server.fastmcp import Context from mcp.server.fastmcp.exceptions import ToolError from mcp.shared.context import RequestContext from mcp_bear import server, AppContext, BASE_URL, Note, NoteID, NoteInfo, ModifiedNote from mcp_bear.cli import generate_file_suffix BEAR_TOKEN = "abcdefg" def _encode_tags(obj: dict) -> dict: if "tags" in obj: obj["tags"] = json.dumps(obj["tags"]) return obj @pytest.fixture def temp_socket() -> Generator[Path, None, None]: while True: uds = Path("/tmp").joinpath(f"mcp-bear-{generate_file_suffix()}.sock") if not uds.exists(): break yield uds if uds.exists(): os.unlink(uds) @pytest.fixture async def mcp_server(temp_socket: Path) -> AsyncGenerator[Tuple[FastMCP, Context[Any, AppContext]], None]: s = server(BEAR_TOKEN, temp_socket) async with s._mcp_server.lifespan(s) as lifespan_context: # noinspection PyTypeChecker ctx = Context( request_context=RequestContext( request_id=random.randint(1, 100), meta=None, session=None, lifespan_context=lifespan_context, request=None, ) ) yield s, ctx @pytest.fixture def mock_webbrowser() -> Generator[MagicMock, None, None]: original_open = webbrowser.open with patch("webbrowser.open") as mock_open: def side_effect(url, _new=0, _autoraise=True) -> bool: queries = parse_qs(urlparse(url).query) callback_url = queries.get("x-success")[0] return original_open(f"{callback_url}?{urlencode(mock_open.stubbed_queries, quote_via=quote)}") mock_open.side_effect = side_effect yield mock_open @pytest.fixture def mock_webbrowser_error() -> Generator[MagicMock, None, None]: original_open = webbrowser.open with patch("webbrowser.open") as mock_open: def side_effect(url, _new=0, _autoraise=True) -> bool: queries = parse_qs(urlparse(url).query) callback_url = queries.get("x-error")[0] params = {"error-Code": "499", "errorMessage": "test error message"} return original_open(f"{callback_url}?{urlencode(params, quote_via=quote)}") mock_open.side_effect = side_effect yield mock_open @pytest.fixture def mock_requests_get() -> Generator[MagicMock, None, None]: with patch("requests.get") as mock_get: mock_response = MagicMock() mock_response.status_code = 200 mock_response.content = b"mocked http request" mock_get.return_value = mock_response yield mock_get @pytest.mark.anyio @pytest.mark.parametrize( "arguments", [{"id": "1234567890"}, {"title": "test note"}, {"id": "1234567890", "title": "test note"}] ) async def test_open_note( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server expect = Note( note="test note" * 16 * 1024, # > 16KB identifier="1234567890", title="test note", tags=["a", "b"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", body="test note", ) mock_webbrowser.stubbed_queries = _encode_tags(expect.model_dump()) res = await s._tool_manager.call_tool("open_note", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "new_window": "no", "float": "no", "show_window": "no", "open_note": "no", "selected": "no", "edit": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/open-note?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_open_note_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("open_note", arguments={"id": "1234567890", "title": "test note"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize( "arguments,expect_req_params", [ ({"text": "test note"}, {"text": "test note"}), ({"title": "test title", "text": "test note"}, {"title": "test title", "text": "test note"}), ({"title": "test title", "text": "# test title\ntest note"}, {"title": "test title", "text": "\ntest note"}), ( {"title": "test title", "text": "test note", "tags": ["a", "b"]}, {"title": "test title", "text": "test note", "tags": "a,b"}, ), ( {"title": "test title", "text": "test note", "timestamp": True}, {"title": "test title", "text": "test note", "timestamp": "yes"}, ), ], ) async def test_create( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, expect_req_params: dict, ) -> None: s, ctx = mcp_server expect = NoteID(identifier="1234567890", title="test title") mock_webbrowser.stubbed_queries = expect.model_dump() res = await s._tool_manager.call_tool("create", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "open_note": "no", "new_window": "no", "float": "no", "show_window": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(expect_req_params) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/create?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_create_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("create", arguments={"text": "test note"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize( "arguments,expect_req_params", [ ({"id": "123456"}, {"id": "123456"}), ({"text": "test note"}, {"text": "test note"}), ( {"title": "test title", "text": "test note"}, { "text": "test note", "title": "test title", }, ), ( {"title": "test title", "text": "test note", "tags": ["a", "b"]}, {"text": "test note", "title": "test title", "tags": "a,b"}, ), ( {"title": "test title", "text": "test note", "timestamp": True}, {"text": "test note", "title": "test title", "timestamp": "yes"}, ), ], ) async def test_replace_note( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, expect_req_params: dict, ) -> None: s, ctx = mcp_server expect = ModifiedNote(note="updated note", title="test title") mock_webbrowser.stubbed_queries = expect.model_dump() res = await s._tool_manager.call_tool("replace_note", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "mode": "replace_all" if "title" in arguments else "replace", "open_note": "no", "new_window": "no", "show_window": "no", "edit": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(expect_req_params) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/add-text?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_replace_note_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("replace_note", arguments={"id": "123456", "text": "new text"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize( "arguments,expect_req_params", [ ( {"id": "123456", "file": "dGVzdA==", "filename": "test.txt"}, {"id": "123456", "file": "dGVzdA==", "filename": "test.txt"}, ), ( {"title": "sample note", "file": "dGVzdA==", "filename": "test.txt"}, {"title": "sample note", "file": "dGVzdA==", "filename": "test.txt"}, ), ( {"id": "123456", "file": "dGVzdA==", "filename": "test.txt", "header": "supplement"}, {"id": "123456", "file": "dGVzdA==", "filename": "test.txt", "header": "supplement"}, ), ( {"id": "123456", "file": "dGVzdA==", "filename": "test.txt", "mode": "prepend"}, {"id": "123456", "file": "dGVzdA==", "filename": "test.txt", "mode": "prepend"}, ), ], ) async def test_add_file( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, expect_req_params: dict, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = {} await s._tool_manager.call_tool("add_file", arguments=arguments, context=ctx) assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "selected": "no", "open_note": "no", "new_window": "no", "show_window": "no", "edit": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(expect_req_params) mock_webbrowser.assert_called_once_with( f"{BASE_URL}/add-file?{urlencode(sorted(req_params.items()), quote_via=quote)}" ) @pytest.mark.anyio @pytest.mark.parametrize( "arguments,expect_req_params", [ ( {"id": "123456", "file": "http://example.com", "filename": "test.txt"}, {"id": "123456", "file": "bW9ja2VkIGh0dHAgcmVxdWVzdA==", "filename": "test.txt"}, ), ( {"title": "sample note", "file": "https://example.com", "filename": "test.txt"}, {"title": "sample note", "file": "bW9ja2VkIGh0dHAgcmVxdWVzdA==", "filename": "test.txt"}, ), ], ) async def test_add_file_http_request( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, mock_requests_get: MagicMock, arguments: dict, expect_req_params: dict, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = {} await s._tool_manager.call_tool("add_file", arguments=arguments, context=ctx) assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "selected": "no", "open_note": "no", "new_window": "no", "show_window": "no", "edit": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(expect_req_params) mock_webbrowser.assert_called_once_with( f"{BASE_URL}/add-file?{urlencode(sorted(req_params.items()), quote_via=quote)}" ) mock_requests_get.assert_called_once_with(arguments["file"]) @pytest.mark.anyio async def test_add_file_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("add_file", arguments={"file": "dGVzdA==", "filename": "test.txt"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio async def test_tags( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = { "tags": json.dumps( [ {"name": "a"}, {"name": "b"}, {"name": "c"}, ] ) } res = await s._tool_manager.call_tool("tags", arguments={}, context=ctx) assert res == ["a", "b", "c"] assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } mock_webbrowser.assert_called_once_with(f"{BASE_URL}/tags?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_tags_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("tags", arguments={}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio async def test_open_tag( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, ) -> None: s, ctx = mcp_server expect = [ NoteInfo( title="note a", identifier="1", tags=["test", "data"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", ), NoteInfo( title="note b", identifier="2", modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z" ), ] mock_webbrowser.stubbed_queries = { "notes": json.dumps([info.model_dump() for info in expect]), } res = await s._tool_manager.call_tool("open_tag", arguments={"name": "test_tag"}, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "name": "test_tag", "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } mock_webbrowser.assert_called_once_with(f"{BASE_URL}/open-tag?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_open_tag_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("open_tag", arguments={"name": "a"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio async def test_rename_tag( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = {} await s._tool_manager.call_tool("rename_tag", arguments={"name": "old name", "new_name": "new name"}, context=ctx) assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "name": "old name", "new_name": "new name", "show_window": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } mock_webbrowser.assert_called_once_with(f"{BASE_URL}/rename-tag?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_rename_tag_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool( "rename_tag", arguments={"name": "old name", "new_name": "new name"}, context=ctx ) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio async def test_delete_tag( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = {} await s._tool_manager.call_tool("delete_tag", arguments={"name": "tag name"}, context=ctx) assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "name": "tag name", "show_window": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } mock_webbrowser.assert_called_once_with(f"{BASE_URL}/delete-tag?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_delete_tag_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("delete_tag", arguments={"name": "tag name"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize( "arguments", [{"id": "1234567890"}, {"search": "test note"}, {"id": "1234567890", "search": "test note"}] ) async def test_trash( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = {} await s._tool_manager.call_tool("trash", arguments=arguments, context=ctx) assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/trash?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_trash_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("trash", arguments={"search": "tag name"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize( "arguments", [{"id": "1234567890"}, {"search": "test note"}, {"id": "1234567890", "search": "test note"}] ) async def test_archive( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server mock_webbrowser.stubbed_queries = {} await s._tool_manager.call_tool("archive", arguments=arguments, context=ctx) assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/archive?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_archive_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("archive", arguments={"search": "tag name"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize("arguments", [{}, {"search": "keyword"}]) async def test_untagged( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server expect = [ NoteInfo( title="note a", identifier="1", tags=["test", "data"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", ), NoteInfo( title="note b", identifier="2", modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z" ), ] mock_webbrowser.stubbed_queries = {"notes": json.dumps([info.model_dump() for info in expect])} res = await s._tool_manager.call_tool("untagged", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/untagged?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_untagged_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("untagged", arguments={}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize("arguments", [{}, {"search": "keyword"}]) async def test_todo( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server expect = [ NoteInfo( title="note a", identifier="1", tags=["test", "data"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", ), NoteInfo( title="note b", identifier="2", modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z" ), ] mock_webbrowser.stubbed_queries = {"notes": json.dumps([info.model_dump() for info in expect])} res = await s._tool_manager.call_tool("todo", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/todo?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_todo_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("todo", arguments={}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize("arguments", [{}, {"search": "keyword"}]) async def test_today( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server expect = [ NoteInfo( title="note a", identifier="1", tags=["test", "data"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", ), NoteInfo( title="note b", identifier="2", modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z" ), ] mock_webbrowser.stubbed_queries = {"notes": json.dumps([info.model_dump() for info in expect])} res = await s._tool_manager.call_tool("today", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/today?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_today_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("today", arguments={}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize("arguments", [{}, {"search": "keyword"}]) async def test_locked( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server expect = [ NoteInfo( title="note a", identifier="1", tags=["test", "data"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", ), NoteInfo( title="note b", identifier="2", modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z" ), ] mock_webbrowser.stubbed_queries = {"notes": json.dumps([info.model_dump() for info in expect])} res = await s._tool_manager.call_tool("locked", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/locked?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_locked_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("locked", arguments={}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize( "arguments", [{"term": "1234567890"}, {"tag": "test note"}, {"term": "1234567890", "tag": "test note"}] ) async def test_search( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, arguments: dict, ) -> None: s, ctx = mcp_server expect = [ NoteInfo( title="note a", identifier="1", tags=["test", "data"], modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z", ), NoteInfo( title="note b", identifier="2", modificationDate="2023-01-01T00:00:00Z", creationDate="2023-01-01T00:00:00Z" ), ] mock_webbrowser.stubbed_queries = {"notes": json.dumps([info.model_dump() for info in expect])} res = await s._tool_manager.call_tool("search", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "show_window": "no", "token": BEAR_TOKEN, "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update(arguments) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/search?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_search_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("search", arguments={"tags": "tag"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0 @pytest.mark.anyio @pytest.mark.parametrize("tags", [None, ["tag 1"], ["tag 1", "tag 2"]]) async def test_grab_url( temp_socket: Path, mcp_server: Tuple[FastMCP, Context], mock_webbrowser: MagicMock, tags: list[str] | None, ) -> None: s, ctx = mcp_server expect = NoteID(identifier="1234567890", title="test title") mock_webbrowser.stubbed_queries = expect.model_dump() arguments = {"url": "https://bear.app"} arguments.update({"tags": tags} if tags else {}) res = await s._tool_manager.call_tool("grab_url", arguments=arguments, context=ctx) assert res == expect assert len(ctx.request_context.lifespan_context.futures) == 0 req_params = { "url": "https://bear.app", "x-success": f"xfwder://{temp_socket.stem}/{ctx.request_id}/success", "x-error": f"xfwder://{temp_socket.stem}/{ctx.request_id}/error", } req_params.update({"tags": ",".join(tags)} if tags else {}) mock_webbrowser.assert_called_once_with(f"{BASE_URL}/grab-url?{urlencode(req_params, quote_via=quote)}") @pytest.mark.anyio async def test_grab_url_failed( mcp_server: Tuple[FastMCP, Context[Any, AppContext]], mock_webbrowser_error: MagicMock ) -> None: s, ctx = mcp_server with pytest.raises(ToolError) as excinfo: await s._tool_manager.call_tool("grab_url", arguments={"url": "https://bear.app"}, context=ctx) assert "test error message" in str(excinfo.value) assert len(ctx.request_context.lifespan_context.futures) == 0