juc500/dist-offline/juc500-offline-update-20260720-174241/scripts/verify_cable_switch.py

#!/usr/bin/env python3
"""Verify JUC500 connect → JUC700 switch via local GUI 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
SWITCH_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:
        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") or last.get("selected_cable_model")
        sel = last.get("selected_cable_model")
        connecting = bool(last.get("connecting"))
        err = last.get("connect_error")
        print(
            f"[{label}] connected={conn} model={model} selected={sel} "
            f"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}")
        now = time.monotonic()
        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 select_cable(model: str) -> dict:
    print(f"\n=== SELECT {model} ===", flush=True)
    return api("/api/cable/select", method="POST", body={"model": model}, timeout=20.0)


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


def main() -> int:
    print("Cable switch 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 []
    models = {d.get("model") for d in devices}
    print(f"Detected devices: {models}", flush=True)
    if "JUC500" not in models or "JUC700" not in models:
        print("WARN: both JUC500 and JUC700 should be plugged in for full test", flush=True)

    sel0 = (s0.get("selected_cable_model") or "JUC500").upper()
    if sel0 != "JUC500":
        select_cable("JUC500")
    elif not s0.get("connected"):
        print("Waiting for autoconnect on JUC500…", flush=True)

    # Step 1: JUC500
    s500 = wait_connected(CONNECT_DEADLINE, label="JUC500")
    assert_model(s500, "JUC500", "JUC500-connected")
    verify_listdir("JUC500")

    # Step 2: switch to JUC700
    select_cable("JUC700")
    s700 = wait_connected(SWITCH_DEADLINE, label="JUC700")
    assert_model(s700, "JUC700", "JUC700-connected")
    verify_listdir("JUC700")

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

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


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