juc500/scripts/verify_cable_switch.py

#!/usr/bin/env python3
"""Verify JUC500 GUI connect + listdir via local API."""
from __future__ import annotations

import json
import sys
import time
import urllib.error
import urllib.request

GUI = "http://127.0.0.1:8765"
POLL = 1.0
CONNECT_DEADLINE = 180.0


def api(path: str, *, method: str = "GET", body: dict | None = None, timeout: float = 15.0):
    data = None
    headers = {}
    if body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(f"{GUI}{path}", data=data, method=method, headers=headers)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())


def status() -> dict:
    return api("/api/status", timeout=8.0)


def wait_connected(deadline: float, *, label: str) -> dict:
    end = time.monotonic() + deadline
    last: dict = {}
    started = time.monotonic()
    last_retry = started
    while time.monotonic() < end:
        now = time.monotonic()
        try:
            last = status()
        except Exception as exc:
            print(f"[{label}] status error: {exc}", flush=True)
            time.sleep(POLL)
            continue
        conn = bool(last.get("connected"))
        model = last.get("cable_model")
        connecting = bool(last.get("connecting"))
        err = last.get("connect_error")
        print(
            f"[{label}] connected={conn} model={model} connecting={connecting} err={err!r}",
            flush=True,
        )
        if conn:
            return last
        if err and not connecting:
            retryable = any(
                k in (err or "")
                for k in (
                    "session stopped",
                    "연결 시간 초과",
                    "연결되지 않았습니다",
                    "Access denied",
                    "점유 실패",
                )
            )
            if retryable and now - last_retry >= 5.0:
                try:
                    api("/api/disconnect", method="POST", body={})
                    time.sleep(1.0)
                    api("/api/connect/auto", method="POST", body={})
                    print(f"[{label}] retry after: {err}", flush=True)
                except Exception as exc:
                    print(f"[{label}] retry error: {exc}", flush=True)
                last_retry = now
                time.sleep(POLL)
                continue
            raise RuntimeError(f"{label} failed: {err}")
        if now - last_retry >= 70.0 and (connecting or now - started >= 25.0):
            try:
                api("/api/disconnect", method="POST", body={})
                time.sleep(1.0)
                api("/api/connect/auto", method="POST", body={})
                print(f"[{label}] retry disconnect+connect/auto", flush=True)
            except Exception as exc:
                print(f"[{label}] retry error: {exc}", flush=True)
            last_retry = now
        time.sleep(POLL)
    raise TimeoutError(f"{label}: not connected within {deadline:.0f}s — last={last}")


def verify_listdir(label: str) -> int:
    try:
        data = api("/api/listdir?path=__roots__", timeout=30.0)
    except urllib.error.HTTPError as exc:
        body = exc.read().decode(errors="replace")
        raise RuntimeError(f"{label} listdir HTTP {exc.code}: {body}") from exc
    n = len(data.get("entries") or [])
    print(f"[{label}] listdir roots={n}", flush=True)
    if n <= 0:
        raise RuntimeError(f"{label} listdir returned no entries")
    return n


def assert_juc500(s: dict, label: str) -> None:
    active = (s.get("cable_model") or "").upper()
    dev = s.get("active_cable") or s.get("device") or {}
    pid = (dev.get("pid") or "").lower()
    if active != "JUC500":
        raise RuntimeError(f"{label}: cable_model={active!r} expected JUC500")
    if pid and pid != "7500":
        raise RuntimeError(f"{label}: unexpected pid={pid} for JUC500")
    print(
        f"[{label}] OK model={active} pid={pid or '7500'} host={s.get('hostname')}",
        flush=True,
    )


def main() -> int:
    print("JUC500 connect verification", flush=True)
    try:
        s0 = status()
    except Exception as exc:
        print(f"GUI not reachable at {GUI}: {exc}", file=sys.stderr)
        return 1

    devices = s0.get("devices") or []
    print(f"Detected devices: {len(devices)}", flush=True)
    if not devices:
        print("WARN: no JUC500 detected — plug cable and retry", flush=True)

    if not s0.get("connected"):
        print("Waiting for autoconnect on JUC500…", flush=True)

    s = wait_connected(CONNECT_DEADLINE, label="JUC500")
    assert_juc500(s, "JUC500-connected")
    verify_listdir("JUC500")

    try:
        ping = api("/api/peer/ping", timeout=15.0)
        print(f"[JUC500] peer ping: {ping}", flush=True)
    except Exception as exc:
        print(f"[JUC500] peer ping skipped: {exc}", flush=True)

    print("\n=== ALL CHECKS PASSED ===", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())