marcus
4c093589c1
test_additional.py tests SYNO.FileStation.List::list with three variants: no additional, comma-separated string, and JSON array — bypasses client error mapping to log the raw DSM response for each. Delete after use. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
"""Throwaway script: test SYNO.FileStation.List::list additional field formats.
|
|
|
|
Run with:
|
|
uv run python test_additional.py
|
|
|
|
Reads config from the default location and credentials from the OS keyring,
|
|
exactly as the MCP server does. Tests three additional variants against a
|
|
real share path and prints the raw DSM response for each.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
|
|
import httpx # noqa: F401 — used at runtime for the HTTP client reference
|
|
|
|
# ── config ────────────────────────────────────────────────────────────────
|
|
|
|
# Change this to any share that exists on your NAS (as returned by list_shares)
|
|
TEST_PATH = "/dev"
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def run() -> None:
|
|
from mcp_synology_filestation.auth import AuthManager
|
|
from mcp_synology_filestation.client import FileStationClient
|
|
from mcp_synology_filestation.config import load_config
|
|
|
|
config = load_config()
|
|
auth = AuthManager(config)
|
|
|
|
async with FileStationClient(
|
|
config.base_url,
|
|
config.connection.verify_ssl,
|
|
config.connection.timeout,
|
|
) as client:
|
|
await client.query_api_info()
|
|
sid = await auth.login(client)
|
|
client.sid = sid
|
|
|
|
info = client._api_cache.get("SYNO.FileStation.List", {})
|
|
print(
|
|
f"SYNO.FileStation.List — path={info.get('path')!r} "
|
|
f"v{info.get('minVersion')}-v{info.get('maxVersion')}\n"
|
|
)
|
|
|
|
variants: list[tuple[str, dict]] = [
|
|
(
|
|
"1 — no additional",
|
|
{
|
|
"folder_path": TEST_PATH,
|
|
"offset": 0,
|
|
"limit": 5,
|
|
"sort_by": "name",
|
|
"sort_direction": "asc",
|
|
},
|
|
),
|
|
(
|
|
'2 — additional="size,time" (comma-separated string)',
|
|
{
|
|
"folder_path": TEST_PATH,
|
|
"offset": 0,
|
|
"limit": 5,
|
|
"sort_by": "name",
|
|
"sort_direction": "asc",
|
|
"additional": "size,time",
|
|
},
|
|
),
|
|
(
|
|
'3 — additional=json.dumps(["size","time"]) (JSON array)',
|
|
{
|
|
"folder_path": TEST_PATH,
|
|
"offset": 0,
|
|
"limit": 5,
|
|
"sort_by": "name",
|
|
"sort_direction": "asc",
|
|
"additional": json.dumps(["size", "time"]),
|
|
},
|
|
),
|
|
]
|
|
|
|
for label, params in variants:
|
|
print(f"{'─' * 60}")
|
|
print(f"Variant {label}")
|
|
print(f" params: {params}")
|
|
try:
|
|
# Call the low-level HTTP layer directly so we see the raw body,
|
|
# bypassing any error-mapping in client.request()
|
|
api = "SYNO.FileStation.List"
|
|
method = "list"
|
|
api_info = client._api_cache[api]
|
|
version = api_info["maxVersion"]
|
|
url = f"{client._base_url}/webapi/{api_info['path']}"
|
|
|
|
req_params: dict = {
|
|
"api": api,
|
|
"version": str(version),
|
|
"method": method,
|
|
"_sid": client.sid,
|
|
}
|
|
req_params.update(params)
|
|
|
|
http = client._http
|
|
resp = await http.get(url, params=req_params)
|
|
body = resp.json()
|
|
print(f" HTTP {resp.status_code}")
|
|
print(f" success: {body.get('success')}")
|
|
if body.get("success"):
|
|
data = body.get("data", {})
|
|
files = data.get("files", [])
|
|
print(f" total: {data.get('total')} returned: {len(files)}")
|
|
if files:
|
|
first = files[0]
|
|
print(f" first file keys: {list(first.keys())}")
|
|
add = first.get("additional")
|
|
print(f" first file additional: {add!r}")
|
|
else:
|
|
print(f" RAW error body: {body!r}")
|
|
except Exception as exc:
|
|
print(f" EXCEPTION: {exc!r}")
|
|
print()
|
|
|
|
await auth.logout(client)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(run())
|
|
except Exception as e:
|
|
print(f"Fatal: {e}", file=sys.stderr)
|
|
sys.exit(1)
|