Timmy-time-dashboard/scripts/test_gabs_connectivity.py

#!/usr/bin/env python3
"""GABS TCP connectivity and JSON-RPC smoke test.

Tests connectivity from Hermes to the Bannerlord.GABS TCP server running on the
Windows VM. Covers:
  1. TCP socket connection (port 4825 reachable)
  2. JSON-RPC ping round-trip
  3. get_game_state call (game must be running)
  4. Latency — target < 100 ms on LAN

Usage:
    python scripts/test_gabs_connectivity.py --host 10.0.0.50
    python scripts/test_gabs_connectivity.py --host 10.0.0.50 --port 4825 --timeout 5

Refs: #1098 (Bannerlord Infra — Windows VM Setup + GABS Mod Installation)
Epic: #1091 (Project Bannerlord)
"""

from __future__ import annotations

import argparse
import json
import socket
import sys
import time
from typing import Any

DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 4825
DEFAULT_TIMEOUT = 5  # seconds
LATENCY_TARGET_MS = 100.0


# ── Low-level TCP helpers ─────────────────────────────────────────────────────


def _tcp_connect(host: str, port: int, timeout: float) -> socket.socket:
    """Open a TCP connection and return the socket. Raises on failure."""
    sock = socket.create_connection((host, port), timeout=timeout)
    sock.settimeout(timeout)
    return sock


def _send_recv(sock: socket.socket, payload: dict[str, Any]) -> dict[str, Any]:
    """Send a newline-delimited JSON-RPC request and return the parsed response."""
    raw = json.dumps(payload) + "\n"
    sock.sendall(raw.encode())

    buf = b""
    while b"\n" not in buf:
        chunk = sock.recv(4096)
        if not chunk:
            raise ConnectionError("Connection closed before response received")
        buf += chunk

    line = buf.split(b"\n", 1)[0]
    return json.loads(line.decode())


def _rpc(sock: socket.socket, method: str, params: dict | None = None, req_id: int = 1) -> dict[str, Any]:
    """Build and send a JSON-RPC 2.0 request, return the response dict."""
    payload: dict[str, Any] = {
        "jsonrpc": "2.0",
        "method": method,
        "id": req_id,
    }
    if params:
        payload["params"] = params
    return _send_recv(sock, payload)


# ── Test cases ────────────────────────────────────────────────────────────────


def test_tcp_connection(host: str, port: int, timeout: float) -> tuple[bool, socket.socket | None]:
    """PASS: TCP connection to host:port succeeds."""
    print(f"\n[1/4] TCP connection → {host}:{port}")
    try:
        t0 = time.monotonic()
        sock = _tcp_connect(host, port, timeout)
        elapsed_ms = (time.monotonic() - t0) * 1000
        print(f"  ✓ Connected ({elapsed_ms:.1f} ms)")
        return True, sock
    except OSError as exc:
        print(f"  ✗ Connection failed: {exc}")
        print(f"  Checklist:")
        print(f"    - Is Bannerlord running with GABS mod enabled?")
        print(f"    - Is port {port} open in Windows Firewall?")
        print(f"    - Is the VM IP correct? (got: {host})")
        return False, None


def test_ping(sock: socket.socket) -> bool:
    """PASS: JSON-RPC ping returns a 2.0 response."""
    print(f"\n[2/4] JSON-RPC ping")
    try:
        t0 = time.monotonic()
        resp = _rpc(sock, "ping", req_id=1)
        elapsed_ms = (time.monotonic() - t0) * 1000
        if resp.get("jsonrpc") == "2.0" and "error" not in resp:
            print(f"  ✓ Ping OK ({elapsed_ms:.1f} ms): {json.dumps(resp)}")
            return True
        print(f"  ✗ Unexpected response ({elapsed_ms:.1f} ms): {json.dumps(resp)}")
        return False
    except Exception as exc:
        print(f"  ✗ Ping failed: {exc}")
        return False


def test_game_state(sock: socket.socket) -> bool:
    """PASS: get_game_state returns a result (game must be in a campaign)."""
    print(f"\n[3/4] get_game_state call")
    try:
        t0 = time.monotonic()
        resp = _rpc(sock, "get_game_state", req_id=2)
        elapsed_ms = (time.monotonic() - t0) * 1000
        if "error" in resp:
            code = resp["error"].get("code", "?")
            msg = resp["error"].get("message", "")
            if code == -32601:
                # Method not found — GABS version may not expose this method
                print(f"  ~ Method not available ({elapsed_ms:.1f} ms): {msg}")
                print(f"    This is acceptable if game is not yet in a campaign.")
                return True
            print(f"  ✗ RPC error ({elapsed_ms:.1f} ms) [{code}]: {msg}")
            return False
        result = resp.get("result", {})
        print(f"  ✓ Game state received ({elapsed_ms:.1f} ms):")
        for k, v in result.items():
            print(f"    {k}: {v}")
        return True
    except Exception as exc:
        print(f"  ✗ get_game_state failed: {exc}")
        return False


def test_latency(host: str, port: int, timeout: float, iterations: int = 5) -> bool:
    """PASS: Average round-trip latency is under LATENCY_TARGET_MS."""
    print(f"\n[4/4] Latency test ({iterations} pings, target < {LATENCY_TARGET_MS:.0f} ms)")
    try:
        times: list[float] = []
        for i in range(iterations):
            sock = _tcp_connect(host, port, timeout)
            try:
                t0 = time.monotonic()
                _rpc(sock, "ping", req_id=i + 10)
                times.append((time.monotonic() - t0) * 1000)
            finally:
                sock.close()

        avg_ms = sum(times) / len(times)
        min_ms = min(times)
        max_ms = max(times)
        print(f"  avg={avg_ms:.1f} ms  min={min_ms:.1f} ms  max={max_ms:.1f} ms")

        if avg_ms <= LATENCY_TARGET_MS:
            print(f"  ✓ Latency within target ({avg_ms:.1f} ms ≤ {LATENCY_TARGET_MS:.0f} ms)")
            return True
        print(
            f"  ✗ Latency too high ({avg_ms:.1f} ms > {LATENCY_TARGET_MS:.0f} ms)\n"
            f"    Check network path between Hermes and the VM."
        )
        return False
    except Exception as exc:
        print(f"  ✗ Latency test failed: {exc}")
        return False


# ── Main ──────────────────────────────────────────────────────────────────────


def main() -> int:
    parser = argparse.ArgumentParser(description="GABS TCP connectivity smoke test")
    parser.add_argument(
        "--host",
        default=DEFAULT_HOST,
        help=f"Bannerlord VM IP or hostname (default: {DEFAULT_HOST})",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help=f"GABS TCP port (default: {DEFAULT_PORT})",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_TIMEOUT,
        help=f"Socket timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    args = parser.parse_args()

    print("=" * 60)
    print(f"GABS Connectivity Test Suite")
    print(f"Target: {args.host}:{args.port}")
    print(f"Timeout: {args.timeout}s")
    print("=" * 60)

    results: dict[str, bool] = {}

    # Test 1: TCP connection (gate — skip remaining if unreachable)
    ok, sock = test_tcp_connection(args.host, args.port, args.timeout)
    results["tcp_connection"] = ok
    if not ok:
        _print_summary(results)
        return 1

    # Tests 2–3 reuse the same socket
    try:
        results["ping"] = test_ping(sock)
        results["game_state"] = test_game_state(sock)
    finally:
        sock.close()

    # Test 4: latency uses fresh connections
    results["latency"] = test_latency(args.host, args.port, args.timeout)

    return _print_summary(results)


def _print_summary(results: dict[str, bool]) -> int:
    passed = sum(results.values())
    total = len(results)
    print("\n" + "=" * 60)
    print(f"Results: {passed}/{total} passed")
    print("=" * 60)
    for name, ok in results.items():
        icon = "✓" if ok else "✗"
        print(f"  {icon} {name}")

    if passed == total:
        print("\n✓ GABS connectivity verified. Timmy can reach the game.")
        print("  Next step: run benchmark level 0 (JSON compliance check).")
    elif not results.get("tcp_connection"):
        print("\n✗ TCP connection failed. VM/firewall setup incomplete.")
        print("  See docs/research/bannerlord-vm-setup.md for checklist.")
    else:
        print("\n~ Partial pass — review failures above.")

    return 0 if passed == total else 1


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