Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| dbab842738 |
@@ -538,6 +538,63 @@ def register_filestation(
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@mcp.tool()
|
||||
async def check_exist(path: str) -> str:
|
||||
"""Check whether one or more files or folders exist on the NAS.
|
||||
|
||||
Accepts a single path or a comma-separated list of paths.
|
||||
Use share paths as returned by list_shares (e.g. "/dev/file.txt").
|
||||
|
||||
Note: SYNO.FileStation.CheckExist returns error 400 on this firmware for all
|
||||
parameter formats. This tool falls back to SYNO.FileStation.List::getinfo, which
|
||||
returns an entry per path with name=None when the path does not exist.
|
||||
|
||||
Args:
|
||||
path: One or more share-relative paths, comma-separated
|
||||
(e.g. "/dev/notes.txt" or "/dev/notes.txt,/data/photo.jpg").
|
||||
|
||||
Returns:
|
||||
Formatted table with each path and whether it exists (Yes / No).
|
||||
"""
|
||||
from mcp_synology_filestation.client import SynologyError
|
||||
|
||||
paths = [p.strip() for p in path.split(",") if p.strip()]
|
||||
if not paths:
|
||||
return "Error: no path provided."
|
||||
|
||||
try:
|
||||
data = await client.request(
|
||||
"SYNO.FileStation.List",
|
||||
"getinfo",
|
||||
params={
|
||||
"path": json.dumps(paths),
|
||||
"additional": json.dumps([]),
|
||||
},
|
||||
)
|
||||
except SynologyError as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
files: list[dict] = data.get("files", [])
|
||||
if not files:
|
||||
return "No information returned for the given path(s)."
|
||||
|
||||
# A path that doesn't exist still gets an entry but with name=None
|
||||
rows = [(f.get("path", ""), "Yes" if f.get("name") is not None else "No") for f in files]
|
||||
|
||||
w_path = max(len("Path"), *(len(r[0]) for r in rows))
|
||||
w_exists = len("Exists") # "Yes" / "No" always shorter
|
||||
|
||||
sep = f"+{'-' * (w_path + 2)}+{'-' * (w_exists + 2)}+"
|
||||
header = f"| {'Path':<{w_path}} | {'Exists':<{w_exists}} |"
|
||||
|
||||
lines = [sep, header, sep]
|
||||
for item_path, exists_str in rows:
|
||||
lines.append(f"| {item_path:<{w_path}} | {exists_str:<{w_exists}} |")
|
||||
lines.append(sep)
|
||||
lines.append(f"\n{len(rows)} path(s) checked.")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
# ── write tools ───────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
|
||||
@@ -1075,3 +1075,125 @@ async def test_upload_dsm_error(config: AppConfig) -> None:
|
||||
|
||||
assert result.startswith("Error:")
|
||||
assert "permission" in result.lower()
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
# check_exist
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_exist_single_existing(config: AppConfig) -> None:
|
||||
"""check_exist returns Yes for a path that exists."""
|
||||
client = MagicMock()
|
||||
client.request = AsyncMock(
|
||||
return_value={
|
||||
"files": [
|
||||
{"path": "/docker", "name": "docker", "isdir": True, "additional": {}},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
tools = _make_mcp_and_tools(config, client)
|
||||
result = await tools["check_exist"](path="/docker")
|
||||
|
||||
assert "/docker" in result
|
||||
assert "Yes" in result
|
||||
assert "No" not in result
|
||||
assert "1 path(s) checked" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_exist_single_missing(config: AppConfig) -> None:
|
||||
"""check_exist returns No for a path that does not exist (name=None from DSM)."""
|
||||
client = MagicMock()
|
||||
client.request = AsyncMock(
|
||||
return_value={
|
||||
"files": [
|
||||
{"path": "/no-such-path", "name": None, "isdir": None, "additional": None},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
tools = _make_mcp_and_tools(config, client)
|
||||
result = await tools["check_exist"](path="/no-such-path")
|
||||
|
||||
assert "/no-such-path" in result
|
||||
assert "No" in result
|
||||
assert "1 path(s) checked" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_exist_multi_path(config: AppConfig) -> None:
|
||||
"""check_exist handles comma-separated paths and reports each correctly."""
|
||||
client = MagicMock()
|
||||
client.request = AsyncMock(
|
||||
return_value={
|
||||
"files": [
|
||||
{"path": "/docker", "name": "docker", "isdir": True, "additional": {}},
|
||||
{"path": "/ghost", "name": None, "isdir": None, "additional": None},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
tools = _make_mcp_and_tools(config, client)
|
||||
result = await tools["check_exist"](path="/docker,/ghost")
|
||||
|
||||
assert "/docker" in result
|
||||
assert "/ghost" in result
|
||||
assert "Yes" in result
|
||||
assert "No" in result
|
||||
assert "2 path(s) checked" in result
|
||||
|
||||
# Verify DSM was called with both paths as a JSON array
|
||||
call_params = client.request.call_args[1]["params"]
|
||||
requested_paths = json.loads(call_params["path"])
|
||||
assert "/docker" in requested_paths
|
||||
assert "/ghost" in requested_paths
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_exist_empty_path(config: AppConfig) -> None:
|
||||
"""check_exist returns Error: when no path is given."""
|
||||
client = MagicMock()
|
||||
client.request = AsyncMock()
|
||||
|
||||
tools = _make_mcp_and_tools(config, client)
|
||||
result = await tools["check_exist"](path=" ")
|
||||
|
||||
assert result.startswith("Error:")
|
||||
client.request.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_exist_dsm_error(config: AppConfig) -> None:
|
||||
"""check_exist propagates DSM errors as Error: messages."""
|
||||
client = MagicMock()
|
||||
client.request = AsyncMock(side_effect=SynologyError("Permission denied", code=105))
|
||||
|
||||
tools = _make_mcp_and_tools(config, client)
|
||||
result = await tools["check_exist"](path="/docker")
|
||||
|
||||
assert result.startswith("Error:")
|
||||
assert "Permission denied" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_exist_uses_getinfo(config: AppConfig) -> None:
|
||||
"""check_exist uses SYNO.FileStation.List::getinfo as its DSM backend."""
|
||||
client = MagicMock()
|
||||
client.request = AsyncMock(
|
||||
return_value={
|
||||
"files": [
|
||||
{"path": "/docker", "name": "docker", "isdir": True, "additional": {}},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
tools = _make_mcp_and_tools(config, client)
|
||||
await tools["check_exist"](path="/docker")
|
||||
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user