chobit/tests/test_gesture_system.py

"""Integration tests for the gesture system via UDP.

Requires Chobit running (`./run start`). Tests the full stack:
  BoneRegistry → GestureRegistry → IdleAnimator → Skeleton3D

Run:  python tests/test_gesture_system.py
"""

from __future__ import annotations

import json
import socket
import sys
import time

GODOT_PORT = 19700
TIMEOUT = 2.0


def send(cmd: str, **kwargs: object) -> dict | None:
    msg = {"cmd": cmd, **kwargs}
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(TIMEOUT)
    try:
        sock.sendto(json.dumps(msg).encode(), ("127.0.0.1", GODOT_PORT))
        data, _ = sock.recvfrom(8192)
        return json.loads(data.decode())
    except (OSError, json.JSONDecodeError) as e:
        return {"error": str(e)}
    finally:
        sock.close()


def assert_eq(label: str, actual: object, expected: object) -> None:
    if actual == expected:
        print(f"  PASS  {label}")
    else:
        print(f"  FAIL  {label}: expected {expected!r}, got {actual!r}")
        raise AssertionError(f"{label} failed")


def assert_in(label: str, item: object, collection: object) -> None:
    if item in collection:
        print(f"  PASS  {label}")
    else:
        print(f"  FAIL  {label}: {item!r} not in {collection!r}")
        raise AssertionError(f"{label} failed")


def assert_not_in(label: str, item: object, collection: object) -> None:
    if item not in collection:
        print(f"  PASS  {label}")
    else:
        print(f"  FAIL  {label}: {item!r} unexpectedly found in {collection!r}")
        raise AssertionError(f"{label} failed")


def assert_ok(label: str, result: dict | None) -> dict:
    if result is None:
        print(f"  FAIL  {label}: no response (is Chobit running?)")
        raise AssertionError(f"{label}: no response")
    if "error" in result:
        print(f"  FAIL  {label}: {result['error']}")
        raise AssertionError(f"{label}: {result['error']}")
    print(f"  PASS  {label}")
    return result


class AssertionError(Exception):
    pass


def test_status() -> None:
    print("\n── test_status ──")
    result = send("status")
    r = assert_ok("status responds", result)
    assert_eq("is running", r.get("running"), True)


def test_list_animations() -> None:
    print("\n── test_list_animations ──")
    result = send("list_animations")
    r = assert_ok("list_animations responds", result)
    gestures = r.get("gestures", [])
    assert_in("wave in gestures", "wave", gestures)
    assert_in("stretch in gestures", "stretch", gestures)
    assert_in("settle in gestures", "settle", gestures)
    assert_in("curious in gestures", "curious", gestures)
    assert_in("sigh in gestures", "sigh", gestures)
    assert_not_in("fart_wave removed", "fart_wave", gestures)
    assert_in("slow_blink in gestures", "slow_blink", gestures)


def test_list_bones() -> None:
    print("\n── test_list_bones ──")
    result = send("list_bones", filter="Right")
    r = assert_ok("list_bones responds", result)
    bones = r.get("bones", [])
    bone_names = [b["name"] for b in bones]
    # VRM humanoid bones should be present
    for expected in ["RightUpperArm", "RightLowerArm", "RightHand"]:
        found = any(expected.lower() in n.lower() for n in bone_names)
        if found:
            print(f"  PASS  bone '{expected}' found (possibly different casing)")
        else:
            print(f"  INFO  bone '{expected}' not found by name — may use Japanese names")
    print(f"  INFO  {len(bones)} right-side bones found")


def test_play_gesture() -> None:
    print("\n── test_play_gesture ──")
    result = send("play_animation", name="wave")
    r = assert_ok("play_animation wave", result)
    assert_eq("played wave", r.get("played"), "wave")


def test_test_pose() -> None:
    print("\n── test_test_pose ──")
    result = send(
        "test_pose",
        bones={"RightUpperArm": [20, 0, -80], "RightLowerArm": [0, -100, 0]},
        duration=1.0,
        oscillations=[{"bone": "RightHand", "axis": [0, 1, 0], "freq": 3.0, "amp_deg": 30.0}],
    )
    r = assert_ok("test_pose responds", result)
    assert_eq("testing has bones", "RightUpperArm" in r.get("testing", {}), True)
    # Wait for gesture to finish
    time.sleep(1.5)


def test_test_pose_novel_bone() -> None:
    print("\n── test_test_pose_novel_bone ──")
    # Register a bone not in any gesture def — dynamic registration
    result = send(
        "test_pose",
        bones={"LeftUpperArm": [20, 0, 80]},
        duration=1.0,
    )
    r = assert_ok("test_pose novel bone", result)
    assert_eq("testing has LeftUpperArm", "LeftUpperArm" in r.get("testing", {}), True)
    time.sleep(1.5)


def test_debug_bones() -> None:
    print("\n── test_debug_bones ──")
    result = send("debug_bones")
    r = assert_ok("debug_bones responds", result)
    bones = r.get("bones", {})
    assert_in("RightUpperArm in debug", "RightUpperArm", bones)


def main() -> int:
    tests = [
        test_status,
        test_list_animations,
        test_list_bones,
        test_debug_bones,
        test_play_gesture,
        test_test_pose,
        test_test_pose_novel_bone,
    ]
    passed = 0
    failed = 0
    for test in tests:
        try:
            test()
            passed += 1
        except (AssertionError, Exception) as e:
            failed += 1
            print(f"  ERROR {e}")

    print(f"\n{'=' * 40}")
    print(f"Results: {passed} passed, {failed} failed")
    return 1 if failed > 0 else 0


if __name__ == "__main__":
    sys.exit(main())