Merge pull request #36 from jkawamoto/add-file

Implement `/add-file` endpoint with tests

Author: jkawamoto

README.md

@@ -82,7 +82,7 @@ Refer to Bear's [X-callback-url Scheme documentation](https://bear.app/faq/x-cal
 - [x] /open-note
 - [x] /create
 - [x] /add-text (partially, via the replace_note method)
-- [ ] /add-file
+- [x] /add-file
 - [x] /tags
 - [x] /open-tag
 - [x] /rename-tag

src/mcp_bear/__init__.py

@@ -6,6 +6,7 @@
 #
 #  http://opensource.org/licenses/mit-license.php
 import asyncio
+import base64
 import json
 import logging
 import webbrowser
@@ -19,6 +20,7 @@
 from typing import cast, AsyncIterator, Final, Any, Mapping
 from urllib.parse import urlencode, quote
 
+import requests
 from fastapi import FastAPI, Request, HTTPException
 from mcp.server import FastMCP
 from mcp.server.fastmcp import Context
@@ -246,6 +248,55 @@ async def replace_note(
         finally:
             del ctx.request_context.lifespan_context.futures[req_id]
 
+    @mcp.tool()
+    async def add_file(
+        ctx: Context[Any, AppContext],
+        id: str | None = Field(description="note unique identifier", default=None),
+        title: str | None = Field(description="note title", default=None),
+        file: str = Field(description="base64 representation of a file or a URL to a file to add to the note"),
+        header: str | None = Field(
+            description="if specified add the file to the corresponding header inside the note", default=None
+        ),
+        filename: str = Field(description="file name with extension"),
+        mode: str | None = Field(description="adding mode (prepend, append)", default=None),
+    ) -> None:
+        """Append or prepend a file to a note identified by its title or id."""
+        req_id = ctx.request_id
+
+        if file.startswith("http://") or file.startswith("https://"):
+            res = requests.get(file)
+            res.raise_for_status()
+            file = base64.b64encode(res.content).decode("ascii")
+
+        params = {
+            "selected": "no",
+            "file": file,
+            "filename": filename,
+            "open_note": "no",
+            "new_window": "no",
+            "show_window": "no",
+            "edit": "no",
+            "x-success": f"xfwder://{uds.stem}/{req_id}/success",
+            "x-error": f"xfwder://{uds.stem}/{req_id}/error",
+        }
+        if id is not None:
+            params["id"] = id
+        if title is not None:
+            params["title"] = title
+        if header is not None:
+            params["header"] = header
+        if mode is not None:
+            params["mode"] = mode
+
+        future = Future[QueryParams]()
+        ctx.request_context.lifespan_context.futures[req_id] = future
+        try:
+            webbrowser.open(f"{BASE_URL}/add-file?{urlencode(sorted(params.items()), quote_via=quote)}")
+            await future
+
+        finally:
+            del ctx.request_context.lifespan_context.futures[req_id]
+
     @mcp.tool()
     async def tags(
         ctx: Context[Any, AppContext],

tests/test_mcp.py

@@ -29,6 +29,8 @@ async def test_list_tools(mcp_client_session: ClientSession) -> None:
 
     assert "open_note" in tools
     assert "create" in tools
+    assert "replace_note" in tools
+    assert "add_file" in tools
     assert "tags" in tools
     assert "open_tag" in tools
     assert "rename_tag" in tools
@@ -41,4 +43,3 @@ async def test_list_tools(mcp_client_session: ClientSession) -> None:
     assert "locked" in tools
     assert "search" in tools
     assert "grab_url" in tools
-    assert "replace_note" in tools

tests/test_server.py

@@ -97,6 +97,16 @@ def side_effect(url, _new=0, _autoraise=True) -> bool:
         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"}]
@@ -268,6 +278,112 @@ async def test_replace_note_failed(
     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,