#!/usr/bin/env python3 """Deterministic, no-LLM end-to-end smoke test for the Python<->Unity bridge. This drives a LIVE Unity Editor (booted in CI via ``McpForUnity.Editor.McpCiBoot.StartStdioForCi``) over the same wire path the real MCP server uses -- ``send_command_with_retry`` from ``transport.legacy.unity_connection`` -- and asserts exact response fields. It replaces the LLM agent in ``claude-nl-suite.yml`` with a fixed script so the Python->C# request/response contract is gated on every PR without an API key. Run locally against a running Unity Editor with the MCP bridge active:: cd Server uv run python tests/e2e/bridge_smoke.py Exit codes: 0 all steps passed 1 a step failed an assertion (real bridge regression) 2 no Unity bridge reachable (setup problem, not a contract failure) """ from __future__ import annotations import argparse import os import sys import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable # Make the server's ``src`` importable whether run from repo root or Server/. _SRC = Path(__file__).resolve().parents[2] / "src" if _SRC.is_dir() and str(_SRC) not in sys.path: sys.path.insert(0, str(_SRC)) from transport.legacy.unity_connection import send_command_with_retry # noqa: E402 class BridgeUnavailable(Exception): """Raised when no Unity instance can be reached at all.""" def _ok(resp: Any) -> bool: """True when Unity reports success, tolerant of both response shapes.""" if not isinstance(resp, dict): return False return resp.get("success") is True or resp.get("status") == "success" def _result(resp: Any) -> dict: """Extract the payload dict from either response shape.""" if isinstance(resp, dict): inner = resp.get("result") if isinstance(inner, dict): return inner if isinstance(inner, str): return {"message": inner} return resp return {} def _message(resp: Any) -> str: if not isinstance(resp, dict): return repr(resp) return str(resp.get("message") or resp.get("error") or _result(resp).get("message") or "") def _dig(resp: Any, key: str) -> Any: """Return the first value found for ``key`` anywhere in a nested response. Tolerates the transport wrapping the C# payload under ``result``/``data`` so the contract checks depend on the field, not the exact envelope shape. """ stack: list[Any] = [resp] while stack: cur = stack.pop() if isinstance(cur, dict): if key in cur: return cur[key] stack.extend(cur.values()) elif isinstance(cur, list): stack.extend(cur) return None @dataclass class Step: name: str command: str params: dict # check(resp) -> None; raise AssertionError on failure. check: Callable[[Any], None] retry_on_reload: bool = True @dataclass class StepResult: name: str passed: bool detail: str elapsed_s: float # A unique tag so repeated runs (and parallel CI shards) never collide. _RUN = uuid.uuid4().hex[:8] GO_EMPTY = f"MCP_E2E_Empty_{_RUN}" GO_CUBE = f"MCP_E2E_Cube_{_RUN}" def _assert(cond: bool, msg: str) -> None: if not cond: raise AssertionError(msg) def build_steps() -> list[Step]: """The ordered contract every bridge build must satisfy.""" # Shared across steps so the find/cleanup checks can assert against the exact # object the create step made. find_gameobjects returns instance IDs only # (never names), so a name-substring check would never match. created: dict[str, int] = {} def check_console(resp: Any) -> None: _assert(_ok(resp), f"read_console did not succeed: {_message(resp)}") def check_create_empty(resp: Any) -> None: _assert(_ok(resp), f"create empty GameObject failed: {_message(resp)}") gid = _dig(resp, "instanceID") _assert(isinstance(gid, int), f"create did not return an int instanceID: {_message(resp)}") created["empty"] = gid def check_found(resp: Any) -> None: _assert(_ok(resp), f"find_gameobjects failed: {_message(resp)}") ids = _dig(resp, "instanceIDs") or [] _assert( created.get("empty") in ids, f"created object '{GO_EMPTY}' (id={created.get('empty')}) not in find results {ids}", ) def check_create_cube(resp: Any) -> None: _assert(_ok(resp), f"create primitive Cube failed: {_message(resp)}") def check_delete(resp: Any) -> None: _assert(_ok(resp), f"delete failed: {_message(resp)}") def check_gone(resp: Any) -> None: _assert(_ok(resp), f"post-delete find_gameobjects failed: {_message(resp)}") ids = _dig(resp, "instanceIDs") or [] _assert( created.get("empty") not in ids, f"'{GO_EMPTY}' (id={created.get('empty')}) still present after delete -- cleanup did not take effect", ) return [ Step("read_console_baseline", "read_console", {"action": "get", "count": "5", "include_stacktrace": False}, check_console), Step("create_empty_gameobject", "manage_gameobject", {"action": "create", "name": GO_EMPTY}, check_create_empty), Step("find_created_gameobject", "find_gameobjects", {"searchMethod": "by_name", "searchTerm": GO_EMPTY}, check_found), Step("create_primitive_with_component", "manage_gameobject", {"action": "create", "name": GO_CUBE, "primitiveType": "Cube", "componentsToAdd": ["Rigidbody"]}, check_create_cube), Step("delete_cube", "manage_gameobject", {"action": "delete", "target": GO_CUBE, "searchMethod": "by_name"}, check_delete), Step("delete_empty", "manage_gameobject", {"action": "delete", "target": GO_EMPTY, "searchMethod": "by_name"}, check_delete), Step("verify_cleanup", "find_gameobjects", {"searchMethod": "by_name", "searchTerm": GO_EMPTY}, check_gone), ] def run(instance_id: str | None, max_retries: int, retry_ms: int) -> list[StepResult]: results: list[StepResult] = [] for step in build_steps(): t0 = time.time() try: resp = send_command_with_retry( step.command, step.params, instance_id=instance_id, max_retries=max_retries, retry_ms=retry_ms, retry_on_reload=step.retry_on_reload, ) except Exception as exc: # connection refused, timeout, etc. elapsed = time.time() - t0 # A failure on the very first call usually means no bridge at all. if not results: raise BridgeUnavailable(str(exc)) from exc results.append(StepResult(step.name, False, f"transport error: {exc}", elapsed)) break elapsed = time.time() - t0 try: step.check(resp) results.append(StepResult(step.name, True, "ok", elapsed)) except AssertionError as err: results.append(StepResult(step.name, False, str(err), elapsed)) # Keep going so cleanup deletes still run, but record the failure. return results def write_junit(path: Path, results: list[StepResult]) -> None: import xml.sax.saxutils as sx failures = sum(0 if r.passed else 1 for r in results) total_time = sum(r.elapsed_s for r in results) cases = [] for r in results: body = "" if r.passed else f"{sx.escape(r.detail)}" cases.append( f' {body}' ) xml = ( '\n' f'\n' + "\n".join(cases) + "\n\n" ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(xml, encoding="utf-8") def main() -> int: ap = argparse.ArgumentParser(description="Deterministic Unity bridge E2E smoke test") ap.add_argument("--instance", default=os.environ.get("UNITY_MCP_DEFAULT_INSTANCE") or None, help="Unity instance id (name, hash, name@hash). Default: env or auto-discover.") ap.add_argument("--max-retries", type=int, default=8, help="Reload retries per command") ap.add_argument("--retry-ms", type=int, default=250, help="Delay between reload retries (ms)") ap.add_argument("--junit", default=os.environ.get("E2E_JUNIT_OUT"), help="Optional path to write a JUnit XML report") args = ap.parse_args() instance = args.instance.strip() if isinstance(args.instance, str) else None print(f"== Unity bridge E2E smoke (run={_RUN}, instance={instance or 'auto'}) ==", flush=True) try: results = run(instance, args.max_retries, args.retry_ms) except BridgeUnavailable as exc: print(f"::error::No Unity bridge reachable: {exc}", flush=True) print("Is a Unity Editor running with the MCP bridge active? " "(set UNITY_MCP_STATUS_DIR / UNITY_MCP_DEFAULT_INSTANCE for CI)", flush=True) return 2 if args.junit: write_junit(Path(args.junit), results) failed = [r for r in results if not r.passed] for r in results: status = "PASS" if r.passed else "FAIL" print(f" [{status}] {r.name} ({r.elapsed_s:.2f}s){'' if r.passed else ' -- ' + r.detail}", flush=True) print(f"== {len(results) - len(failed)}/{len(results)} passed ==", flush=True) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())