From 54412931b44aba3528b20902e4240eee6eef8382 Mon Sep 17 00:00:00 2001 From: air17 Date: Fri, 12 Jun 2026 17:50:56 +0000 Subject: [PATCH 1/3] Refine nexus bridge typing --- Editor/nexus_bridge/_transport.py | 13 +- Editor/nexus_bridge/_types.py | 76 +++++++++ Editor/nexus_bridge/routing.py | 252 +++++++++++++++++++----------- Editor/nexus_bridge/schemas.py | 8 +- 4 files changed, 250 insertions(+), 99 deletions(-) create mode 100644 Editor/nexus_bridge/_types.py diff --git a/Editor/nexus_bridge/_transport.py b/Editor/nexus_bridge/_transport.py index b9fabdf..efc7cb2 100644 --- a/Editor/nexus_bridge/_transport.py +++ b/Editor/nexus_bridge/_transport.py @@ -5,7 +5,8 @@ import os import sys import urllib.request -from typing import Any + +from ._types import JsonObject, JsonRpcError, JsonRpcRequest, JsonRpcResponse DEFAULT_PORT: int = 8081 @@ -44,8 +45,8 @@ def _read_timeout() -> float: UNITY_TIMEOUT_SECONDS: float = _read_timeout() -def call_unity(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]: - payload: dict[str, Any] = {"jsonrpc": "2.0", "method": method, "params": params or {}, "id": 1} +def call_unity(method: str, params: JsonObject | None = None) -> JsonRpcResponse: + payload: JsonRpcRequest = {"jsonrpc": "2.0", "method": method, "params": params or {}, "id": 1} data: bytes = json.dumps(payload).encode("utf-8") req: urllib.request.Request = urllib.request.Request( UNITY_URL, @@ -56,4 +57,8 @@ def call_unity(method: str, params: dict[str, Any] | None = None) -> dict[str, A with urllib.request.urlopen(req, timeout=UNITY_TIMEOUT_SECONDS) as response: return json.loads(response.read().decode("utf-8")) except Exception as error: - return {"error": {"code": -32000, "message": f"Unity Server unreachable. Error: {error}"}} + error_payload: JsonRpcError = { + "code": -32000, + "message": f"Unity Server unreachable. Error: {error}", + } + return {"error": error_payload} diff --git a/Editor/nexus_bridge/_types.py b/Editor/nexus_bridge/_types.py new file mode 100644 index 0000000..bae9566 --- /dev/null +++ b/Editor/nexus_bridge/_types.py @@ -0,0 +1,76 @@ +"""Private type definitions for the NexusUnity Python bridge.""" +from __future__ import annotations + +from typing import Any, TypeAlias, TypedDict + +JsonObject: TypeAlias = dict[str, Any] + + +class JsonRpcError(TypedDict): + code: int + message: str + + +class JsonRpcRequest(TypedDict): + jsonrpc: str + method: str + params: JsonObject + id: int + + +class ToolDefinition(TypedDict): + name: str + description: str + inputSchema: JsonObject + + +class ResourceDefinition(TypedDict): + uri: str + name: str + mimeType: str + + +class JsonRpcResponse(TypedDict, total=False): + result: JsonObject + error: JsonRpcError + + +class TransformArguments(TypedDict, total=False): + instance_id: int + position: JsonObject + rotation: JsonObject + scale: JsonObject + eulerAngles: JsonObject + localScale: JsonObject + + +class WriteFileSpec(TypedDict): + path: str + content: str + + +class WriteError(TypedDict): + path: str + error: JsonRpcError + + +class WaitResultPayload(TypedDict): + status: str + time_waited_seconds: float + + +class TestResultsPayload(WaitResultPayload, total=False): + timestamp_utc: str + message: str + result_path: str + trigger: JsonObject + + +class WriteAndCompileSuccessPayload(WaitResultPayload): + compiler_errors: list[JsonObject] + + +class WriteAndCompileFailurePayload(TypedDict): + status: str + message: str + errors: list[WriteError] diff --git a/Editor/nexus_bridge/routing.py b/Editor/nexus_bridge/routing.py index aaddea6..e9db828 100644 --- a/Editor/nexus_bridge/routing.py +++ b/Editor/nexus_bridge/routing.py @@ -7,106 +7,149 @@ from __future__ import annotations import time -from typing import Any +from typing import Any, Mapping, Sequence, cast from ._logging import logger from ._transport import call_unity from .schemas import STATIC_TOOLS - - -def _compact(params: dict[str, Any]) -> dict[str, Any]: +from ._types import ( + JsonObject, + JsonRpcError, + JsonRpcResponse, + TestResultsPayload, + TransformArguments, + WaitResultPayload, + WriteAndCompileFailurePayload, + WriteAndCompileSuccessPayload, + WriteError, + WriteFileSpec, +) + + +def _compact(params: JsonObject) -> JsonObject: return {key: value for key, value in params.items() if value is not None} -def _alias(action: str | None, aliases: dict[str, str]) -> str | None: - return aliases.get(action, action) # type: ignore[arg-type] +def _alias(action_name: str | None, aliases: Mapping[str, str]) -> str | None: + if action_name is None: + return None + return aliases.get(action_name, action_name) -def _invalid_action(action: str | None, valid_actions: list[str]) -> dict[str, Any]: +def _invalid_action(action_name: str | None, valid_actions: Sequence[str]) -> JsonRpcResponse: valid = ", ".join(valid_actions) - return {"error": {"code": -32602, "message": f"Invalid action: {action}. Valid actions: {valid}"}} + error_payload: JsonRpcError = { + "code": -32602, + "message": f"Invalid action: {action_name}. Valid actions: {valid}", + } + return {"error": error_payload} + + +def _result_object(response: JsonRpcResponse | None) -> JsonObject: + if not response: + return {} + result_payload = response.get("result") + return result_payload if isinstance(result_payload, dict) else {} -def _transform_params(args: dict[str, Any], instance_id: int | None = None) -> dict[str, Any]: - params: dict[str, Any] = {"instance_id": instance_id if instance_id is not None else args.get("instance_id")} +def _error_object(response: JsonRpcResponse | None) -> JsonRpcError | None: + if not response: + return None + return response.get("error") + + +def _transform_params(args: JsonObject, instance_id: int | None = None) -> JsonObject: + params: TransformArguments = { + "instance_id": instance_id if instance_id is not None else args.get("instance_id") + } for key in ["position", "rotation", "scale", "eulerAngles", "localScale"]: params[key] = args.get(key) return _compact(params) -def _extract_created_instance_id(response: dict[str, Any]) -> int | None: - if not isinstance(response, dict) or "error" in response: +def _extract_created_instance_id(response: JsonRpcResponse) -> int | None: + if "error" in response: return None - result = response.get("result", {}) - data = result.get("data", {}) if isinstance(result, dict) else {} + result_payload = _result_object(response) + data = result_payload.get("data", {}) return data.get("instance_id") if isinstance(data, dict) else None -def _apply_created_transform(response: dict[str, Any], args: dict[str, Any]) -> dict[str, Any]: +def _apply_created_transform(response: JsonRpcResponse, args: JsonObject) -> JsonRpcResponse: instance_id = _extract_created_instance_id(response) if not instance_id: return response params = _transform_params(args, instance_id) if len(params) <= 1: return response - transform = call_unity("set_transform", params) - if transform and "error" in transform: - return transform + transform_response = call_unity("set_transform", params) + if transform_response and "error" in transform_response: + return transform_response return response -def _run_tests_wait(args: dict[str, Any]) -> dict[str, Any]: +def _run_tests_wait(args: JsonObject) -> JsonRpcResponse: timeout = args.get("timeout_seconds", 180) poll_interval = args.get("poll_interval_seconds", 1.0) start_time = time.time() - before = call_unity("get_test_results") - before_result = before.get("result", {}) if isinstance(before, dict) else {} - before_timestamp = before_result.get("timestamp_utc") if before_result.get("status") == "Success" else None + previous_results_response = call_unity("get_test_results") + previous_results_payload = _result_object(previous_results_response) + previous_timestamp = ( + previous_results_payload.get("timestamp_utc") + if previous_results_payload.get("status") == "Success" + else None + ) run_params = _compact({ "mode": args.get("mode", "EditMode"), "filter": args.get("filter"), }) - trigger = call_unity("run_tests", run_params) - if trigger and "error" in trigger: - return trigger + trigger_response = call_unity("run_tests", run_params) + if trigger_response and "error" in trigger_response: + return trigger_response - trigger_result = trigger.get("result", {}) if isinstance(trigger, dict) else {} - result_path = trigger_result.get("result_path") + trigger_payload = _result_object(trigger_response) + result_path = trigger_payload.get("result_path") while time.time() - start_time < timeout: params = {"result_path": result_path} if result_path else {} - current = call_unity("get_test_results", params) - if current and "error" in current: - return current - - result = current.get("result", {}) if isinstance(current, dict) else {} - if result.get("status") == "Success" and result.get("timestamp_utc") != before_timestamp: - result["time_waited_seconds"] = round(time.time() - start_time, 2) - return {"result": result} + current_results_response = call_unity("get_test_results", params) + if current_results_response and "error" in current_results_response: + return current_results_response + + current_results_payload = _result_object(current_results_response) + if ( + current_results_payload.get("status") == "Success" + and current_results_payload.get("timestamp_utc") != previous_timestamp + ): + test_results = cast(TestResultsPayload, dict(current_results_payload)) + test_results["time_waited_seconds"] = round(time.time() - start_time, 2) + return {"result": test_results} time.sleep(poll_interval) + timeout_result: TestResultsPayload = { + "status": "Timeout", + "message": "Timed out waiting for a new Unity TestResults XML file.", + "time_waited_seconds": round(time.time() - start_time, 2), + "trigger": trigger_payload, + } + if isinstance(result_path, str): + timeout_result["result_path"] = result_path return { - "result": { - "status": "Timeout", - "message": "Timed out waiting for a new Unity TestResults XML file.", - "time_waited_seconds": round(time.time() - start_time, 2), - "result_path": result_path, - "trigger": trigger_result, - } + "result": timeout_result } -def _wait_for_compilation(timeout: float, start_time: float | None = None) -> dict[str, Any]: +def _wait_for_compilation(timeout: float, start_time: float | None = None) -> JsonRpcResponse: start_time = time.time() if start_time is None else start_time status: str = "Ready" reload_started: bool = False while time.time() - start_time < 20: - res: dict[str, Any] = call_unity("initialize") - if res is None or "error" in res: + initialize_response = call_unity("initialize") + if initialize_response is None or "error" in initialize_response: reload_started = True break time.sleep(0.5) @@ -115,61 +158,75 @@ def _wait_for_compilation(timeout: float, start_time: float | None = None) -> di call_unity("refresh_asset_database") while time.time() - start_time < timeout: - res = call_unity("initialize") - if res and "result" in res: + initialize_response = call_unity("initialize") + if initialize_response and "result" in initialize_response: time.sleep(2.0) - state: dict[str, Any] = call_unity("get_editor_state") - if state and "result" in state: - if not state["result"].get("is_compiling") and not state["result"].get("is_updating"): + editor_state_response = call_unity("get_editor_state") + editor_state = _result_object(editor_state_response) + if editor_state_response and "result" in editor_state_response: + if not editor_state.get("is_compiling") and not editor_state.get("is_updating"): break time.sleep(1.0) else: status = "Timeout" + wait_result: WaitResultPayload = { + "status": status, + "time_waited_seconds": round(time.time() - start_time, 2), + } return { - "result": { - "status": status, - "time_waited_seconds": round(time.time() - start_time, 2), - } + "result": wait_result } -def route_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: +def route_tool(name: str, args: JsonObject) -> JsonRpcResponse: if name in ["tools/list", "list_tools", "listTools"]: return {"result": {"tools": STATIC_TOOLS}} if name == "write_and_compile": - files: list[dict[str, Any]] = args.get("files", []) + files: list[WriteFileSpec] = args.get("files", []) start_time: float = time.time() call_unity("clear_logs") - write_errors: list[dict[str, Any]] = [] + write_errors: list[WriteError] = [] for file_info in files: - res = call_unity("write_file", {"path": file_info["path"], "content": file_info["content"]}) - if res and "error" in res: - write_errors.append({"path": file_info["path"], "error": res["error"]}) + write_file_response = call_unity( + "write_file", + {"path": file_info["path"], "content": file_info["content"]}, + ) + write_error = _error_object(write_file_response) + if write_error is not None: + write_errors.append({"path": file_info["path"], "error": write_error}) if write_errors: - return {"result": {"status": "Failed", "message": "Failed to write some files", "errors": write_errors}} + failure_result: WriteAndCompileFailurePayload = { + "status": "Failed", + "message": "Failed to write some files", + "errors": write_errors, + } + return {"result": failure_result} else: - wait_result: dict[str, Any] = _wait_for_compilation(timeout=90, start_time=start_time) - wait_status: str = wait_result["result"]["status"] - time_waited_seconds: float = wait_result["result"]["time_waited_seconds"] + wait_response = _wait_for_compilation(timeout=90, start_time=start_time) + wait_result = _result_object(wait_response) + wait_status: str = wait_result["status"] + time_waited_seconds: float = wait_result["time_waited_seconds"] - compiler_errors: list[dict[str, Any]] = [] + compiler_errors: list[JsonObject] = [] if wait_status == "Ready": - log_res = call_unity("read_logs", {"count": 200}) - if log_res and "result" in log_res: - for log_entry in log_res["result"].get("logs", []): + log_response = call_unity("read_logs", {"count": 200}) + log_payload = _result_object(log_response) + if log_response and "result" in log_response: + for log_entry in log_payload.get("logs", []): if log_entry.get("Type") in ["Error", "Exception", "Assert"]: compiler_errors.append(log_entry) + success_result: WriteAndCompileSuccessPayload = { + "status": "Failed" if compiler_errors else wait_status, + "time_waited_seconds": time_waited_seconds, + "compiler_errors": compiler_errors, + } return { - "result": { - "status": "Failed" if compiler_errors else wait_status, - "time_waited_seconds": time_waited_seconds, - "compiler_errors": compiler_errors - } + "result": success_result } elif name == "scene_manager": @@ -302,37 +359,50 @@ def route_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: else: return _invalid_action(action, ["get", "set", "delete", "list"]) elif name == "wait": - cond: Any = args.get("condition") + condition: Any = args.get("condition") timeout: float = args.get("timeout_seconds", 60) start_time: float = time.time() status: str = "Ready" - if cond == "compilation": + if condition == "compilation": return _wait_for_compilation(timeout=timeout, start_time=start_time) - elif cond == "play_mode": + elif condition == "play_mode": target_state = args.get("state", True) while time.time() - start_time < timeout: - state_res = call_unity("get_editor_state") - if state_res and "result" in state_res: - if state_res["result"].get("is_playing") == target_state: break + editor_state_response = call_unity("get_editor_state") + editor_state = _result_object(editor_state_response) + if editor_state_response and "result" in editor_state_response: + if editor_state.get("is_playing") == target_state: + break time.sleep(1.0) - else: status = "Timeout" - elif cond == "import": + else: + status = "Timeout" + elif condition == "import": while time.time() - start_time < timeout: - res = call_unity("is_asset_import_idle") - if res and "result" in res: - if res["result"].get("is_idle"): break + import_idle_response = call_unity("is_asset_import_idle") + import_idle_state = _result_object(import_idle_response) + if import_idle_response and "result" in import_idle_response: + if import_idle_state.get("is_idle"): + break time.sleep(1.0) - else: status = "Timeout" - elif cond == "editor_idle": + else: + status = "Timeout" + elif condition == "editor_idle": while time.time() - start_time < timeout: - res = call_unity("is_editor_idle") - if res and "result" in res: - if res["result"].get("is_idle"): break + editor_idle_response = call_unity("is_editor_idle") + editor_idle_state = _result_object(editor_idle_response) + if editor_idle_response and "result" in editor_idle_response: + if editor_idle_state.get("is_idle"): + break time.sleep(1.0) - else: status = "Timeout" + else: + status = "Timeout" - return {"result": {"status": status, "time_waited_seconds": round(time.time() - start_time, 2)}} + wait_result: WaitResultPayload = { + "status": status, + "time_waited_seconds": round(time.time() - start_time, 2), + } + return {"result": wait_result} else: return call_unity(name, args) diff --git a/Editor/nexus_bridge/schemas.py b/Editor/nexus_bridge/schemas.py index 8739458..fdb100c 100644 --- a/Editor/nexus_bridge/schemas.py +++ b/Editor/nexus_bridge/schemas.py @@ -8,10 +8,10 @@ """ from __future__ import annotations -from typing import Any +from ._types import JsonObject, ResourceDefinition, ToolDefinition # --- Shared sub-schemas --- -VECTOR3_SCHEMA: dict[str, Any] = { +VECTOR3_SCHEMA: JsonObject = { "type": "object", "properties": { "x": {"type": "number"}, @@ -20,7 +20,7 @@ }, } -STATIC_TOOLS: list[dict[str, Any]] = [ +STATIC_TOOLS: list[ToolDefinition] = [ # --- Consolidated Core Managers --- { "name": "unity_scene_manager", @@ -185,7 +185,7 @@ {"name": "unity_lint_project", "description": "Run Roslyn-based C# audit of the entire project", "inputSchema": {"type": "object", "properties": {}}} ] -STATIC_RESOURCES: list[dict[str, Any]] = [ +STATIC_RESOURCES: list[ResourceDefinition] = [ { "uri": "unity://docs/api-reference", "name": "API Reference", From 3d06a2d60db533e2c6d2f6ece72b11adf077e113 Mon Sep 17 00:00:00 2001 From: air17 Date: Fri, 12 Jun 2026 18:07:37 +0000 Subject: [PATCH 2/3] fix schemas and typing --- Editor/nexus_bridge/_types.py | 4 +- Editor/nexus_bridge/_types.py.meta | 7 ++ Editor/nexus_bridge/routing.py | 2 +- Editor/nexus_bridge/schemas.py | 132 +++++++++++++++++++++++++---- Editor/tests/test_routing.py | 14 +++ Editor/tests/test_schemas.py | 82 ++++++++++++++++++ Editor/tests/test_schemas.py.meta | 7 ++ 7 files changed, 227 insertions(+), 21 deletions(-) create mode 100644 Editor/nexus_bridge/_types.py.meta create mode 100644 Editor/tests/test_schemas.py create mode 100644 Editor/tests/test_schemas.py.meta diff --git a/Editor/nexus_bridge/_types.py b/Editor/nexus_bridge/_types.py index bae9566..6f4532d 100644 --- a/Editor/nexus_bridge/_types.py +++ b/Editor/nexus_bridge/_types.py @@ -1,7 +1,7 @@ """Private type definitions for the NexusUnity Python bridge.""" from __future__ import annotations -from typing import Any, TypeAlias, TypedDict +from typing import Any, NotRequired, TypeAlias, TypedDict JsonObject: TypeAlias = dict[str, Any] @@ -27,7 +27,7 @@ class ToolDefinition(TypedDict): class ResourceDefinition(TypedDict): uri: str name: str - mimeType: str + mimeType: NotRequired[str] class JsonRpcResponse(TypedDict, total=False): diff --git a/Editor/nexus_bridge/_types.py.meta b/Editor/nexus_bridge/_types.py.meta new file mode 100644 index 0000000..63f4e9b --- /dev/null +++ b/Editor/nexus_bridge/_types.py.meta @@ -0,0 +1,7 @@ +fileFormatVersion: 2 +guid: f0bead8511344b4f920937b1c98230ac +DefaultImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Editor/nexus_bridge/routing.py b/Editor/nexus_bridge/routing.py index e9db828..e460f58 100644 --- a/Editor/nexus_bridge/routing.py +++ b/Editor/nexus_bridge/routing.py @@ -77,7 +77,7 @@ def _extract_created_instance_id(response: JsonRpcResponse) -> int | None: def _apply_created_transform(response: JsonRpcResponse, args: JsonObject) -> JsonRpcResponse: instance_id = _extract_created_instance_id(response) - if not instance_id: + if instance_id is None: return response params = _transform_params(args, instance_id) if len(params) <= 1: diff --git a/Editor/nexus_bridge/schemas.py b/Editor/nexus_bridge/schemas.py index fdb100c..a58f22f 100644 --- a/Editor/nexus_bridge/schemas.py +++ b/Editor/nexus_bridge/schemas.py @@ -41,24 +41,120 @@ "description": "Unified GameObject hierarchy and lifecycle management", "inputSchema": { "type": "object", - "properties": { - "action": {"type": "string", "enum": ["create_empty", "create", "create_gameobject", "create_game_object", "create_primitive", "create_hierarchy", "destroy", "duplicate", "rename", "set_name", "set_transform", "set_active", "set_parent", "set_sibling_index"]}, - "instance_id": {"type": "integer"}, - "name": {"type": "string"}, - "new_name": {"type": "string"}, - "parent_id": {"type": "integer"}, - "primitive_type": {"type": "string", "enum": ["Cube", "Sphere", "Capsule", "Cylinder", "Plane", "Quad"]}, - "position": VECTOR3_SCHEMA, - "rotation": VECTOR3_SCHEMA, - "scale": VECTOR3_SCHEMA, - "eulerAngles": VECTOR3_SCHEMA, - "localScale": VECTOR3_SCHEMA, - "material_path": {"type": "string"}, - "tree": {"type": "object"}, - "active": {"type": "boolean"}, - "index": {"type": "string"} - }, - "required": ["action"] + "oneOf": [ + { + "description": "Create an empty GameObject, including common create-action aliases", + "properties": { + "action": {"type": "string", "enum": ["create_empty", "create", "create_gameobject", "create_game_object"]}, + "name": {"type": "string"}, + "parent_id": {"type": "integer"}, + "position": VECTOR3_SCHEMA, + "rotation": VECTOR3_SCHEMA, + "scale": VECTOR3_SCHEMA, + "eulerAngles": VECTOR3_SCHEMA, + "localScale": VECTOR3_SCHEMA, + }, + "required": ["action", "name"] + }, + { + "description": "Create a primitive GameObject", + "properties": { + "action": {"const": "create_primitive"}, + "primitive_type": {"type": "string", "enum": ["Cube", "Sphere", "Capsule", "Cylinder", "Plane", "Quad"]}, + "name": {"type": "string"}, + "parent_id": {"type": "integer"}, + "position": VECTOR3_SCHEMA, + "rotation": VECTOR3_SCHEMA, + "scale": VECTOR3_SCHEMA, + "material_path": {"type": "string"}, + }, + "required": ["action", "primitive_type"] + }, + { + "description": "Batch-create a hierarchy of GameObjects", + "properties": { + "action": {"const": "create_hierarchy"}, + "tree": {"type": "object"}, + "parent_id": {"type": "integer"}, + }, + "required": ["action", "tree"] + }, + { + "description": "Destroy a GameObject", + "properties": { + "action": {"const": "destroy"}, + "instance_id": {"type": "integer"}, + }, + "required": ["action", "instance_id"] + }, + { + "description": "Duplicate a GameObject", + "properties": { + "action": {"const": "duplicate"}, + "instance_id": {"type": "integer"}, + }, + "required": ["action", "instance_id"] + }, + { + "description": "Rename a GameObject, including the rename alias", + "properties": { + "action": {"type": "string", "enum": ["rename", "set_name"]}, + "instance_id": {"type": "integer"}, + "name": {"type": "string"}, + "new_name": {"type": "string"}, + }, + "required": ["action", "instance_id"], + "anyOf": [ + {"required": ["name"]}, + {"required": ["new_name"]} + ] + }, + { + "description": "Move, rotate, or scale a GameObject, including the transform alias", + "properties": { + "action": {"type": "string", "enum": ["set_transform", "transform"]}, + "instance_id": {"type": "integer"}, + "position": VECTOR3_SCHEMA, + "rotation": VECTOR3_SCHEMA, + "scale": VECTOR3_SCHEMA, + "eulerAngles": VECTOR3_SCHEMA, + "localScale": VECTOR3_SCHEMA, + }, + "required": ["action", "instance_id"] + }, + { + "description": "Enable or disable a GameObject", + "properties": { + "action": {"const": "set_active"}, + "instance_id": {"type": "integer"}, + "active": {"type": "boolean"}, + }, + "required": ["action", "instance_id", "active"] + }, + { + "description": "Reparent a GameObject", + "properties": { + "action": {"const": "set_parent"}, + "instance_id": {"type": "integer"}, + "parent_id": {"type": "integer"}, + }, + "required": ["action", "instance_id", "parent_id"] + }, + { + "description": "Reorder a GameObject within its siblings", + "properties": { + "action": {"const": "set_sibling_index"}, + "instance_id": {"type": "integer"}, + "index": { + "oneOf": [ + {"type": "integer"}, + {"type": "string", "enum": ["first", "last"]} + ] + }, + }, + "required": ["action", "instance_id", "index"] + } + ] } }, { diff --git a/Editor/tests/test_routing.py b/Editor/tests/test_routing.py index 39418f3..be05dc8 100644 --- a/Editor/tests/test_routing.py +++ b/Editor/tests/test_routing.py @@ -18,6 +18,7 @@ def test_route_tool_dispatches_to_expected_unity_methods(self) -> None: test_cases: list[tuple[str, dict[str, Any], str, tuple[Any, ...]]] = [ ("scene_manager", {"action": "open", "path": "Assets/TestScene.unity"}, "open_scene", ({"path": "Assets/TestScene.unity"},)), ("hierarchy_manager", {"action": "destroy", "instance_id": 42}, "destroy_game_object", ({"instance_id": 42},)), + ("hierarchy_manager", {"action": "set_sibling_index", "instance_id": 42, "index": 2}, "set_sibling_index", ({"instance_id": 42, "index": 2},)), ("component_manager", {"action": "add", "instance_id": 42, "component_name": "BoxCollider"}, "add_component", ({"instance_id": 42, "component_name": "BoxCollider"},)), ("search_manager", {"strategy": "path", "query": "/Canvas/Button"}, "find_by_path", ({"path": "/Canvas/Button"},)), ("asset_manager", {"action": "refresh"}, "refresh_asset_database", tuple()), @@ -34,6 +35,19 @@ def test_route_tool_dispatches_to_expected_unity_methods(self) -> None: self.assertEqual({"result": {"ok": True}}, response) mock_call_unity.assert_called_once_with(unity_method, *unity_params) + def test_apply_created_transform_accepts_zero_instance_id(self) -> None: + response: dict[str, Any] = {"result": {"data": {"instance_id": 0}}} + args: dict[str, Any] = {"position": {"x": 1, "y": 2, "z": 3}} + + with patch("nexus_bridge.routing.call_unity", return_value={"result": {"ok": True}}) as mock_call_unity: + returned = routing._apply_created_transform(response, args) + + self.assertIs(returned, response) + mock_call_unity.assert_called_once_with( + "set_transform", + {"instance_id": 0, "position": {"x": 1, "y": 2, "z": 3}}, + ) + class RunTestsWaitTests(unittest.TestCase): def test_run_tests_wait_returns_success_when_new_results_appear(self) -> None: diff --git a/Editor/tests/test_schemas.py b/Editor/tests/test_schemas.py new file mode 100644 index 0000000..9962919 --- /dev/null +++ b/Editor/tests/test_schemas.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import os +import sys +import unittest +from typing import Any + +EDITOR_DIR: str = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if EDITOR_DIR not in sys.path: + sys.path.insert(0, EDITOR_DIR) + +from nexus_bridge.schemas import STATIC_TOOLS + + +def _get_tool(name: str) -> dict[str, Any]: + for tool in STATIC_TOOLS: + if tool["name"] == name: + return tool + raise AssertionError(f"Could not find tool {name}.") + + +def _action_values(variant: dict[str, Any]) -> set[str]: + action_schema = variant["properties"]["action"] + if "const" in action_schema: + return {action_schema["const"]} + return set(action_schema["enum"]) + + +class HierarchyManagerSchemaTests(unittest.TestCase): + def test_hierarchy_manager_uses_action_specific_variants(self) -> None: + hierarchy_manager = _get_tool("unity_hierarchy_manager") + input_schema = hierarchy_manager["inputSchema"] + + self.assertNotIn("required", input_schema) + self.assertIn("oneOf", input_schema) + self.assertGreaterEqual(len(input_schema["oneOf"]), 1) + + def test_rename_variant_requires_instance_id_and_name_or_new_name(self) -> None: + hierarchy_manager = _get_tool("unity_hierarchy_manager") + rename_variant = next( + variant + for variant in hierarchy_manager["inputSchema"]["oneOf"] + if "rename" in _action_values(variant) + ) + + self.assertCountEqual(["action", "instance_id"], rename_variant["required"]) + self.assertEqual( + [{"required": ["name"]}, {"required": ["new_name"]}], + rename_variant["anyOf"], + ) + + def test_transform_alias_is_advertised(self) -> None: + hierarchy_manager = _get_tool("unity_hierarchy_manager") + transform_variant = next( + variant + for variant in hierarchy_manager["inputSchema"]["oneOf"] + if "set_transform" in _action_values(variant) + ) + + self.assertIn("transform", _action_values(transform_variant)) + + def test_set_sibling_index_accepts_integer_or_edge_keywords(self) -> None: + hierarchy_manager = _get_tool("unity_hierarchy_manager") + sibling_variant = next( + variant + for variant in hierarchy_manager["inputSchema"]["oneOf"] + if "set_sibling_index" in _action_values(variant) + ) + + self.assertCountEqual(["action", "instance_id", "index"], sibling_variant["required"]) + index_schema = sibling_variant["properties"]["index"] + self.assertEqual( + [ + {"type": "integer"}, + {"type": "string", "enum": ["first", "last"]}, + ], + index_schema["oneOf"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/Editor/tests/test_schemas.py.meta b/Editor/tests/test_schemas.py.meta new file mode 100644 index 0000000..910a12a --- /dev/null +++ b/Editor/tests/test_schemas.py.meta @@ -0,0 +1,7 @@ +fileFormatVersion: 2 +guid: 9046bb8bd125460690905034f6644fcf +DefaultImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: From 2dc74d45fac31e926db6e8f27f61329a2e98b4d4 Mon Sep 17 00:00:00 2001 From: air17 Date: Fri, 12 Jun 2026 18:19:23 +0000 Subject: [PATCH 3/3] Refactor routing handlers and fix compilation wait timeout --- Editor/nexus_bridge/routing.py | 533 +++++++++++++++++++-------------- Editor/tests/test_routing.py | 158 +++++++++- 2 files changed, 465 insertions(+), 226 deletions(-) diff --git a/Editor/nexus_bridge/routing.py b/Editor/nexus_bridge/routing.py index e460f58..b418b14 100644 --- a/Editor/nexus_bridge/routing.py +++ b/Editor/nexus_bridge/routing.py @@ -7,7 +7,7 @@ from __future__ import annotations import time -from typing import Any, Mapping, Sequence, cast +from typing import Any, Callable, Mapping, Sequence, cast from ._logging import logger from ._transport import call_unity @@ -25,6 +25,8 @@ WriteFileSpec, ) +RouteHandler = Callable[[JsonObject], JsonRpcResponse] + def _compact(params: JsonObject) -> JsonObject: return {key: value for key, value in params.items() if value is not None} @@ -147,7 +149,8 @@ def _wait_for_compilation(timeout: float, start_time: float | None = None) -> Js status: str = "Ready" reload_started: bool = False - while time.time() - start_time < 20: + reload_wait_timeout = min(20.0, timeout) + while time.time() - start_time < reload_wait_timeout: initialize_response = call_unity("initialize") if initialize_response is None or "error" in initialize_response: reload_started = True @@ -179,230 +182,310 @@ def _wait_for_compilation(timeout: float, start_time: float | None = None) -> Js } -def route_tool(name: str, args: JsonObject) -> JsonRpcResponse: - if name in ["tools/list", "list_tools", "listTools"]: - return {"result": {"tools": STATIC_TOOLS}} - - if name == "write_and_compile": - files: list[WriteFileSpec] = args.get("files", []) - start_time: float = time.time() - call_unity("clear_logs") - - write_errors: list[WriteError] = [] - for file_info in files: - write_file_response = call_unity( - "write_file", - {"path": file_info["path"], "content": file_info["content"]}, - ) - write_error = _error_object(write_file_response) - if write_error is not None: - write_errors.append({"path": file_info["path"], "error": write_error}) - - if write_errors: - failure_result: WriteAndCompileFailurePayload = { - "status": "Failed", - "message": "Failed to write some files", - "errors": write_errors, - } - return {"result": failure_result} +def _route_list_tools(_: JsonObject) -> JsonRpcResponse: + return {"result": {"tools": STATIC_TOOLS}} + + +def _route_write_and_compile(args: JsonObject) -> JsonRpcResponse: + files: list[WriteFileSpec] = args.get("files", []) + start_time: float = time.time() + call_unity("clear_logs") + + write_errors: list[WriteError] = [] + for file_info in files: + write_file_response = call_unity( + "write_file", + {"path": file_info["path"], "content": file_info["content"]}, + ) + write_error = _error_object(write_file_response) + if write_error is not None: + write_errors.append({"path": file_info["path"], "error": write_error}) + + if write_errors: + failure_result: WriteAndCompileFailurePayload = { + "status": "Failed", + "message": "Failed to write some files", + "errors": write_errors, + } + return {"result": failure_result} + + wait_response = _wait_for_compilation(timeout=90, start_time=start_time) + wait_result = _result_object(wait_response) + wait_status: str = wait_result["status"] + time_waited_seconds: float = wait_result["time_waited_seconds"] + + compiler_errors: list[JsonObject] = [] + if wait_status == "Ready": + log_response = call_unity("read_logs", {"count": 200}) + log_payload = _result_object(log_response) + if log_response and "result" in log_response: + for log_entry in log_payload.get("logs", []): + if log_entry.get("Type") in ["Error", "Exception", "Assert"]: + compiler_errors.append(log_entry) + + success_result: WriteAndCompileSuccessPayload = { + "status": "Failed" if compiler_errors else wait_status, + "time_waited_seconds": time_waited_seconds, + "compiler_errors": compiler_errors, + } + return { + "result": success_result + } + + +def _route_scene_manager(args: JsonObject) -> JsonRpcResponse: + aliases = {"create_scene": "create", "open_scene": "open", "save_scene": "save", "list_scenes": "list"} + action = _alias(args.get("action"), aliases) + if action == "create": + return call_unity("create_scene", _compact({"name": args.get("name"), "path": args.get("path"), "open_if_exists": args.get("open_if_exists")})) + if action == "open": + return call_unity("open_scene", {"path": args.get("path")}) + if action == "save": + return call_unity("save_scene", {"path": args.get("path")}) + if action == "list": + return call_unity("list_scenes") + return _invalid_action(args.get("action"), ["create", "create_scene", "open", "open_scene", "save", "save_scene", "list", "list_scenes"]) + + +def _route_hierarchy_manager(args: JsonObject) -> JsonRpcResponse: + aliases = { + "create": "create_empty", + "create_gameobject": "create_empty", + "create_game_object": "create_empty", + "rename": "set_name", + "transform": "set_transform", + } + action = _alias(args.get("action"), aliases) + if action == "create_empty": + response = call_unity("create_game_object", _compact({"name": args.get("name"), "parent_id": args.get("parent_id")})) + return _apply_created_transform(response, args) + if action == "create_primitive": + return call_unity("create_primitive", _compact({ + "primitive_type": args.get("primitive_type"), + "name": args.get("name"), + "parent_id": args.get("parent_id"), + "position": args.get("position"), + "rotation": args.get("rotation"), + "scale": args.get("scale"), + "material_path": args.get("material_path"), + })) + if action == "create_hierarchy": + return call_unity("create_hierarchy", _compact({"tree": args.get("tree"), "parent_id": args.get("parent_id")})) + if action == "set_name": + return call_unity("set_property", {"instance_id": args.get("instance_id"), "property_name": "m_Name", "value": args.get("name") or args.get("new_name")}) + if action == "set_transform": + return call_unity("set_transform", _transform_params(args)) + if action == "destroy": + return call_unity("destroy_game_object", {"instance_id": args.get("instance_id")}) + if action == "duplicate": + return call_unity("duplicate_object", {"instance_id": args.get("instance_id")}) + if action == "set_active": + return call_unity("set_active", {"instance_id": args.get("instance_id"), "active": args.get("active")}) + if action == "set_parent": + return call_unity("set_parent", {"instance_id": args.get("instance_id"), "parent_id": args.get("parent_id")}) + if action == "set_sibling_index": + return call_unity("set_sibling_index", {"instance_id": args.get("instance_id"), "index": args.get("index")}) + return _invalid_action(args.get("action"), ["create_empty", "create", "create_gameobject", "create_game_object", "create_primitive", "create_hierarchy", "destroy", "duplicate", "rename", "set_name", "set_transform", "set_active", "set_parent", "set_sibling_index"]) + + +def _route_component_manager(args: JsonObject) -> JsonRpcResponse: + action = args.get("action") + if action == "add": + return call_unity("add_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) + if action == "remove": + return call_unity("remove_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) + if action == "inspect": + return call_unity("inspect_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) + if action == "get_schema": + return call_unity("get_component_schema", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) + if action == "update_properties": + return call_unity("update_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name"), "properties": args.get("properties")}) + if action == "set_property": + return call_unity("set_property", {"instance_id": args.get("instance_id"), "property_name": args.get("property_name"), "value": args.get("value")}) + if action == "set_enabled": + return call_unity("set_enabled", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name"), "enabled": args.get("enabled")}) + return _invalid_action(action, ["add", "remove", "inspect", "get_schema", "update_properties", "set_property", "set_enabled"]) + + +def _route_search_manager(args: JsonObject) -> JsonRpcResponse: + strategy = args.get("strategy") + if strategy == "regex": + return call_unity("find_objects", {"name": args.get("query"), "tag": args.get("tag"), "type": args.get("type")}) + if strategy == "path": + return call_unity("find_by_path", {"path": args.get("query")}) + if strategy == "semantic": + return call_unity("semantic_find", {"query": args.get("query")}) + if strategy == "references": + return call_unity("find_references", {"target_id": args.get("target_id"), "target_guid": args.get("target_guid")}) + return {"error": {"code": -32602, "message": f"Invalid strategy: {strategy}. Valid strategies: regex, path, semantic, references"}} + + +def _route_asset_manager(args: JsonObject) -> JsonRpcResponse: + action = args.get("action") + if action == "search": + return call_unity("list_assets", {"filter": args.get("filter")}) + if action == "explore": + return call_unity("explore_asset", {"path": args.get("path")}) + if action == "create_material": + return call_unity("create_material", _compact({ + "name": args.get("name"), + "shader": args.get("shader"), + "path": args.get("path"), + "base_color": args.get("base_color") or args.get("color"), + "emission_color": args.get("emission_color") or args.get("emission"), + })) + if action == "import": + return call_unity("import_asset", {"path": args.get("path")}) + if action == "refresh": + return call_unity("refresh_asset_database") + if action == "instantiate_prefab": + return call_unity("instantiate_prefab", {"path": args.get("path")}) + if action == "create_prefab": + return call_unity("create_prefab", {"instance_id": args.get("instance_id"), "path": args.get("path")}) + if action == "apply_overrides": + return call_unity("apply_prefab_overrides", {"instance_id": args.get("instance_id")}) + if action == "revert_overrides": + return call_unity("revert_prefab_overrides", {"instance_id": args.get("instance_id")}) + return _invalid_action(action, ["search", "explore", "create_material", "import", "refresh", "instantiate_prefab", "create_prefab", "apply_overrides", "revert_overrides"]) + + +def _route_editor_controller(args: JsonObject) -> JsonRpcResponse: + action = args.get("action") + if action == "undo": + return call_unity("undo") + if action == "redo": + return call_unity("redo") + if action == "play": + return call_unity("toggle_play_mode", {"value": args.get("state")}) + if action == "pause": + return call_unity("pause_play_mode", {"value": args.get("state")}) + if action == "step": + return call_unity("step_frame") + if action == "menu": + return call_unity("execute_menu_item", {"item_path": args.get("item_path")}) + if action == "read_logs": + return call_unity("read_logs", {"count": args.get("count", 100)}) + if action == "clear_logs": + return call_unity("clear_logs") + if action == "get_state": + return call_unity("get_editor_state") + if action == "get_server_status": + return call_unity("get_server_status") + if action == "refresh_assets": + return call_unity("refresh_asset_database") + if action == "run_tests": + return call_unity("run_tests", _compact({"mode": args.get("mode", "EditMode"), "filter": args.get("filter")})) + if action == "get_test_results": + return call_unity("get_test_results", _compact({"result_path": args.get("result_path")})) + if action == "run_tests_wait": + return _run_tests_wait(args) + if action == "get_tool_usage_stats": + return call_unity("get_tool_usage_stats") + if action == "reset_tool_usage_stats": + return call_unity("reset_tool_usage_stats") + return _invalid_action(action, ["undo", "redo", "play", "pause", "step", "menu", "read_logs", "clear_logs", "get_state", "get_server_status", "refresh_assets", "run_tests", "get_test_results", "run_tests_wait", "get_tool_usage_stats", "reset_tool_usage_stats"]) + + +def _route_ui_automation(args: JsonObject) -> JsonRpcResponse: + action = args.get("action") + if action == "list_windows": + return call_unity("ui_list_windows") + if action == "get_hierarchy": + return call_unity("ui_get_hierarchy", _compact({"window_title": args.get("window_title"), "deep": args.get("deep")})) + if action == "query": + return call_unity("ui_query_elements", _compact({"window_title": args.get("window_title"), "name": args.get("name"), "text": args.get("text"), "class_name": args.get("class_name")})) + if action == "get_window_rect": + return call_unity("ui_get_window_rect", {"window_title": args.get("window_title")}) + if action == "set_window_rect": + return call_unity("ui_set_window_rect", _compact({"window_title": args.get("window_title"), "x": args.get("x"), "y": args.get("y"), "width": args.get("width"), "height": args.get("height")})) + if action == "capture_window_snapshot": + return call_unity("ui_capture_window_snapshot", _compact({"window_title": args.get("window_title"), "include_image": args.get("include_image"), "include_hierarchy": args.get("include_hierarchy")})) + if action == "click": + return call_unity("ui_click", {"window_title": args.get("window_title"), "element_name": args.get("element_name")}) + if action == "input": + return call_unity("ui_input_text", {"window_title": args.get("window_title"), "element_name": args.get("element_name"), "text": args.get("text")}) + return _invalid_action(action, ["list_windows", "get_hierarchy", "query", "get_window_rect", "set_window_rect", "capture_window_snapshot", "click", "input"]) + + +def _route_playerprefs_manager(args: JsonObject) -> JsonRpcResponse: + action = args.get("action") + if action == "get": + return call_unity("get_player_pref", {"key": args.get("key"), "type": args.get("type", "string")}) + if action == "set": + return call_unity("set_player_pref", {"key": args.get("key"), "value": args.get("value"), "type": args.get("type", "string")}) + if action == "delete": + return call_unity("delete_player_pref", {"key": args.get("key")}) + if action == "list": + return call_unity("list_player_prefs") + return _invalid_action(action, ["get", "set", "delete", "list"]) + + +def _route_wait(args: JsonObject) -> JsonRpcResponse: + condition: Any = args.get("condition") + timeout: float = args.get("timeout_seconds", 60) + start_time: float = time.time() + status: str = "Ready" + + if condition == "compilation": + return _wait_for_compilation(timeout=timeout, start_time=start_time) + if condition == "play_mode": + target_state = args.get("state", True) + while time.time() - start_time < timeout: + editor_state_response = call_unity("get_editor_state") + editor_state = _result_object(editor_state_response) + if editor_state_response and "result" in editor_state_response: + if editor_state.get("is_playing") == target_state: + break + time.sleep(1.0) else: - wait_response = _wait_for_compilation(timeout=90, start_time=start_time) - wait_result = _result_object(wait_response) - wait_status: str = wait_result["status"] - time_waited_seconds: float = wait_result["time_waited_seconds"] - - compiler_errors: list[JsonObject] = [] - if wait_status == "Ready": - log_response = call_unity("read_logs", {"count": 200}) - log_payload = _result_object(log_response) - if log_response and "result" in log_response: - for log_entry in log_payload.get("logs", []): - if log_entry.get("Type") in ["Error", "Exception", "Assert"]: - compiler_errors.append(log_entry) - - success_result: WriteAndCompileSuccessPayload = { - "status": "Failed" if compiler_errors else wait_status, - "time_waited_seconds": time_waited_seconds, - "compiler_errors": compiler_errors, - } - return { - "result": success_result - } - - elif name == "scene_manager": - aliases = {"create_scene": "create", "open_scene": "open", "save_scene": "save", "list_scenes": "list"} - action = _alias(args.get("action"), aliases) - if action == "create": - return call_unity("create_scene", _compact({"name": args.get("name"), "path": args.get("path"), "open_if_exists": args.get("open_if_exists")})) - elif action == "open": - return call_unity("open_scene", {"path": args.get("path")}) - elif action == "save": - return call_unity("save_scene", {"path": args.get("path")}) - elif action == "list": - return call_unity("list_scenes") + status = "Timeout" + elif condition == "import": + while time.time() - start_time < timeout: + import_idle_response = call_unity("is_asset_import_idle") + import_idle_state = _result_object(import_idle_response) + if import_idle_response and "result" in import_idle_response: + if import_idle_state.get("is_idle"): + break + time.sleep(1.0) else: - return _invalid_action(args.get("action"), ["create", "create_scene", "open", "open_scene", "save", "save_scene", "list", "list_scenes"]) - - elif name == "hierarchy_manager": - aliases = { - "create": "create_empty", - "create_gameobject": "create_empty", - "create_game_object": "create_empty", - "rename": "set_name", - "transform": "set_transform", - } - action = _alias(args.get("action"), aliases) - if action == "create_empty": - res = call_unity("create_game_object", _compact({"name": args.get("name"), "parent_id": args.get("parent_id")})) - return _apply_created_transform(res, args) - elif action == "create_primitive": - return call_unity("create_primitive", _compact({ - "primitive_type": args.get("primitive_type"), - "name": args.get("name"), - "parent_id": args.get("parent_id"), - "position": args.get("position"), - "rotation": args.get("rotation"), - "scale": args.get("scale"), - "material_path": args.get("material_path"), - })) - elif action == "create_hierarchy": - return call_unity("create_hierarchy", _compact({"tree": args.get("tree"), "parent_id": args.get("parent_id")})) - elif action == "set_name": - return call_unity("set_property", {"instance_id": args.get("instance_id"), "property_name": "m_Name", "value": args.get("name") or args.get("new_name")}) - elif action == "set_transform": - return call_unity("set_transform", _transform_params(args)) - elif action == "destroy": return call_unity("destroy_game_object", {"instance_id": args.get("instance_id")}) - elif action == "duplicate": return call_unity("duplicate_object", {"instance_id": args.get("instance_id")}) - elif action == "set_active": return call_unity("set_active", {"instance_id": args.get("instance_id"), "active": args.get("active")}) - elif action == "set_parent": return call_unity("set_parent", {"instance_id": args.get("instance_id"), "parent_id": args.get("parent_id")}) - elif action == "set_sibling_index": return call_unity("set_sibling_index", {"instance_id": args.get("instance_id"), "index": args.get("index")}) + status = "Timeout" + elif condition == "editor_idle": + while time.time() - start_time < timeout: + editor_idle_response = call_unity("is_editor_idle") + editor_idle_state = _result_object(editor_idle_response) + if editor_idle_response and "result" in editor_idle_response: + if editor_idle_state.get("is_idle"): + break + time.sleep(1.0) else: - return _invalid_action(args.get("action"), ["create_empty", "create", "create_gameobject", "create_game_object", "create_primitive", "create_hierarchy", "destroy", "duplicate", "rename", "set_name", "set_transform", "set_active", "set_parent", "set_sibling_index"]) - - elif name == "component_manager": - action = args.get("action") - if action == "add": return call_unity("add_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) - elif action == "remove": return call_unity("remove_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) - elif action == "inspect": return call_unity("inspect_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) - elif action == "get_schema": return call_unity("get_component_schema", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name")}) - elif action == "update_properties": return call_unity("update_component", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name"), "properties": args.get("properties")}) - elif action == "set_property": return call_unity("set_property", {"instance_id": args.get("instance_id"), "property_name": args.get("property_name"), "value": args.get("value")}) - elif action == "set_enabled": return call_unity("set_enabled", {"instance_id": args.get("instance_id"), "component_name": args.get("component_name"), "enabled": args.get("enabled")}) - else: return _invalid_action(action, ["add", "remove", "inspect", "get_schema", "update_properties", "set_property", "set_enabled"]) - - elif name == "search_manager": - strategy = args.get("strategy") - if strategy == "regex": return call_unity("find_objects", {"name": args.get("query"), "tag": args.get("tag"), "type": args.get("type")}) - elif strategy == "path": return call_unity("find_by_path", {"path": args.get("query")}) - elif strategy == "semantic": return call_unity("semantic_find", {"query": args.get("query")}) - elif strategy == "references": return call_unity("find_references", {"target_id": args.get("target_id"), "target_guid": args.get("target_guid")}) - else: return {"error": {"code": -32602, "message": f"Invalid strategy: {strategy}. Valid strategies: regex, path, semantic, references"}} - - elif name == "asset_manager": - action = args.get("action") - if action == "search": return call_unity("list_assets", {"filter": args.get("filter")}) - elif action == "explore": return call_unity("explore_asset", {"path": args.get("path")}) - elif action == "create_material": - return call_unity("create_material", _compact({ - "name": args.get("name"), - "shader": args.get("shader"), - "path": args.get("path"), - "base_color": args.get("base_color") or args.get("color"), - "emission_color": args.get("emission_color") or args.get("emission"), - })) - elif action == "import": return call_unity("import_asset", {"path": args.get("path")}) - elif action == "refresh": return call_unity("refresh_asset_database") - elif action == "instantiate_prefab": return call_unity("instantiate_prefab", {"path": args.get("path")}) - elif action == "create_prefab": return call_unity("create_prefab", {"instance_id": args.get("instance_id"), "path": args.get("path")}) - elif action == "apply_overrides": return call_unity("apply_prefab_overrides", {"instance_id": args.get("instance_id")}) - elif action == "revert_overrides": return call_unity("revert_prefab_overrides", {"instance_id": args.get("instance_id")}) - else: return _invalid_action(action, ["search", "explore", "create_material", "import", "refresh", "instantiate_prefab", "create_prefab", "apply_overrides", "revert_overrides"]) - - elif name == "editor_controller": - action = args.get("action") - if action == "undo": return call_unity("undo") - elif action == "redo": return call_unity("redo") - elif action == "play": return call_unity("toggle_play_mode", {"value": args.get("state")}) - elif action == "pause": return call_unity("pause_play_mode", {"value": args.get("state")}) - elif action == "step": return call_unity("step_frame") - elif action == "menu": return call_unity("execute_menu_item", {"item_path": args.get("item_path")}) - elif action == "read_logs": return call_unity("read_logs", {"count": args.get("count", 100)}) - elif action == "clear_logs": return call_unity("clear_logs") - elif action == "get_state": return call_unity("get_editor_state") - elif action == "get_server_status": return call_unity("get_server_status") - elif action == "refresh_assets": return call_unity("refresh_asset_database") - elif action == "run_tests": return call_unity("run_tests", _compact({"mode": args.get("mode", "EditMode"), "filter": args.get("filter")})) - elif action == "get_test_results": return call_unity("get_test_results", _compact({"result_path": args.get("result_path")})) - elif action == "run_tests_wait": return _run_tests_wait(args) - elif action == "get_tool_usage_stats": return call_unity("get_tool_usage_stats") - elif action == "reset_tool_usage_stats": return call_unity("reset_tool_usage_stats") - else: return _invalid_action(action, ["undo", "redo", "play", "pause", "step", "menu", "read_logs", "clear_logs", "get_state", "get_server_status", "refresh_assets", "run_tests", "get_test_results", "run_tests_wait", "get_tool_usage_stats", "reset_tool_usage_stats"]) - - elif name == "ui_automation": - action = args.get("action") - if action == "list_windows": return call_unity("ui_list_windows") - elif action == "get_hierarchy": return call_unity("ui_get_hierarchy", _compact({"window_title": args.get("window_title"), "deep": args.get("deep")})) - elif action == "query": return call_unity("ui_query_elements", _compact({"window_title": args.get("window_title"), "name": args.get("name"), "text": args.get("text"), "class_name": args.get("class_name")})) - elif action == "get_window_rect": return call_unity("ui_get_window_rect", {"window_title": args.get("window_title")}) - elif action == "set_window_rect": return call_unity("ui_set_window_rect", _compact({"window_title": args.get("window_title"), "x": args.get("x"), "y": args.get("y"), "width": args.get("width"), "height": args.get("height")})) - elif action == "capture_window_snapshot": return call_unity("ui_capture_window_snapshot", _compact({"window_title": args.get("window_title"), "include_image": args.get("include_image"), "include_hierarchy": args.get("include_hierarchy")})) - elif action == "click": return call_unity("ui_click", {"window_title": args.get("window_title"), "element_name": args.get("element_name")}) - elif action == "input": return call_unity("ui_input_text", {"window_title": args.get("window_title"), "element_name": args.get("element_name"), "text": args.get("text")}) - else: return _invalid_action(action, ["list_windows", "get_hierarchy", "query", "get_window_rect", "set_window_rect", "capture_window_snapshot", "click", "input"]) - - elif name == "playerprefs_manager": - action = args.get("action") - if action == "get": return call_unity("get_player_pref", {"key": args.get("key"), "type": args.get("type", "string")}) - elif action == "set": return call_unity("set_player_pref", {"key": args.get("key"), "value": args.get("value"), "type": args.get("type", "string")}) - elif action == "delete": return call_unity("delete_player_pref", {"key": args.get("key")}) - elif action == "list": return call_unity("list_player_prefs") - else: return _invalid_action(action, ["get", "set", "delete", "list"]) - - elif name == "wait": - condition: Any = args.get("condition") - timeout: float = args.get("timeout_seconds", 60) - start_time: float = time.time() - status: str = "Ready" - - if condition == "compilation": - return _wait_for_compilation(timeout=timeout, start_time=start_time) - elif condition == "play_mode": - target_state = args.get("state", True) - while time.time() - start_time < timeout: - editor_state_response = call_unity("get_editor_state") - editor_state = _result_object(editor_state_response) - if editor_state_response and "result" in editor_state_response: - if editor_state.get("is_playing") == target_state: - break - time.sleep(1.0) - else: - status = "Timeout" - elif condition == "import": - while time.time() - start_time < timeout: - import_idle_response = call_unity("is_asset_import_idle") - import_idle_state = _result_object(import_idle_response) - if import_idle_response and "result" in import_idle_response: - if import_idle_state.get("is_idle"): - break - time.sleep(1.0) - else: - status = "Timeout" - elif condition == "editor_idle": - while time.time() - start_time < timeout: - editor_idle_response = call_unity("is_editor_idle") - editor_idle_state = _result_object(editor_idle_response) - if editor_idle_response and "result" in editor_idle_response: - if editor_idle_state.get("is_idle"): - break - time.sleep(1.0) - else: - status = "Timeout" - - wait_result: WaitResultPayload = { - "status": status, - "time_waited_seconds": round(time.time() - start_time, 2), - } - return {"result": wait_result} + status = "Timeout" - else: - return call_unity(name, args) + wait_result: WaitResultPayload = { + "status": status, + "time_waited_seconds": round(time.time() - start_time, 2), + } + return {"result": wait_result} + + +_HANDLERS: dict[str, RouteHandler] = { + "tools/list": _route_list_tools, + "list_tools": _route_list_tools, + "listTools": _route_list_tools, + "write_and_compile": _route_write_and_compile, + "scene_manager": _route_scene_manager, + "hierarchy_manager": _route_hierarchy_manager, + "component_manager": _route_component_manager, + "search_manager": _route_search_manager, + "asset_manager": _route_asset_manager, + "editor_controller": _route_editor_controller, + "ui_automation": _route_ui_automation, + "playerprefs_manager": _route_playerprefs_manager, + "wait": _route_wait, +} + + +def route_tool(name: str, args: JsonObject) -> JsonRpcResponse: + handler = _HANDLERS.get(name) + if handler is not None: + return handler(args) + return call_unity(name, args) diff --git a/Editor/tests/test_routing.py b/Editor/tests/test_routing.py index be05dc8..cb17ebe 100644 --- a/Editor/tests/test_routing.py +++ b/Editor/tests/test_routing.py @@ -4,7 +4,7 @@ import sys import unittest from typing import Any -from unittest.mock import call, patch +from unittest.mock import Mock, call, patch EDITOR_DIR: str = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if EDITOR_DIR not in sys.path: @@ -14,6 +14,24 @@ class RouteToolDispatchTests(unittest.TestCase): + def test_route_tool_uses_registered_handler(self) -> None: + handler = Mock(return_value={"result": {"handled": True}}) + + with patch.dict(routing._HANDLERS, {"synthetic": handler}, clear=True): + with patch("nexus_bridge.routing.call_unity") as mock_call_unity: + response: dict[str, Any] = routing.route_tool("synthetic", {"value": 7}) + + self.assertEqual({"result": {"handled": True}}, response) + handler.assert_called_once_with({"value": 7}) + mock_call_unity.assert_not_called() + + def test_route_tool_falls_back_to_call_unity_for_unknown_route(self) -> None: + with patch("nexus_bridge.routing.call_unity", return_value={"result": {"ok": True}}) as mock_call_unity: + response: dict[str, Any] = routing.route_tool("invoke_method", {"instance_id": 42}) + + self.assertEqual({"result": {"ok": True}}, response) + mock_call_unity.assert_called_once_with("invoke_method", {"instance_id": 42}) + def test_route_tool_dispatches_to_expected_unity_methods(self) -> None: test_cases: list[tuple[str, dict[str, Any], str, tuple[Any, ...]]] = [ ("scene_manager", {"action": "open", "path": "Assets/TestScene.unity"}, "open_scene", ({"path": "Assets/TestScene.unity"},)), @@ -49,6 +67,123 @@ def test_apply_created_transform_accepts_zero_instance_id(self) -> None: ) +class RouteHandlerTests(unittest.TestCase): + def test_scene_manager_create_alias_routes_to_create_scene(self) -> None: + with patch("nexus_bridge.routing.call_unity", return_value={"result": {"ok": True}}) as mock_call_unity: + response: dict[str, Any] = routing._route_scene_manager( + {"action": "create_scene", "name": "Arena", "open_if_exists": True} + ) + + self.assertEqual({"result": {"ok": True}}, response) + mock_call_unity.assert_called_once_with( + "create_scene", + {"name": "Arena", "open_if_exists": True}, + ) + + def test_scene_manager_invalid_action_returns_error(self) -> None: + response: dict[str, Any] = routing._route_scene_manager({"action": "delete"}) + + self.assertEqual(-32602, response["error"]["code"]) + self.assertIn("Invalid action: delete", response["error"]["message"]) + + def test_editor_controller_run_tests_wait_delegates_to_helper(self) -> None: + expected_response: dict[str, Any] = {"result": {"status": "Success"}} + + with patch("nexus_bridge.routing._run_tests_wait", return_value=expected_response) as mock_run_tests_wait: + response: dict[str, Any] = routing._route_editor_controller({"action": "run_tests_wait", "timeout_seconds": 9}) + + self.assertEqual(expected_response, response) + mock_run_tests_wait.assert_called_once_with({"action": "run_tests_wait", "timeout_seconds": 9}) + + def test_hierarchy_manager_create_applies_transform_after_create(self) -> None: + create_response: dict[str, Any] = {"result": {"data": {"instance_id": 12}}} + transform_response: dict[str, Any] = {"result": {"ok": True}} + + with patch("nexus_bridge.routing.call_unity", side_effect=[create_response, transform_response]) as mock_call_unity: + response: dict[str, Any] = routing._route_hierarchy_manager( + {"action": "create", "name": "Cube", "position": {"x": 1, "y": 2, "z": 3}} + ) + + self.assertEqual(create_response, response) + self.assertEqual( + [ + call("create_game_object", {"name": "Cube"}), + call("set_transform", {"instance_id": 12, "position": {"x": 1, "y": 2, "z": 3}}), + ], + mock_call_unity.call_args_list, + ) + + +class WriteAndCompileTests(unittest.TestCase): + def test_write_and_compile_returns_write_errors_without_waiting(self) -> None: + files = [ + {"path": "Assets/One.cs", "content": "one"}, + {"path": "Assets/Two.cs", "content": "two"}, + ] + call_results: list[dict[str, Any] | None] = [ + None, + {"error": {"code": -32000, "message": "disk full"}}, + {"result": {"ok": True}}, + ] + + with patch("nexus_bridge.routing.call_unity", side_effect=call_results) as mock_call_unity: + with patch("nexus_bridge.routing._wait_for_compilation") as mock_wait_for_compilation: + response: dict[str, Any] = routing._route_write_and_compile({"files": files}) + + self.assertEqual("Failed", response["result"]["status"]) + self.assertEqual("Failed to write some files", response["result"]["message"]) + self.assertEqual( + [{"path": "Assets/One.cs", "error": {"code": -32000, "message": "disk full"}}], + response["result"]["errors"], + ) + mock_wait_for_compilation.assert_not_called() + self.assertEqual( + [ + call("clear_logs"), + call("write_file", {"path": "Assets/One.cs", "content": "one"}), + call("write_file", {"path": "Assets/Two.cs", "content": "two"}), + ], + mock_call_unity.call_args_list, + ) + + def test_write_and_compile_waits_and_filters_compiler_errors(self) -> None: + files = [{"path": "Assets/Test.cs", "content": "class Test {}"}] + wait_response: dict[str, Any] = {"result": {"status": "Ready", "time_waited_seconds": 1.5}} + log_response: dict[str, Any] = { + "result": { + "logs": [ + {"Type": "Error", "Message": "compile failed"}, + {"Type": "Assert", "Message": "assertion"}, + {"Type": "Log", "Message": "ignore"}, + ] + } + } + + with patch("nexus_bridge.routing.time.time", return_value=10.0): + with patch("nexus_bridge.routing.call_unity", side_effect=[None, {"result": {"ok": True}}, log_response]) as mock_call_unity: + with patch("nexus_bridge.routing._wait_for_compilation", return_value=wait_response) as mock_wait_for_compilation: + response: dict[str, Any] = routing._route_write_and_compile({"files": files}) + + self.assertEqual("Failed", response["result"]["status"]) + self.assertEqual(1.5, response["result"]["time_waited_seconds"]) + self.assertEqual( + [ + {"Type": "Error", "Message": "compile failed"}, + {"Type": "Assert", "Message": "assertion"}, + ], + response["result"]["compiler_errors"], + ) + mock_wait_for_compilation.assert_called_once_with(timeout=90, start_time=10.0) + self.assertEqual( + [ + call("clear_logs"), + call("write_file", {"path": "Assets/Test.cs", "content": "class Test {}"}), + call("read_logs", {"count": 200}), + ], + mock_call_unity.call_args_list, + ) + + class RunTestsWaitTests(unittest.TestCase): def test_run_tests_wait_returns_success_when_new_results_appear(self) -> None: call_results: list[dict[str, Any]] = [ @@ -138,6 +273,27 @@ def test_wait_for_compilation_returns_timeout_after_refresh(self) -> None: ) mock_sleep.assert_called_once_with(0.5) + def test_wait_for_compilation_respects_timeout_smaller_than_reload_probe_window(self) -> None: + call_results: list[dict[str, Any]] = [ + {"result": {}}, + {"result": {}}, + ] + + with patch("nexus_bridge.routing.call_unity", side_effect=call_results) as mock_call_unity: + with patch("nexus_bridge.routing.time.time", side_effect=[100.0, 100.0, 105.1, 105.1, 106.1]): + with patch("nexus_bridge.routing.time.sleep") as mock_sleep: + response: dict[str, Any] = routing._wait_for_compilation(timeout=5.0) + + self.assertEqual({"result": {"status": "Timeout", "time_waited_seconds": 6.1}}, response) + self.assertEqual( + [ + call("initialize"), + call("refresh_asset_database"), + ], + mock_call_unity.call_args_list, + ) + mock_sleep.assert_called_once_with(0.5) + def test_wait_route_delegates_compilation_condition_to_helper(self) -> None: expected_response: dict[str, Any] = {"result": {"status": "Ready", "time_waited_seconds": 4.5}}