mcp-synology-filestation/test_dirsize_md5.py

"""Wegwerfskript: DirSize + MD5 direkt gegen die NAS testen.

Finale Befunde:
  DirSize: start v2 (path als plain string ODER JSON-Array), status v1 (0ms delay)
  MD5:     start v2, status v1 (0ms delay, ONE-SHOT)

DirSize status-Response Felder:
  finished: bool
  num_dir:  int   (Anzahl Unterordner)
  num_file: int   (Anzahl Dateien)
  total_size: int (Gesamtgroesse in Bytes)

MD5 status-Response Felder:
  finished: bool
  md5:      str   (Hex-String, 32 Zeichen)

Ausfuehren: uv run python test_dirsize_md5.py
"""

import asyncio
import json
import time

import httpx

from mcp_synology_filestation.auth import AuthManager
from mcp_synology_filestation.client import FileStationClient
from mcp_synology_filestation.config import load_config

DIRSIZE_PATH = "/test-mcp"
MD5_PATH = "/test-mcp/test.zip"


def pp(label: str, data: object, elapsed_ms: float | None = None) -> None:
    print(f"\n{'='*60}")
    suffix = f"  [{elapsed_ms:.1f} ms nach start]" if elapsed_ms is not None else ""
    print(f"  {label}{suffix}")
    print("=" * 60)
    print(json.dumps(data, indent=2, ensure_ascii=False))


async def raw(http: httpx.AsyncClient, url: str, sid: str, **params) -> dict:
    r = await http.get(url, params={"_sid": sid, **params})
    r.raise_for_status()
    try:
        return r.json()
    except Exception:
        return {"_raw": r.text[:300], "_http_status": r.status_code}


async def probe_dirsize_path_variants(
    http: httpx.AsyncClient, sid: str, api_url: str
) -> None:
    """Test all three path encoding variants for DirSize start."""
    print(f"\n{'#'*60}")
    print("  DIRSIZE path-Varianten (a=plain, b=json.dumps, c=manuell)")
    print(f"{'#'*60}")

    variants = [
        ("a) plain string", DIRSIZE_PATH),
        ("b) json.dumps([path])", json.dumps([DIRSIZE_PATH])),
        ("c) json.dumps(path)", json.dumps(DIRSIZE_PATH)),
    ]

    for label, path_val in variants:
        print(f"\n--- Variante {label} ---")
        print(f"    path={path_val!r}")

        t0 = time.perf_counter()
        start_body = await raw(
            http, api_url, sid,
            api="SYNO.FileStation.DirSize", version="2", method="start",
            path=path_val,
        )
        pp(f"DirSize::start [{label}]", start_body, (time.perf_counter() - t0) * 1000)

        taskid = (start_body.get("data") or {}).get("taskid")
        if not taskid:
            print("    => Kein taskid (start fehlgeschlagen)")
            continue

        # Poll immediately
        t1 = time.perf_counter()
        status_body = await raw(
            http, api_url, sid,
            api="SYNO.FileStation.DirSize", version="1", method="status",
            taskid=taskid,
        )
        pp(f"DirSize::status [{label}]", status_body, (t1 - t0) * 1000)

        data = (status_body.get("data") or {})
        if data.get("finished"):
            print(f"    => OK: num_dir={data.get('num_dir')} "
                  f"num_file={data.get('num_file')} "
                  f"total_size={data.get('total_size')}")
        else:
            code = (status_body.get("error") or {}).get("code", "?")
            print(f"    => FEHLER code={code}")


async def probe_md5(http: httpx.AsyncClient, sid: str, api_url: str) -> None:
    """Test MD5 with status v1 at 0ms (correct settings)."""
    print(f"\n{'#'*60}")
    print("  MD5 -- start v2, status v1, 0ms delay")
    print(f"{'#'*60}")

    t0 = time.perf_counter()
    start_body = await raw(
        http, api_url, sid,
        api="SYNO.FileStation.MD5", version="2", method="start",
        file_path=json.dumps(MD5_PATH),
    )
    pp("MD5::start", start_body, (time.perf_counter() - t0) * 1000)

    taskid = (start_body.get("data") or {}).get("taskid")
    if not taskid:
        print("[!] No taskid")
        return

    t1 = time.perf_counter()
    r = await raw(
        http, api_url, sid,
        api="SYNO.FileStation.MD5", version="1", method="status",
        taskid=taskid,
    )
    pp("MD5::status v1 [0ms]", r, (t1 - t0) * 1000)
    data = (r.get("data") or {})
    if data.get("finished"):
        print(f"    => OK: md5={data.get('md5')}")
    else:
        print(f"    => FEHLER: {r}")


async def main() -> None:
    config = load_config()
    auth = AuthManager(config)

    async with FileStationClient(config.base_url, config.connection.verify_ssl) as client:
        client.set_auth_manager(auth)
        await client._ensure_initialized()  # noqa: SLF001
        sid = client.sid
        base = config.base_url

        for api_name in ["SYNO.FileStation.DirSize", "SYNO.FileStation.MD5"]:
            info = client._api_cache.get(api_name)  # noqa: SLF001
            if info:
                print(f"[*] {api_name}: path={info['path']} "
                      f"v{info['minVersion']}-v{info['maxVersion']}")
            else:
                print(f"[!] {api_name}: NOT in API cache!")

        dirsize_info = client._api_cache.get("SYNO.FileStation.DirSize", {})  # noqa: SLF001
        md5_info = client._api_cache.get("SYNO.FileStation.MD5", {})  # noqa: SLF001
        dirsize_url = f"{base}/webapi/{dirsize_info.get('path', 'entry.cgi')}"
        md5_url = f"{base}/webapi/{md5_info.get('path', 'entry.cgi')}"

        async with httpx.AsyncClient(verify=config.connection.verify_ssl, timeout=30.0) as http:
            await probe_dirsize_path_variants(http, sid, dirsize_url)
            await probe_md5(http, sid, md5_url)

        await auth.logout(client)
        print("\n[*] Logout OK.")


if __name__ == "__main__":
    asyncio.run(main())