marcus
c47221dc8f
Implements SYNO.FileStation.Search with async polling (exponential backoff 200ms→2s, 60s timeout) and SYNO.FileStation.Download with base64 output and a 10 MB size cap. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
677 lines
23 KiB
Python
677 lines
23 KiB
Python
"""Tests for tools/filestation.py: list_shares and list_dir."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from mcp_synology_filestation.client import SynologyError
|
||
from mcp_synology_filestation.config import AppConfig, ConnectionConfig
|
||
|
||
|
||
@pytest.fixture()
|
||
def config() -> AppConfig:
|
||
return AppConfig(
|
||
schema_version=1,
|
||
connection=ConnectionConfig(host="nas.example.com"),
|
||
)
|
||
|
||
|
||
def _make_mcp_and_tools(config: AppConfig, client: MagicMock) -> dict:
|
||
"""Register FileStation tools on a mock FastMCP and collect them by name."""
|
||
from mcp_synology_filestation.tools.filestation import register_filestation
|
||
|
||
registered: dict[str, object] = {}
|
||
|
||
mcp = MagicMock()
|
||
|
||
def tool_decorator():
|
||
"""Simulate @mcp.tool() — capture the decorated function."""
|
||
|
||
def decorator(fn):
|
||
registered[fn.__name__] = fn
|
||
return fn
|
||
|
||
return decorator
|
||
|
||
mcp.tool = tool_decorator
|
||
register_filestation(mcp, config, client)
|
||
return registered
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# list_shares
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_shares_success(config: AppConfig) -> None:
|
||
"""list_shares returns a formatted table on success."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
return_value={
|
||
"shares": [
|
||
{
|
||
"name": "data",
|
||
"path": "/data",
|
||
"additional": {
|
||
"real_path": "/volume1/data",
|
||
"volume_status": {
|
||
"totalspace": 2_000_000_000,
|
||
"usedspace": 500_000_000,
|
||
},
|
||
},
|
||
},
|
||
{
|
||
"name": "photos",
|
||
"path": "/photos",
|
||
"additional": {
|
||
"real_path": "/volume1/photos",
|
||
"volume_status": {},
|
||
},
|
||
},
|
||
]
|
||
}
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_shares"]()
|
||
|
||
assert "data" in result
|
||
assert "/data" in result # share path, not volume path
|
||
assert "/volume1/data" not in result
|
||
assert "photos" in result
|
||
assert "2 share(s) found" in result
|
||
# Check usage percentage appears
|
||
assert "%" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_shares_empty(config: AppConfig) -> None:
|
||
"""list_shares returns a friendly message when no shares exist."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(return_value={"shares": []})
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_shares"]()
|
||
|
||
assert "No shared folders" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_shares_dsm_error(config: AppConfig) -> None:
|
||
"""list_shares returns an Error: message on SynologyError."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
side_effect=SynologyError("Permission denied — check DSM user permissions", code=105)
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_shares"]()
|
||
|
||
assert result.startswith("Error:")
|
||
assert "Permission denied" in result
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# list_dir
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_success(config: AppConfig) -> None:
|
||
"""list_dir returns a formatted table with name, type, size, and modified columns."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
return_value={
|
||
"total": 3,
|
||
"files": [
|
||
{
|
||
"name": "documents",
|
||
"isdir": True,
|
||
"additional": {"size": 0, "time": {"mtime": 1700000000}},
|
||
},
|
||
{
|
||
"name": "photo.jpg",
|
||
"isdir": False,
|
||
"additional": {"size": 2_048_000, "time": {"mtime": 1710000000}},
|
||
},
|
||
{
|
||
"name": "readme.txt",
|
||
"isdir": False,
|
||
"additional": {"size": 512, "time": {"mtime": 1720000000}},
|
||
},
|
||
],
|
||
}
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_dir"](path="/dev")
|
||
|
||
assert "documents" in result
|
||
assert "photo.jpg" in result
|
||
assert "readme.txt" in result
|
||
assert "dir" in result
|
||
assert "file" in result
|
||
# Size column: dirs show "-", files show human-readable size
|
||
assert "2 MB" in result or "1 MB" in result # photo.jpg ~2 MB
|
||
assert "512 B" in result # readme.txt
|
||
# Modified column present
|
||
assert "Modified" in result
|
||
assert "Showing 1–3 of 3 item(s)" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_pagination(config: AppConfig) -> None:
|
||
"""list_dir shows pagination hint when more items are available."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
return_value={
|
||
"total": 200,
|
||
"files": [
|
||
{
|
||
"name": f"file{i}.txt",
|
||
"isdir": False,
|
||
"additional": {"size": 100, "time": {"mtime": 1700000000}},
|
||
}
|
||
for i in range(100)
|
||
],
|
||
}
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_dir"](path="/volume1/data", limit=100)
|
||
|
||
assert "Showing 1–100 of 200" in result
|
||
assert "offset=100" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_empty(config: AppConfig) -> None:
|
||
"""list_dir returns a friendly message for empty directories."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(return_value={"total": 0, "files": []})
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_dir"](path="/volume1/empty")
|
||
|
||
assert "empty" in result.lower() or "does not exist" in result.lower()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_invalid_sort_by(config: AppConfig) -> None:
|
||
"""list_dir returns an Error: message for invalid sort_by."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock()
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_dir"](path="/volume1/data", sort_by="invalid_field")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "sort_by" in result
|
||
client.request.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_invalid_sort_direction(config: AppConfig) -> None:
|
||
"""list_dir returns an Error: message for invalid sort_direction."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock()
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_dir"](path="/volume1/data", sort_direction="random")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "sort_direction" in result
|
||
client.request.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_limit_clamped(config: AppConfig) -> None:
|
||
"""list_dir clamps limit to _MAX_LIMIT (500)."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(return_value={"total": 1, "files": []})
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
await tools["list_dir"](path="/volume1/data", limit=9999)
|
||
|
||
call_params = client.request.call_args[1]["params"]
|
||
assert call_params["limit"] == 500
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_dir_dsm_error(config: AppConfig) -> None:
|
||
"""list_dir returns Error: on SynologyError (e.g. path not found)."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(side_effect=SynologyError("File or folder not found", code=1800))
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["list_dir"](path="/volume1/nonexistent")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "not found" in result.lower()
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# get_info
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_info_single_file(config: AppConfig) -> None:
|
||
"""get_info returns a table with metadata for a single file."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
return_value={
|
||
"files": [
|
||
{
|
||
"path": "/dev/notes.txt",
|
||
"name": "notes.txt",
|
||
"isdir": False,
|
||
"additional": {
|
||
"real_path": "/volume1/dev/notes.txt",
|
||
"size": 1024,
|
||
"time": {"mtime": 1700000000, "crtime": 1690000000},
|
||
"owner": {"user": "marcus", "group": "users"},
|
||
"perm": {"posix": 0o644},
|
||
},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["get_info"](path="/dev/notes.txt")
|
||
|
||
assert "/dev/notes.txt" in result
|
||
assert "file" in result
|
||
assert "1 KB" in result or "1024 B" in result
|
||
assert "marcus" in result
|
||
assert "users" in result
|
||
assert "644" in result
|
||
assert "/volume1/dev/notes.txt" in result
|
||
assert "1 item(s)" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_info_directory(config: AppConfig) -> None:
|
||
"""get_info shows '-' for size of a directory."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
return_value={
|
||
"files": [
|
||
{
|
||
"path": "/dev",
|
||
"name": "dev",
|
||
"isdir": True,
|
||
"additional": {
|
||
"real_path": "/volume1/dev",
|
||
"size": 0,
|
||
"time": {"mtime": 1700000000, "crtime": 1690000000},
|
||
"owner": {"user": "marcus", "group": "users"},
|
||
"perm": {"posix": 0o755},
|
||
},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["get_info"](path="/dev")
|
||
|
||
assert "dir" in result
|
||
assert "755" in result
|
||
# Size for directory should be "-"
|
||
rows = [line for line in result.splitlines() if "/dev" in line and "|" in line]
|
||
assert rows, "expected a data row containing /dev"
|
||
size_col = rows[0].split("|")[3].strip()
|
||
assert size_col == "-"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_info_multiple_paths(config: AppConfig) -> None:
|
||
"""get_info handles comma-separated paths and returns one row per item."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
return_value={
|
||
"files": [
|
||
{
|
||
"path": "/dev/a.txt",
|
||
"name": "a.txt",
|
||
"isdir": False,
|
||
"additional": {
|
||
"size": 100,
|
||
"time": {"mtime": 1700000000, "crtime": 1690000000},
|
||
"owner": {"user": "marcus", "group": "users"},
|
||
"perm": {"posix": 0o644},
|
||
},
|
||
},
|
||
{
|
||
"path": "/data/b.txt",
|
||
"name": "b.txt",
|
||
"isdir": False,
|
||
"additional": {
|
||
"size": 200,
|
||
"time": {"mtime": 1700000001, "crtime": 1690000001},
|
||
"owner": {"user": "marcus", "group": "users"},
|
||
"perm": {"posix": 0o644},
|
||
},
|
||
},
|
||
]
|
||
}
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["get_info"](path="/dev/a.txt,/data/b.txt")
|
||
|
||
assert "/dev/a.txt" in result
|
||
assert "/data/b.txt" in result
|
||
assert "2 item(s)" in result
|
||
|
||
# Verify the API received paths as a JSON array (confirmed working format)
|
||
call_params = client.request.call_args[1]["params"]
|
||
assert call_params["path"] == json.dumps(["/dev/a.txt", "/data/b.txt"])
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_info_empty_path(config: AppConfig) -> None:
|
||
"""get_info returns Error: when path is empty."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock()
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["get_info"](path=" ")
|
||
|
||
assert result.startswith("Error:")
|
||
client.request.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_info_dsm_error(config: AppConfig) -> None:
|
||
"""get_info returns Error: on SynologyError."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(side_effect=SynologyError("File or folder not found", code=1800))
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["get_info"](path="/dev/missing.txt")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "not found" in result.lower()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_info_uses_getinfo_method(config: AppConfig) -> None:
|
||
"""get_info calls SYNO.FileStation.List with method='getinfo'."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(return_value={"files": []})
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
await tools["get_info"](path="/dev/file.txt")
|
||
|
||
client.request.assert_called_once()
|
||
call_args = client.request.call_args
|
||
assert call_args[0][0] == "SYNO.FileStation.List"
|
||
assert call_args[0][1] == "getinfo"
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# search
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
_SEARCH_FILE = {
|
||
"path": "/docker/app/compose.yaml",
|
||
"name": "compose.yaml",
|
||
"isdir": False,
|
||
"additional": {"size": 1024, "time": {"mtime": 1700000000}},
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_success(config: AppConfig) -> None:
|
||
"""search returns a formatted table after polling two rounds until finished=True."""
|
||
client = MagicMock()
|
||
|
||
call_count = 0
|
||
|
||
async def _request(api, method, **kwargs):
|
||
nonlocal call_count
|
||
call_count += 1
|
||
if method == "start":
|
||
return {"taskid": "abc123", "has_not_index_share": False}
|
||
if method == "list":
|
||
# First poll: not finished yet; second poll: finished
|
||
finished = call_count >= 4 # start=1, list1=2, list2=3 → finished on call 3
|
||
return {"files": [_SEARCH_FILE], "finished": finished, "total": 1}
|
||
if method == "clean":
|
||
return {}
|
||
return {}
|
||
|
||
client.request = AsyncMock(side_effect=_request)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
result = await tools["search"](path="/docker", pattern="*.yaml")
|
||
|
||
assert "/docker/app/compose.yaml" in result
|
||
assert "file" in result
|
||
assert "1 match(es) found" in result
|
||
# Verify start was called with correct params
|
||
start_call = client.request.call_args_list[0]
|
||
assert start_call[0][0] == "SYNO.FileStation.Search"
|
||
assert start_call[0][1] == "start"
|
||
assert start_call[1]["params"]["folder_path"] == "/docker"
|
||
assert start_call[1]["params"]["pattern"] == "*.yaml"
|
||
assert start_call[1]["params"]["recursive"] == "true"
|
||
# Verify clean was called last
|
||
last_call = client.request.call_args_list[-1]
|
||
assert last_call[0][1] == "clean"
|
||
assert last_call[1]["params"]["taskid"] == "abc123"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_polls_until_finished(config: AppConfig) -> None:
|
||
"""search keeps polling when finished=False and stops once finished=True."""
|
||
client = MagicMock()
|
||
poll_calls = 0
|
||
|
||
async def _request(api, method, **kwargs):
|
||
nonlocal poll_calls
|
||
if method == "start":
|
||
return {"taskid": "t1"}
|
||
if method == "list":
|
||
poll_calls += 1
|
||
return {
|
||
"files": [_SEARCH_FILE],
|
||
"finished": poll_calls >= 3, # finish on third list call
|
||
"total": 1,
|
||
}
|
||
return {}
|
||
|
||
client.request = AsyncMock(side_effect=_request)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
result = await tools["search"](path="/docker", pattern="*.yaml")
|
||
|
||
assert poll_calls == 3
|
||
assert "1 match(es) found" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_no_results(config: AppConfig) -> None:
|
||
"""search returns a friendly message when no files are found."""
|
||
client = MagicMock()
|
||
|
||
async def _request(api, method, **kwargs):
|
||
if method == "start":
|
||
return {"taskid": "t2"}
|
||
if method == "list":
|
||
return {"files": [], "finished": True, "total": 0}
|
||
return {}
|
||
|
||
client.request = AsyncMock(side_effect=_request)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
result = await tools["search"](path="/docker", pattern="*.txt")
|
||
|
||
assert "No files" in result or "not found" in result.lower()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_start_dsm_error(config: AppConfig) -> None:
|
||
"""search returns Error: immediately when the start call fails."""
|
||
client = MagicMock()
|
||
client.request = AsyncMock(
|
||
side_effect=SynologyError("Permission denied — check DSM user permissions", code=105)
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
result = await tools["search"](path="/docker", pattern="*.yaml")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "Permission denied" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_list_dsm_error(config: AppConfig) -> None:
|
||
"""search returns Error: and cleans up when a list poll fails."""
|
||
client = MagicMock()
|
||
|
||
async def _request(api, method, **kwargs):
|
||
if method == "start":
|
||
return {"taskid": "t3"}
|
||
if method == "list":
|
||
raise SynologyError("Unknown error", code=100)
|
||
return {}
|
||
|
||
client.request = AsyncMock(side_effect=_request)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
result = await tools["search"](path="/docker", pattern="*.yaml")
|
||
|
||
assert result.startswith("Error:")
|
||
# clean should have been attempted
|
||
clean_calls = [c for c in client.request.call_args_list if c[0][1] == "clean"]
|
||
assert len(clean_calls) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_recursive_false(config: AppConfig) -> None:
|
||
"""search passes recursive=false when recursive=False."""
|
||
client = MagicMock()
|
||
|
||
async def _request(api, method, **kwargs):
|
||
if method == "start":
|
||
return {"taskid": "t4"}
|
||
if method == "list":
|
||
return {"files": [], "finished": True, "total": 0}
|
||
return {}
|
||
|
||
client.request = AsyncMock(side_effect=_request)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
await tools["search"](path="/docker", pattern="*.yaml", recursive=False)
|
||
|
||
start_call = client.request.call_args_list[0]
|
||
assert start_call[1]["params"]["recursive"] == "false"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_additional_format(config: AppConfig) -> None:
|
||
"""search uses json.dumps(["size","time"]) for additional parameter."""
|
||
client = MagicMock()
|
||
|
||
async def _request(api, method, **kwargs):
|
||
if method == "start":
|
||
return {"taskid": "t5"}
|
||
if method == "list":
|
||
return {"files": [], "finished": True, "total": 0}
|
||
return {}
|
||
|
||
client.request = AsyncMock(side_effect=_request)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
await tools["search"](path="/docker", pattern="*.yaml")
|
||
|
||
list_call = client.request.call_args_list[1]
|
||
assert list_call[1]["params"]["additional"] == json.dumps(["size", "time"])
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# download
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_success(config: AppConfig) -> None:
|
||
"""download returns JSON with filename, size, and valid base64 content."""
|
||
content = b"hello, world"
|
||
client = MagicMock()
|
||
client.download_bytes = AsyncMock(return_value=("compose.yaml", content))
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["download"](path="/docker/app/compose.yaml")
|
||
|
||
parsed = json.loads(result)
|
||
assert parsed["filename"] == "compose.yaml"
|
||
assert parsed["size"] == len(content)
|
||
assert base64.b64decode(parsed["content_base64"]) == content
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_size_limit(config: AppConfig) -> None:
|
||
"""download returns Error: when file exceeds 10 MB."""
|
||
large_content = b"x" * (10 * 1024 * 1024 + 1)
|
||
client = MagicMock()
|
||
client.download_bytes = AsyncMock(return_value=("bigfile.bin", large_content))
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["download"](path="/data/bigfile.bin")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "10 MB" in result or "exceeds" in result
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_dsm_error(config: AppConfig) -> None:
|
||
"""download returns Error: on SynologyError."""
|
||
client = MagicMock()
|
||
client.download_bytes = AsyncMock(
|
||
side_effect=SynologyError("File or folder not found", code=1800)
|
||
)
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["download"](path="/data/missing.txt")
|
||
|
||
assert result.startswith("Error:")
|
||
assert "not found" in result.lower()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_exactly_10mb(config: AppConfig) -> None:
|
||
"""download accepts files exactly at the 10 MB boundary."""
|
||
content = b"x" * (10 * 1024 * 1024)
|
||
client = MagicMock()
|
||
client.download_bytes = AsyncMock(return_value=("edge.bin", content))
|
||
|
||
tools = _make_mcp_and_tools(config, client)
|
||
result = await tools["download"](path="/data/edge.bin")
|
||
|
||
parsed = json.loads(result)
|
||
assert parsed["size"] == len(content)
|