[claude] Bannerlord VM setup guide + GABS connectivity test (#1098) (#1116)

scripts/test_gabs_connectivity.py

@@ -0,0 +1,244 @@
+#!/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())