PHXH
/
unity-mcp
51
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Autoformat (#297)

main
Marcus Sanatan 2025-09-30 16:25:33 -04:00 committed by GitHub
parent 47ec46ce93
commit f6796e61f7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
58 changed files with 2393 additions and 2246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
.github/scripts/mark_skipped.py vendored
View File

@ -29,6 +29,7 @@ PATTERNS = [
r"validation error .* ctx",
]
def should_skip(msg: str) -> bool:
if not msg:
return False
@ -38,6 +39,7 @@ def should_skip(msg: str) -> bool:
return True
return False
def summarize_counts(ts: ET.Element):
tests = 0
failures = 0
@ -53,6 +55,7 @@ def summarize_counts(ts: ET.Element):
skipped += 1
return tests, failures, errors, skipped
def main(path: str) -> int:
if not os.path.exists(path):
print(f"[mark_skipped] No JUnit at {path}; nothing to do.")
@ -79,7 +82,8 @@ def main(path: str) -> int:
for n in nodes:
msg = (n.get("message") or "") + "\n" + (n.text or "")
if should_skip(msg):
first_match_text = (n.text or "").strip() or first_match_text
first_match_text = (
n.text or "").strip() or first_match_text
to_skip = True
if to_skip:
for n in nodes:
@ -98,12 +102,14 @@ def main(path: str) -> int:
if changed:
tree.write(path, encoding="utf-8", xml_declaration=True)
print(f"[mark_skipped] Updated {path}: converted environmental failures to skipped.")
print(
f"[mark_skipped] Updated {path}: converted environmental failures to skipped.")
else:
print(f"[mark_skipped] No environmental failures detected in {path}.")
return 0
if __name__ == "__main__":
target = (
sys.argv[1]

7
.github/workflows/claude-nl-suite.yml vendored
View File

@ -47,7 +47,7 @@ jobs:
# ---------- Python env for MCP server (uv) ----------
- uses: astral-sh/setup-uv@v4
with:
python-version: '3.11'
python-version: "3.11"
- name: Install MCP server
run: |
@ -366,7 +366,6 @@ jobs:
# (removed) Revert helper and baseline snapshot are no longer used
# ---------- Run suite in two passes ----------
- name: Run Claude NL pass
uses: anthropics/claude-code-base-action@beta
@ -389,7 +388,6 @@ jobs:
timeout_minutes: "30"
anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
- name: Run Claude T pass A-J
uses: anthropics/claude-code-base-action@beta
if: steps.detect.outputs.anthropic_ok == 'true' && steps.detect.outputs.unity_ok == 'true'
@ -932,7 +930,7 @@ jobs:
if: always()
uses: mikepenz/action-junit-report@v5
with:
report_paths: '${{ env.JUNIT_OUT }}'
report_paths: "${{ env.JUNIT_OUT }}"
include_passed: true
detailed_summary: true
annotate_notice: true
@ -966,4 +964,3 @@ jobs:
UNITY_EMAIL: ${{ secrets.UNITY_EMAIL }}
UNITY_PASSWORD: ${{ secrets.UNITY_PASSWORD }}
UNITY_SERIAL: ${{ secrets.UNITY_SERIAL }}

1
.github/workflows/github-repo-stats.yml vendored
View File

@ -17,4 +17,3 @@ jobs:
uses: jgehrcke/github-repo-stats@RELEASE
with:
ghtoken: ${{ secrets.ghrs_github_api_token }}

5
TestProjects/UnityMCPTests/Assets/Scripts/Hello.cs
View File

@ -3,13 +3,8 @@ using System.Collections;
public class Hello : MonoBehaviour
{
// Use this for initialization
void Start()
{
Debug.Log("Hello World");
}
}

2
TestProjects/UnityMCPTests/Assets/Scripts/LongUnityScriptClaudeTest.cs
View File

@ -2035,5 +2035,3 @@ public class LongUnityScriptClaudeTest : MonoBehaviour
#endregion
}

1
UnityMcpBridge/Editor/Data/DefaultServerConfig.cs
View File

@ -15,4 +15,3 @@ namespace MCPForUnity.Editor.Data
public new float retryDelay = 1.0f;
}
}

2
UnityMcpBridge/Editor/Helpers/ExecPath.cs
View File

@ -276,5 +276,3 @@ namespace MCPForUnity.Editor.Helpers
#endif
}
}

3
UnityMcpBridge/Editor/Helpers/GameObjectSerializer.cs
View File

@ -244,7 +244,8 @@ namespace MCPForUnity.Editor.Helpers
// Basic filtering (readable, not indexer, not transform which is handled elsewhere)
if (!propInfo.CanRead || propInfo.GetIndexParameters().Length > 0 || propInfo.Name == "transform") continue;
// Add if not already added (handles overrides - keep the most derived version)
if (!propertiesToCache.Any(p => p.Name == propInfo.Name)) {
if (!propertiesToCache.Any(p => p.Name == propInfo.Name))
{
propertiesToCache.Add(propInfo);
}
}

1
UnityMcpBridge/Editor/Helpers/McpConfigFileHelper.cs
View File

@ -184,4 +184,3 @@ namespace MCPForUnity.Editor.Helpers
}
}
}

2
UnityMcpBridge/Editor/Helpers/McpLog.cs
View File

@ -29,5 +29,3 @@ namespace MCPForUnity.Editor.Helpers
}
}
}

2
UnityMcpBridge/Editor/Helpers/PackageDetector.cs
View File

@ -105,5 +105,3 @@ namespace MCPForUnity.Editor.Helpers
}
}
}

1
UnityMcpBridge/Editor/Helpers/Response.cs
View File

@ -60,4 +60,3 @@ namespace MCPForUnity.Editor.Helpers
}
}
}

2
UnityMcpBridge/Editor/Helpers/ServerPathResolver.cs
View File

@ -147,5 +147,3 @@ namespace MCPForUnity.Editor.Helpers
}
}
}

1
UnityMcpBridge/Editor/Helpers/Vector3Helper.cs
View File

@ -22,4 +22,3 @@ namespace MCPForUnity.Editor.Helpers
}
}
}

3
UnityMcpBridge/Editor/MCPForUnityBridge.cs
View File

@ -151,7 +151,8 @@ namespace MCPForUnity.Editor
IoInfo($"[IO] ✗ write FAIL tag={item.Tag} reqId={(item.ReqId?.ToString() ?? "?")} {ex.GetType().Name}: {ex.Message}");
}
}
}) { IsBackground = true, Name = "MCP-Writer" };
})
{ IsBackground = true, Name = "MCP-Writer" };
writerThread.Start();
}
catch { }

1
UnityMcpBridge/Editor/Tools/CommandRegistry.cs
View File

@ -48,4 +48,3 @@ namespace MCPForUnity.Editor.Tools
}
}
}

33
UnityMcpBridge/Editor/Tools/ManageGameObject.cs
View File

@ -1716,7 +1716,8 @@ namespace MCPForUnity.Editor.Tools
propInfo.SetValue(target, convertedValue);
return true;
}
else {
else
{
Debug.LogWarning($"[SetProperty] Conversion failed for property '{memberName}' (Type: {propInfo.PropertyType.Name}) from token: {value.ToString(Formatting.None)}");
}
}
@ -1732,7 +1733,8 @@ namespace MCPForUnity.Editor.Tools
fieldInfo.SetValue(target, convertedValue);
return true;
}
else {
else
{
Debug.LogWarning($"[SetProperty] Conversion failed for field '{memberName}' (Type: {fieldInfo.FieldType.Name}) from token: {value.ToString(Formatting.None)}");
}
}
@ -1881,12 +1883,17 @@ namespace MCPForUnity.Editor.Tools
if (value is JArray jArray)
{
// Try converting to known types that SetColor/SetVector accept
if (jArray.Count == 4) {
if (jArray.Count == 4)
{
try { Color color = value.ToObject<Color>(inputSerializer); material.SetColor(finalPart, color); return true; } catch { }
try { Vector4 vec = value.ToObject<Vector4>(inputSerializer); material.SetVector(finalPart, vec); return true; } catch { }
} else if (jArray.Count == 3) {
}
else if (jArray.Count == 3)
{
try { Color color = value.ToObject<Color>(inputSerializer); material.SetColor(finalPart, color); return true; } catch { } // ToObject handles conversion to Color
} else if (jArray.Count == 2) {
}
else if (jArray.Count == 2)
{
try { Vector2 vec = value.ToObject<Vector2>(inputSerializer); material.SetVector(finalPart, vec); return true; } catch { }
}
}
@ -1901,13 +1908,16 @@ namespace MCPForUnity.Editor.Tools
else if (value.Type == JTokenType.String)
{
// Try converting to Texture using the serializer/converter
try {
try
{
Texture texture = value.ToObject<Texture>(inputSerializer);
if (texture != null) {
if (texture != null)
{
material.SetTexture(finalPart, texture);
return true;
}
} catch { }
}
catch { }
}
Debug.LogWarning(
@ -1927,7 +1937,8 @@ namespace MCPForUnity.Editor.Tools
finalPropInfo.SetValue(currentObject, convertedValue);
return true;
}
else {
else
{
Debug.LogWarning($"[SetNestedProperty] Final conversion failed for property '{finalPart}' (Type: {finalPropInfo.PropertyType.Name}) from token: {value.ToString(Formatting.None)}");
}
}
@ -1943,7 +1954,8 @@ namespace MCPForUnity.Editor.Tools
finalFieldInfo.SetValue(currentObject, convertedValue);
return true;
}
else {
else
{
Debug.LogWarning($"[SetNestedProperty] Final conversion failed for field '{finalPart}' (Type: {finalFieldInfo.FieldType.Name}) from token: {value.ToString(Formatting.None)}");
}
}
@ -2533,4 +2545,3 @@ namespace MCPForUnity.Editor.Tools
// They are now in Helpers.GameObjectSerializer
}
}

1
UnityMcpBridge/Editor/Tools/ManageScene.cs
View File

@ -472,4 +472,3 @@ namespace MCPForUnity.Editor.Tools
}
}
}

1
UnityMcpBridge/Editor/Tools/ManageScript.cs
View File

@ -2658,4 +2658,3 @@ static class ManageScriptRefreshHelpers
#endif
}
}

1
UnityMcpBridge/Editor/Tools/ReadConsole.cs
View File

@ -568,4 +568,3 @@ namespace MCPForUnity.Editor.Tools
*/
}
}

22
UnityMcpBridge/UnityMcpServer~/src/server.py
View File

@ -1,3 +1,4 @@
from telemetry import record_telemetry, record_milestone, RecordType, MilestoneType
from mcp.server.fastmcp import FastMCP
import logging
from logging.handlers import RotatingFileHandler
@ -21,10 +22,12 @@ logger = logging.getLogger("mcp-for-unity-server")
# Also write logs to a rotating file so logs are available when launched via stdio
try:
import os as _os
_log_dir = _os.path.join(_os.path.expanduser("~/Library/Application Support/UnityMCP"), "Logs")
_log_dir = _os.path.join(_os.path.expanduser(
"~/Library/Application Support/UnityMCP"), "Logs")
_os.makedirs(_log_dir, exist_ok=True)
_file_path = _os.path.join(_log_dir, "unity_mcp_server.log")
_fh = RotatingFileHandler(_file_path, maxBytes=512*1024, backupCount=2, encoding="utf-8")
_fh = RotatingFileHandler(
_file_path, maxBytes=512*1024, backupCount=2, encoding="utf-8")
_fh.setFormatter(logging.Formatter(config.log_format))
_fh.setLevel(getattr(logging, config.log_level))
logger.addHandler(_fh)
@ -42,7 +45,8 @@ except Exception:
# Quieten noisy third-party loggers to avoid clutter during stdio handshake
for noisy in ("httpx", "urllib3"):
try:
logging.getLogger(noisy).setLevel(max(logging.WARNING, getattr(logging, config.log_level)))
logging.getLogger(noisy).setLevel(
max(logging.WARNING, getattr(logging, config.log_level)))
except Exception:
pass
@ -50,13 +54,11 @@ for noisy in ("httpx", "urllib3"):
# Ensure a slightly higher telemetry timeout unless explicitly overridden by env
try:
# Ensure generous timeout unless explicitly overridden by env
if not os.environ.get("UNITY_MCP_TELEMETRY_TIMEOUT"):
os.environ["UNITY_MCP_TELEMETRY_TIMEOUT"] = "5.0"
except Exception:
pass
from telemetry import record_telemetry, record_milestone, RecordType, MilestoneType
# Global connection state
_unity_connection: UnityConnection = None
@ -79,6 +81,7 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]:
server_version = "unknown"
# Defer initial telemetry by 1s to avoid stdio handshake interference
import threading
def _emit_startup():
try:
record_telemetry(RecordType.STARTUP, {
@ -91,9 +94,11 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]:
threading.Timer(1.0, _emit_startup).start()
try:
skip_connect = os.environ.get("UNITY_MCP_SKIP_STARTUP_CONNECT", "").lower() in ("1", "true", "yes", "on")
skip_connect = os.environ.get(
"UNITY_MCP_SKIP_STARTUP_CONNECT", "").lower() in ("1", "true", "yes", "on")
if skip_connect:
logger.info("Skipping Unity connection on startup (UNITY_MCP_SKIP_STARTUP_CONNECT=1)")
logger.info(
"Skipping Unity connection on startup (UNITY_MCP_SKIP_STARTUP_CONNECT=1)")
else:
_unity_connection = get_unity_connection()
logger.info("Connected to Unity on startup")
@ -124,7 +129,8 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]:
}
)).start()
except Exception as e:
logger.warning("Unexpected error connecting to Unity on startup: %s", e)
logger.warning(
"Unexpected error connecting to Unity on startup: %s", e)
_unity_connection = None
import threading as _t
_err_msg = str(e)[:200]

24
mcp_source.py
View File

@ -32,7 +32,8 @@ def run_git(repo: pathlib.Path, *args: str) -> str:
"git", "-C", str(repo), *args
], capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip() or f"git {' '.join(args)} failed")
raise RuntimeError(result.stderr.strip()
or f"git {' '.join(args)} failed")
return result.stdout.strip()
@ -77,7 +78,8 @@ def find_manifest(explicit: Optional[str]) -> pathlib.Path:
candidate = parent / "Packages" / "manifest.json"
if candidate.exists():
return candidate
raise FileNotFoundError("Could not find Packages/manifest.json from current directory. Use --manifest to specify a path.")
raise FileNotFoundError(
"Could not find Packages/manifest.json from current directory. Use --manifest to specify a path.")
def read_json(path: pathlib.Path) -> dict:
@ -103,16 +105,21 @@ def build_options(repo_root: pathlib.Path, branch: str, origin_https: str):
origin_remote = origin
return [
("[1] Upstream main", upstream),
("[2] Remote current branch", f"{origin_remote}?path=/{BRIDGE_SUBPATH}#{branch}"),
("[3] Local workspace", f"file:{(repo_root / BRIDGE_SUBPATH).as_posix()}"),
("[2] Remote current branch",
f"{origin_remote}?path=/{BRIDGE_SUBPATH}#{branch}"),
("[3] Local workspace",
f"file:{(repo_root / BRIDGE_SUBPATH).as_posix()}"),
]
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Switch MCP for Unity package source")
p = argparse.ArgumentParser(
description="Switch MCP for Unity package source")
p.add_argument("--manifest", help="Path to Packages/manifest.json")
p.add_argument("--repo", help="Path to unity-mcp repo root (for local file option)")
p.add_argument("--choice", choices=["1", "2", "3"], help="Pick option non-interactively")
p.add_argument(
"--repo", help="Path to unity-mcp repo root (for local file option)")
p.add_argument(
"--choice", choices=["1", "2", "3"], help="Pick option non-interactively")
return p.parse_args()
@ -153,7 +160,8 @@ def main() -> None:
data = read_json(manifest_path)
deps = data.get("dependencies", {})
if PKG_NAME not in deps:
print(f"Error: '{PKG_NAME}' not found in manifest dependencies.", file=sys.stderr)
print(
f"Error: '{PKG_NAME}' not found in manifest dependencies.", file=sys.stderr)
sys.exit(1)
print(f"\nUpdating {PKG_NAME} → {chosen}")

15
test_unity_socket_framing.py
View File

@ -1,5 +1,8 @@
#!/usr/bin/env python3
import socket, struct, json, sys
import socket
import struct
import json
import sys
HOST = "127.0.0.1"
PORT = 6400
@ -10,6 +13,7 @@ except (IndexError, ValueError):
FILL = "R"
MAX_FRAME = 64 * 1024 * 1024
def recv_exact(sock, n):
buf = bytearray(n)
view = memoryview(buf)
@ -21,6 +25,7 @@ def recv_exact(sock, n):
off += r
return bytes(buf)
def is_valid_json(b):
try:
json.loads(b.decode("utf-8"))
@ -28,6 +33,7 @@ def is_valid_json(b):
except Exception:
return False
def recv_legacy_json(sock, timeout=60):
sock.settimeout(timeout)
chunks = []
@ -45,6 +51,7 @@ def recv_legacy_json(sock, timeout=60):
if is_valid_json(data):
return data
def main():
# Cap filler to stay within framing limit (reserve small overhead for JSON)
safe_max = max(1, MAX_FRAME - 4096)
@ -83,7 +90,8 @@ def main():
print(f"Response framed length: {resp_len}")
MAX_RESP = MAX_FRAME
if resp_len <= 0 or resp_len > MAX_RESP:
raise RuntimeError(f"invalid framed length: {resp_len} (max {MAX_RESP})")
raise RuntimeError(
f"invalid framed length: {resp_len} (max {MAX_RESP})")
resp = recv_exact(s, resp_len)
else:
s.sendall(body_bytes)
@ -92,7 +100,6 @@ def main():
print(f"Response bytes: {len(resp)}")
print(f"Response head: {resp[:120].decode('utf-8', 'ignore')}")
if __name__ == "__main__":
main()

1
tests/conftest.py
View File

@ -5,4 +5,3 @@ import os
os.environ.setdefault("DISABLE_TELEMETRY", "true")
os.environ.setdefault("UNITY_MCP_DISABLE_TELEMETRY", "true")
os.environ.setdefault("MCP_DISABLE_TELEMETRY", "true")

36
tests/test_edit_normalization_and_noop.py
View File

@ -12,7 +12,12 @@ sys.path.insert(0, str(SRC))
mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy: pass
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -21,22 +26,27 @@ sys.modules.setdefault("mcp", mcp_pkg)
sys.modules.setdefault("mcp.server", server_pkg)
sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
def _load(path: pathlib.Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
manage_script = _load(SRC / "tools" / "manage_script.py", "manage_script_mod2")
manage_script_edits = _load(SRC / "tools" / "manage_script_edits.py", "manage_script_edits_mod2")
manage_script_edits = _load(
SRC / "tools" / "manage_script_edits.py", "manage_script_edits_mod2")
class DummyMCP:
def __init__(self): self.tools = {}
def tool(self, *args, **kwargs):
def deco(fn): self.tools[fn.__name__] = fn; return fn
return deco
def setup_tools():
mcp = DummyMCP()
manage_script.register_manage_script_tools(mcp)
@ -59,7 +69,8 @@ def test_normalizes_lsp_and_index_ranges(monkeypatch):
"range": {"start": {"line": 10, "character": 2}, "end": {"line": 10, "character": 2}},
"newText": "// lsp\n"
}]
apply(None, uri="unity://path/Assets/Scripts/F.cs", edits=edits, precondition_sha256="x")
apply(None, uri="unity://path/Assets/Scripts/F.cs",
edits=edits, precondition_sha256="x")
p = calls[-1]
e = p["edits"][0]
assert e["startLine"] == 11 and e["startCol"] == 3
@ -68,12 +79,14 @@ def test_normalizes_lsp_and_index_ranges(monkeypatch):
calls.clear()
edits = [{"range": [0, 0], "text": "// idx\n"}]
# fake read to provide contents length
def fake_read(cmd, params):
if params.get("action") == "read":
return {"success": True, "data": {"contents": "hello\n"}}
return {"success": True}
monkeypatch.setattr(manage_script, "send_command_with_retry", fake_read)
apply(None, uri="unity://path/Assets/Scripts/F.cs", edits=edits, precondition_sha256="x")
apply(None, uri="unity://path/Assets/Scripts/F.cs",
edits=edits, precondition_sha256="x")
# last call is apply_text_edits
@ -81,11 +94,13 @@ def test_noop_evidence_shape(monkeypatch):
tools = setup_tools()
apply = tools["apply_text_edits"]
# Route response from Unity indicating no-op
def fake_send(cmd, params):
return {"success": True, "data": {"no_op": True, "evidence": {"reason": "identical_content"}}}
monkeypatch.setattr(manage_script, "send_command_with_retry", fake_send)
resp = apply(None, uri="unity://path/Assets/Scripts/F.cs", edits=[{"startLine":1,"startCol":1,"endLine":1,"endCol":1,"newText":""}], precondition_sha256="x")
resp = apply(None, uri="unity://path/Assets/Scripts/F.cs", edits=[
{"startLine": 1, "startCol": 1, "endLine": 1, "endCol": 1, "newText": ""}], precondition_sha256="x")
assert resp["success"] is True
assert resp.get("data", {}).get("no_op") is True
@ -93,9 +108,11 @@ def test_noop_evidence_shape(monkeypatch):
def test_atomic_multi_span_and_relaxed(monkeypatch):
tools_text = setup_tools()
apply_text = tools_text["apply_text_edits"]
tools_struct = DummyMCP(); manage_script_edits.register_manage_script_edits_tools(tools_struct)
tools_struct = DummyMCP()
manage_script_edits.register_manage_script_edits_tools(tools_struct)
# Fake send for read and write; verify atomic applyMode and validate=relaxed passes through
sent = {}
def fake_send(cmd, params):
if params.get("action") == "read":
return {"success": True, "data": {"contents": "public class C{\nvoid M(){ int x=2; }\n}\n"}}
@ -105,12 +122,13 @@ def test_atomic_multi_span_and_relaxed(monkeypatch):
edits = [
{"startLine": 2, "startCol": 14, "endLine": 2, "endCol": 15, "newText": "3"},
{"startLine": 3, "startCol": 2, "endLine": 3, "endCol": 2, "newText": "// tail\n"}
{"startLine": 3, "startCol": 2, "endLine": 3,
"endCol": 2, "newText": "// tail\n"}
]
resp = apply_text(None, uri="unity://path/Assets/Scripts/C.cs", edits=edits, precondition_sha256="sha", options={"validate": "relaxed", "applyMode": "atomic"})
resp = apply_text(None, uri="unity://path/Assets/Scripts/C.cs", edits=edits,
precondition_sha256="sha", options={"validate": "relaxed", "applyMode": "atomic"})
assert resp["success"] is True
# Last manage_script call should include options with applyMode atomic and validate relaxed
last = sent["calls"][-1]
assert last.get("options", {}).get("applyMode") == "atomic"
assert last.get("options", {}).get("validate") == "relaxed"

25
tests/test_edit_strict_and_warnings.py
View File

@ -12,7 +12,12 @@ sys.path.insert(0, str(SRC))
mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy: pass
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -34,6 +39,7 @@ manage_script = _load(SRC / "tools" / "manage_script.py", "manage_script_mod3")
class DummyMCP:
def __init__(self): self.tools = {}
def tool(self, *args, **kwargs):
def deco(fn): self.tools[fn.__name__] = fn; return fn
return deco
@ -56,13 +62,16 @@ def test_explicit_zero_based_normalized_warning(monkeypatch):
monkeypatch.setattr(manage_script, "send_command_with_retry", fake_send)
# Explicit fields given as 0-based (invalid); SDK should normalize and warn
edits = [{"startLine": 0, "startCol": 0, "endLine": 0, "endCol": 0, "newText": "//x"}]
resp = apply_edits(None, uri="unity://path/Assets/Scripts/F.cs", edits=edits, precondition_sha256="sha")
edits = [{"startLine": 0, "startCol": 0,
"endLine": 0, "endCol": 0, "newText": "//x"}]
resp = apply_edits(None, uri="unity://path/Assets/Scripts/F.cs",
edits=edits, precondition_sha256="sha")
assert resp["success"] is True
data = resp.get("data", {})
assert "normalizedEdits" in data
assert any(w == "zero_based_explicit_fields_normalized" for w in data.get("warnings", []))
assert any(
w == "zero_based_explicit_fields_normalized" for w in data.get("warnings", []))
ne = data["normalizedEdits"][0]
assert ne["startLine"] == 1 and ne["startCol"] == 1 and ne["endLine"] == 1 and ne["endCol"] == 1
@ -76,9 +85,9 @@ def test_strict_zero_based_error(monkeypatch):
monkeypatch.setattr(manage_script, "send_command_with_retry", fake_send)
edits = [{"startLine": 0, "startCol": 0, "endLine": 0, "endCol": 0, "newText": "//x"}]
resp = apply_edits(None, uri="unity://path/Assets/Scripts/F.cs", edits=edits, precondition_sha256="sha", strict=True)
edits = [{"startLine": 0, "startCol": 0,
"endLine": 0, "endCol": 0, "newText": "//x"}]
resp = apply_edits(None, uri="unity://path/Assets/Scripts/F.cs",
edits=edits, precondition_sha256="sha", strict=True)
assert resp["success"] is False
assert resp.get("code") == "zero_based_explicit_fields"

10
tests/test_find_in_file_minimal.py
View File

@ -1,3 +1,4 @@
from tools.resource_tools import register_resource_tools # type: ignore
import sys
import pathlib
import importlib.util
@ -9,7 +10,6 @@ ROOT = pathlib.Path(__file__).resolve().parents[1]
SRC = ROOT / "UnityMcpBridge" / "UnityMcpServer~" / "src"
sys.path.insert(0, str(SRC))
from tools.resource_tools import register_resource_tools # type: ignore
class DummyMCP:
def __init__(self):
@ -21,12 +21,14 @@ class DummyMCP:
return fn
return deco
@pytest.fixture()
def resource_tools():
mcp = DummyMCP()
register_resource_tools(mcp)
return mcp.tools
def test_find_in_file_returns_positions(resource_tools, tmp_path):
proj = tmp_path
assets = proj / "Assets"
@ -37,9 +39,11 @@ def test_find_in_file_returns_positions(resource_tools, tmp_path):
loop = asyncio.new_event_loop()
try:
resp = loop.run_until_complete(
find_in_file(uri="unity://path/Assets/A.txt", pattern="world", ctx=None, project_root=str(proj))
find_in_file(uri="unity://path/Assets/A.txt",
pattern="world", ctx=None, project_root=str(proj))
)
finally:
loop.close()
assert resp["success"] is True
assert resp["data"]["matches"] == [{"startLine": 1, "startCol": 7, "endLine": 1, "endCol": 12}]
assert resp["data"]["matches"] == [
{"startLine": 1, "startCol": 7, "endLine": 1, "endCol": 12}]

6
tests/test_get_sha.py
View File

@ -13,9 +13,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -32,7 +34,8 @@ def _load_module(path: pathlib.Path, name: str):
return mod
manage_script = _load_module(SRC / "tools" / "manage_script.py", "manage_script_mod")
manage_script = _load_module(
SRC / "tools" / "manage_script.py", "manage_script_mod")
class DummyMCP:
@ -72,4 +75,3 @@ def test_get_sha_param_shape_and_routing(monkeypatch):
assert captured["params"]["path"].endswith("Assets/Scripts")
assert resp["success"] is True
assert resp["data"] == {"sha256": "abc", "lengthBytes": 1}

29
tests/test_improved_anchor_matching.py
View File

@ -17,9 +17,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -28,13 +30,17 @@ sys.modules.setdefault("mcp", mcp_pkg)
sys.modules.setdefault("mcp.server", server_pkg)
sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
def load_module(path, name):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
manage_script_edits_module = load_module(SRC / "tools" / "manage_script_edits.py", "manage_script_edits_module")
manage_script_edits_module = load_module(
SRC / "tools" / "manage_script_edits.py", "manage_script_edits_module")
def test_improved_anchor_matching():
"""Test that our improved anchor matching finds the right closing brace."""
@ -69,7 +75,9 @@ public class TestClass : MonoBehaviour
match_pos = best_match.start()
line_num = test_code[:match_pos].count('\n') + 1
total_lines = test_code.count('\n') + 1
assert line_num >= total_lines - 2, f"expected match near end (>= {total_lines-2}), got line {line_num}"
assert line_num >= total_lines - \
2, f"expected match near end (>= {total_lines-2}), got line {line_num}"
def test_old_vs_new_matching():
"""Compare old vs new matching behavior."""
@ -104,18 +112,22 @@ public class TestClass : MonoBehaviour
# Old behavior (first match)
old_match = re.search(anchor_pattern, test_code, flags)
old_line = test_code[:old_match.start()].count('\n') + 1 if old_match else None
old_line = test_code[:old_match.start()].count(
'\n') + 1 if old_match else None
# New behavior (improved matching)
new_match = manage_script_edits_module._find_best_anchor_match(
anchor_pattern, test_code, flags, prefer_last=True
)
new_line = test_code[:new_match.start()].count('\n') + 1 if new_match else None
new_line = test_code[:new_match.start()].count(
'\n') + 1 if new_match else None
assert old_line is not None and new_line is not None, "failed to locate anchors"
assert new_line > old_line, f"improved matcher should choose a later line (old={old_line}, new={new_line})"
total_lines = test_code.count('\n') + 1
assert new_line >= total_lines - 2, f"expected class-end match near end (>= {total_lines-2}), got {new_line}"
assert new_line >= total_lines - \
2, f"expected class-end match near end (>= {total_lines-2}), got {new_line}"
def test_apply_edits_with_improved_matching():
"""Test that _apply_edits_locally uses improved matching."""
@ -140,14 +152,17 @@ public class TestClass : MonoBehaviour
"text": "\n public void NewMethod() { Debug.Log(\"Added at class end\"); }\n"
}]
result = manage_script_edits_module._apply_edits_locally(original_code, edits)
result = manage_script_edits_module._apply_edits_locally(
original_code, edits)
lines = result.split('\n')
try:
idx = next(i for i, line in enumerate(lines) if "NewMethod" in line)
except StopIteration:
assert False, "NewMethod not found in result"
total_lines = len(lines)
assert idx >= total_lines - 5, f"method inserted too early (idx={idx}, total_lines={total_lines})"
assert idx >= total_lines - \
5, f"method inserted too early (idx={idx}, total_lines={total_lines})"
if __name__ == "__main__":
print("Testing improved anchor matching...")

6
tests/test_logging_stdout.py
View File

@ -64,5 +64,7 @@ def test_no_print_statements_in_codebase():
v.visit(tree)
if v.hit:
offenders.append(py_file.relative_to(SRC))
assert not syntax_errors, "syntax errors in: " + ", ".join(str(e) for e in syntax_errors)
assert not offenders, "stdout writes found in: " + ", ".join(str(o) for o in offenders)
assert not syntax_errors, "syntax errors in: " + \
", ".join(str(e) for e in syntax_errors)
assert not offenders, "stdout writes found in: " + \
", ".join(str(o) for o in offenders)

24
tests/test_manage_script_uri.py
View File

@ -1,3 +1,4 @@
import tools.manage_script as manage_script # type: ignore
import sys
import types
from pathlib import Path
@ -5,7 +6,6 @@ from pathlib import Path
import pytest
# Locate server src dynamically to avoid hardcoded layout assumptions (same as other tests)
ROOT = Path(__file__).resolve().parents[1]
candidates = [
@ -25,7 +25,12 @@ sys.path.insert(0, str(SRC))
mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy: pass
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -36,7 +41,6 @@ sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
# Import target module after path injection
import tools.manage_script as manage_script # type: ignore
class DummyMCP:
@ -83,10 +87,13 @@ def test_split_uri_unity_path(monkeypatch):
@pytest.mark.parametrize(
"uri, expected_name, expected_path",
[
("file:///Users/alex/Project/Assets/Scripts/Foo%20Bar.cs", "Foo Bar", "Assets/Scripts"),
("file:///Users/alex/Project/Assets/Scripts/Foo%20Bar.cs",
"Foo Bar", "Assets/Scripts"),
("file://localhost/Users/alex/Project/Assets/Hello.cs", "Hello", "Assets"),
("file:///C:/Users/Alex/Proj/Assets/Scripts/Hello.cs", "Hello", "Assets/Scripts"),
("file:///tmp/Other.cs", "Other", "tmp"), # outside Assets → fall back to normalized dir
("file:///C:/Users/Alex/Proj/Assets/Scripts/Hello.cs",
"Hello", "Assets/Scripts"),
# outside Assets → fall back to normalized dir
("file:///tmp/Other.cs", "Other", "tmp"),
],
)
def test_split_uri_file_urls(monkeypatch, uri, expected_name, expected_path):
@ -118,9 +125,8 @@ def test_split_uri_plain_path(monkeypatch):
monkeypatch.setattr(manage_script, "send_command_with_retry", fake_send)
fn = tools['apply_text_edits']
fn(DummyCtx(), uri="Assets/Scripts/Thing.cs", edits=[], precondition_sha256=None)
fn(DummyCtx(), uri="Assets/Scripts/Thing.cs",
edits=[], precondition_sha256=None)
assert captured['params']['name'] == 'Thing'
assert captured['params']['path'] == 'Assets/Scripts'

19
tests/test_read_console_truncate.py
View File

@ -12,9 +12,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -23,13 +25,17 @@ sys.modules.setdefault("mcp", mcp_pkg)
sys.modules.setdefault("mcp.server", server_pkg)
sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
def _load_module(path: pathlib.Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
read_console_mod = _load_module(SRC / "tools" / "read_console.py", "read_console_mod")
read_console_mod = _load_module(
SRC / "tools" / "read_console.py", "read_console_mod")
class DummyMCP:
def __init__(self):
@ -41,11 +47,13 @@ class DummyMCP:
return fn
return deco
def setup_tools():
mcp = DummyMCP()
read_console_mod.register_read_console_tools(mcp)
return mcp.tools
def test_read_console_full_default(monkeypatch):
tools = setup_tools()
read_console = tools["read_console"]
@ -60,7 +68,8 @@ def test_read_console_full_default(monkeypatch):
}
monkeypatch.setattr(read_console_mod, "send_command_with_retry", fake_send)
monkeypatch.setattr(read_console_mod, "get_unity_connection", lambda: object())
monkeypatch.setattr(
read_console_mod, "get_unity_connection", lambda: object())
resp = read_console(ctx=None, count=10)
assert resp == {
@ -85,8 +94,10 @@ def test_read_console_truncated(monkeypatch):
}
monkeypatch.setattr(read_console_mod, "send_command_with_retry", fake_send)
monkeypatch.setattr(read_console_mod, "get_unity_connection", lambda: object())
monkeypatch.setattr(
read_console_mod, "get_unity_connection", lambda: object())
resp = read_console(ctx=None, count=10, include_stacktrace=False)
assert resp == {"success": True, "data": {"lines": [{"level": "error", "message": "oops"}]}}
assert resp == {"success": True, "data": {
"lines": [{"level": "error", "message": "oops"}]}}
assert captured["params"]["includeStacktrace"] is False

8
tests/test_read_resource_minimal.py
View File

@ -1,3 +1,4 @@
from tools.resource_tools import register_resource_tools # type: ignore
import sys
import pathlib
import asyncio
@ -13,9 +14,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -24,8 +27,6 @@ sys.modules.setdefault("mcp", mcp_pkg)
sys.modules.setdefault("mcp.server", server_pkg)
sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
from tools.resource_tools import register_resource_tools # type: ignore
class DummyMCP:
def __init__(self):
@ -57,7 +58,8 @@ def test_read_resource_minimal_metadata_only(resource_tools, tmp_path):
loop = asyncio.new_event_loop()
try:
resp = loop.run_until_complete(
read_resource(uri="unity://path/Assets/A.txt", ctx=None, project_root=str(proj))
read_resource(uri="unity://path/Assets/A.txt",
ctx=None, project_root=str(proj))
)
finally:
loop.close()

13
tests/test_resources_api.py
View File

@ -1,3 +1,4 @@
from tools.resource_tools import register_resource_tools # type: ignore
import pytest
@ -21,17 +22,18 @@ if SRC is None:
)
sys.path.insert(0, str(SRC))
from tools.resource_tools import register_resource_tools # type: ignore
class DummyMCP:
def __init__(self):
self._tools = {}
def tool(self, *args, **kwargs): # accept kwargs like description
def deco(fn):
self._tools[fn.__name__] = fn
return fn
return deco
@pytest.fixture()
def resource_tools():
mcp = DummyMCP()
@ -60,7 +62,8 @@ def test_resource_list_filters_and_rejects_traversal(resource_tools, tmp_path, m
# Only .cs under Assets should be listed
import asyncio
resp = asyncio.get_event_loop().run_until_complete(
list_resources(ctx=None, pattern="*.cs", under="Assets", limit=50, project_root=str(proj))
list_resources(ctx=None, pattern="*.cs", under="Assets",
limit=50, project_root=str(proj))
)
assert resp["success"] is True
uris = resp["data"]["uris"]
@ -75,7 +78,9 @@ def test_resource_list_rejects_outside_paths(resource_tools, tmp_path):
list_resources = resource_tools["list_resources"]
import asyncio
resp = asyncio.get_event_loop().run_until_complete(
list_resources(ctx=None, pattern="*.cs", under="..", limit=10, project_root=str(proj))
list_resources(ctx=None, pattern="*.cs", under="..",
limit=10, project_root=str(proj))
)
assert resp["success"] is False
assert "Assets" in resp.get("error", "") or "under project root" in resp.get("error", "")
assert "Assets" in resp.get(
"error", "") or "under project root" in resp.get("error", "")

54
tests/test_script_tools.py
View File

@ -15,9 +15,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -26,14 +28,18 @@ sys.modules.setdefault("mcp", mcp_pkg)
sys.modules.setdefault("mcp.server", server_pkg)
sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
def load_module(path, name):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
manage_script_module = load_module(SRC / "tools" / "manage_script.py", "manage_script_module")
manage_asset_module = load_module(SRC / "tools" / "manage_asset.py", "manage_asset_module")
manage_script_module = load_module(
SRC / "tools" / "manage_script.py", "manage_script_module")
manage_asset_module = load_module(
SRC / "tools" / "manage_asset.py", "manage_asset_module")
class DummyMCP:
@ -46,16 +52,19 @@ class DummyMCP:
return func
return decorator
def setup_manage_script():
mcp = DummyMCP()
manage_script_module.register_manage_script_tools(mcp)
return mcp.tools
def setup_manage_asset():
mcp = DummyMCP()
manage_asset_module.register_manage_asset_tools(mcp)
return mcp.tools
def test_apply_text_edits_long_file(monkeypatch):
tools = setup_manage_script()
apply_edits = tools["apply_text_edits"]
@ -66,15 +75,18 @@ def test_apply_text_edits_long_file(monkeypatch):
captured["params"] = params
return {"success": True}
monkeypatch.setattr(manage_script_module, "send_command_with_retry", fake_send)
monkeypatch.setattr(manage_script_module,
"send_command_with_retry", fake_send)
edit = {"startLine": 1005, "startCol": 0, "endLine": 1005, "endCol": 5, "newText": "Hello"}
edit = {"startLine": 1005, "startCol": 0,
"endLine": 1005, "endCol": 5, "newText": "Hello"}
resp = apply_edits(None, "unity://path/Assets/Scripts/LongFile.cs", [edit])
assert captured["cmd"] == "manage_script"
assert captured["params"]["action"] == "apply_text_edits"
assert captured["params"]["edits"][0]["startLine"] == 1005
assert resp["success"] is True
def test_sequential_edits_use_precondition(monkeypatch):
tools = setup_manage_script()
apply_edits = tools["apply_text_edits"]
@ -84,12 +96,16 @@ def test_sequential_edits_use_precondition(monkeypatch):
calls.append(params)
return {"success": True, "sha256": f"hash{len(calls)}"}
monkeypatch.setattr(manage_script_module, "send_command_with_retry", fake_send)
monkeypatch.setattr(manage_script_module,
"send_command_with_retry", fake_send)
edit1 = {"startLine": 1, "startCol": 0, "endLine": 1, "endCol": 0, "newText": "//header\n"}
edit1 = {"startLine": 1, "startCol": 0, "endLine": 1,
"endCol": 0, "newText": "//header\n"}
resp1 = apply_edits(None, "unity://path/Assets/Scripts/File.cs", [edit1])
edit2 = {"startLine": 2, "startCol": 0, "endLine": 2, "endCol": 0, "newText": "//second\n"}
resp2 = apply_edits(None, "unity://path/Assets/Scripts/File.cs", [edit2], precondition_sha256=resp1["sha256"])
edit2 = {"startLine": 2, "startCol": 0, "endLine": 2,
"endCol": 0, "newText": "//second\n"}
resp2 = apply_edits(None, "unity://path/Assets/Scripts/File.cs",
[edit2], precondition_sha256=resp1["sha256"])
assert calls[1]["precondition_sha256"] == resp1["sha256"]
assert resp2["sha256"] == "hash2"
@ -104,10 +120,12 @@ def test_apply_text_edits_forwards_options(monkeypatch):
captured["params"] = params
return {"success": True}
monkeypatch.setattr(manage_script_module, "send_command_with_retry", fake_send)
monkeypatch.setattr(manage_script_module,
"send_command_with_retry", fake_send)
opts = {"validate": "relaxed", "applyMode": "atomic", "refresh": "immediate"}
apply_edits(None, "unity://path/Assets/Scripts/File.cs", [{"startLine":1,"startCol":1,"endLine":1,"endCol":1,"newText":"x"}], options=opts)
apply_edits(None, "unity://path/Assets/Scripts/File.cs",
[{"startLine": 1, "startCol": 1, "endLine": 1, "endCol": 1, "newText": "x"}], options=opts)
assert captured["params"].get("options") == opts
@ -120,16 +138,20 @@ def test_apply_text_edits_defaults_atomic_for_multi_span(monkeypatch):
captured["params"] = params
return {"success": True}
monkeypatch.setattr(manage_script_module, "send_command_with_retry", fake_send)
monkeypatch.setattr(manage_script_module,
"send_command_with_retry", fake_send)
edits = [
{"startLine": 2, "startCol": 2, "endLine": 2, "endCol": 3, "newText": "A"},
{"startLine": 3, "startCol": 2, "endLine": 3, "endCol": 2, "newText": "// tail\n"},
{"startLine": 3, "startCol": 2, "endLine": 3,
"endCol": 2, "newText": "// tail\n"},
]
apply_edits(None, "unity://path/Assets/Scripts/File.cs", edits, precondition_sha256="x")
apply_edits(None, "unity://path/Assets/Scripts/File.cs",
edits, precondition_sha256="x")
opts = captured["params"].get("options", {})
assert opts.get("applyMode") == "atomic"
def test_manage_asset_prefab_modify_request(monkeypatch):
tools = setup_manage_asset()
manage_asset = tools["manage_asset"]
@ -140,8 +162,10 @@ def test_manage_asset_prefab_modify_request(monkeypatch):
captured["params"] = params
return {"success": True}
monkeypatch.setattr(manage_asset_module, "async_send_command_with_retry", fake_async)
monkeypatch.setattr(manage_asset_module, "get_unity_connection", lambda: object())
monkeypatch.setattr(manage_asset_module,
"async_send_command_with_retry", fake_async)
monkeypatch.setattr(manage_asset_module,
"get_unity_connection", lambda: object())
async def run():
resp = await manage_asset(

19
tests/test_telemetry_endpoint_validation.py
View File

@ -1,18 +1,21 @@
import os
import importlib
def test_endpoint_rejects_non_http(tmp_path, monkeypatch):
# Point data dir to temp to avoid touching real files
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path))
monkeypatch.setenv("UNITY_MCP_TELEMETRY_ENDPOINT", "file:///etc/passwd")
telemetry = importlib.import_module("UnityMcpBridge.UnityMcpServer~.src.telemetry")
telemetry = importlib.import_module(
"UnityMcpBridge.UnityMcpServer~.src.telemetry")
importlib.reload(telemetry)
tc = telemetry.TelemetryCollector()
# Should have fallen back to default endpoint
assert tc.config.endpoint == tc.config.default_endpoint
def test_config_preferred_then_env_override(tmp_path, monkeypatch):
# Simulate config telemetry endpoint
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path))
@ -20,27 +23,32 @@ def test_config_preferred_then_env_override(tmp_path, monkeypatch):
# Patch config.telemetry_endpoint via import mocking
import importlib
cfg_mod = importlib.import_module("UnityMcpBridge.UnityMcpServer~.src.config")
cfg_mod = importlib.import_module(
"UnityMcpBridge.UnityMcpServer~.src.config")
old_endpoint = cfg_mod.config.telemetry_endpoint
cfg_mod.config.telemetry_endpoint = "https://example.com/telemetry"
try:
telemetry = importlib.import_module("UnityMcpBridge.UnityMcpServer~.src.telemetry")
telemetry = importlib.import_module(
"UnityMcpBridge.UnityMcpServer~.src.telemetry")
importlib.reload(telemetry)
tc = telemetry.TelemetryCollector()
assert tc.config.endpoint == "https://example.com/telemetry"
# Env should override config
monkeypatch.setenv("UNITY_MCP_TELEMETRY_ENDPOINT", "https://override.example/ep")
monkeypatch.setenv("UNITY_MCP_TELEMETRY_ENDPOINT",
"https://override.example/ep")
importlib.reload(telemetry)
tc2 = telemetry.TelemetryCollector()
assert tc2.config.endpoint == "https://override.example/ep"
finally:
cfg_mod.config.telemetry_endpoint = old_endpoint
def test_uuid_preserved_on_malformed_milestones(tmp_path, monkeypatch):
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path))
telemetry = importlib.import_module("UnityMcpBridge.UnityMcpServer~.src.telemetry")
telemetry = importlib.import_module(
"UnityMcpBridge.UnityMcpServer~.src.telemetry")
importlib.reload(telemetry)
tc1 = telemetry.TelemetryCollector()
@ -53,4 +61,3 @@ def test_uuid_preserved_on_malformed_milestones(tmp_path, monkeypatch):
importlib.reload(telemetry)
tc2 = telemetry.TelemetryCollector()
assert tc2._customer_uuid == first_uuid

10
tests/test_telemetry_queue_worker.py
View File

@ -16,9 +16,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -72,12 +74,12 @@ def test_telemetry_queue_backpressure_and_single_worker(monkeypatch, caplog):
time.sleep(0.3)
# Verify drops were logged (queue full backpressure)
dropped_logs = [m for m in caplog.messages if "Telemetry queue full; dropping" in m]
dropped_logs = [
m for m in caplog.messages if "Telemetry queue full; dropping" in m]
assert len(dropped_logs) >= 1
# Ensure only one worker thread exists and is alive
assert collector._worker.is_alive()
worker_threads = [t for t in threading.enumerate() if t is collector._worker]
worker_threads = [
t for t in threading.enumerate() if t is collector._worker]
assert len(worker_threads) == 1

5
tests/test_telemetry_subaction.py
View File

@ -3,7 +3,8 @@ import importlib
def _get_decorator_module():
# Import the telemetry_decorator module from the Unity MCP server src
mod = importlib.import_module("UnityMcpBridge.UnityMcpServer~.src.telemetry_decorator")
mod = importlib.import_module(
"UnityMcpBridge.UnityMcpServer~.src.telemetry_decorator")
return mod
@ -79,5 +80,3 @@ def test_subaction_none_when_not_present(monkeypatch):
_ = wrapped(None, name="X")
assert captured["tool_name"] == "apply_text_edits"
assert captured["sub_action"] is None

14
tests/test_transport_framing.py
View File

@ -1,3 +1,4 @@
from unity_connection import UnityConnection
import sys
import json
import struct
@ -24,8 +25,6 @@ if SRC is None:
)
sys.path.insert(0, str(SRC))
from unity_connection import UnityConnection
def start_dummy_server(greeting: bytes, respond_ping: bool = False):
"""Start a minimal TCP server for handshake tests."""
@ -159,7 +158,10 @@ def test_unframed_data_disconnect():
def test_zero_length_payload_heartbeat():
# Server that sends handshake and a zero-length heartbeat frame followed by a pong payload
import socket, struct, threading, time
import socket
import struct
import threading
import time
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 0))
@ -181,8 +183,10 @@ def test_zero_length_payload_heartbeat():
conn.sendall(struct.pack(">Q", len(payload)) + payload)
time.sleep(0.02)
finally:
try: conn.close()
except Exception: pass
try:
conn.close()
except Exception:
pass
sock.close()
threading.Thread(target=_run, daemon=True).start()

10
tests/test_validate_script_summary.py
View File

@ -12,9 +12,11 @@ mcp_pkg = types.ModuleType("mcp")
server_pkg = types.ModuleType("mcp.server")
fastmcp_pkg = types.ModuleType("mcp.server.fastmcp")
class _Dummy:
pass
fastmcp_pkg.FastMCP = _Dummy
fastmcp_pkg.Context = _Dummy
server_pkg.fastmcp = fastmcp_pkg
@ -23,13 +25,17 @@ sys.modules.setdefault("mcp", mcp_pkg)
sys.modules.setdefault("mcp.server", server_pkg)
sys.modules.setdefault("mcp.server.fastmcp", fastmcp_pkg)
def _load_module(path: pathlib.Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
manage_script = _load_module(SRC / "tools" / "manage_script.py", "manage_script_mod")
manage_script = _load_module(
SRC / "tools" / "manage_script.py", "manage_script_mod")
class DummyMCP:
def __init__(self):
@ -41,11 +47,13 @@ class DummyMCP:
return fn
return deco
def setup_tools():
mcp = DummyMCP()
manage_script.register_manage_script_tools(mcp)
return mcp.tools
def test_validate_script_returns_counts(monkeypatch):
tools = setup_tools()
validate_script = tools["validate_script"]

59
tools/stress_mcp.py
View File

@ -21,7 +21,8 @@ def dlog(*args):
def find_status_files() -> list[Path]:
home = Path.home()
status_dir = Path(os.environ.get("UNITY_MCP_STATUS_DIR", home / ".unity-mcp"))
status_dir = Path(os.environ.get(
"UNITY_MCP_STATUS_DIR", home / ".unity-mcp"))
if not status_dir.exists():
return []
return sorted(status_dir.glob("unity-mcp-status-*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
@ -87,7 +88,8 @@ def make_ping_frame() -> bytes:
def make_execute_menu_item(menu_path: str) -> bytes:
# Retained for manual debugging; not used in normal stress runs
payload = {"type": "execute_menu_item", "params": {"action": "execute", "menu_path": menu_path}}
payload = {"type": "execute_menu_item", "params": {
"action": "execute", "menu_path": menu_path}}
return json.dumps(payload).encode("utf-8")
@ -102,7 +104,8 @@ async def client_loop(idx: int, host: str, port: int, stop_time: float, stats: d
await asyncio.wait_for(do_handshake(reader), timeout=TIMEOUT)
# Send a quick ping first
await write_frame(writer, make_ping_frame())
_ = await asyncio.wait_for(read_frame(reader), timeout=TIMEOUT) # ignore content
# ignore content
_ = await asyncio.wait_for(read_frame(reader), timeout=TIMEOUT)
# Main activity loop (keep-alive + light load). Edit spam handled by reload_churn_task.
while time.time() < stop_time:
@ -182,7 +185,8 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
if relative:
# Derive name and directory for ManageScript and compute precondition SHA + EOF position
name_base = Path(relative).stem
dir_path = str(Path(relative).parent).replace('\\', '/')
dir_path = str(
Path(relative).parent).replace('\\', '/')
# 1) Read current contents via manage_script.read to compute SHA and true EOF location
contents = None
@ -203,8 +207,10 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
await write_frame(writer, json.dumps(read_payload).encode("utf-8"))
resp = await asyncio.wait_for(read_frame(reader), timeout=TIMEOUT)
read_obj = json.loads(resp.decode("utf-8", errors="ignore"))
result = read_obj.get("result", read_obj) if isinstance(read_obj, dict) else {}
read_obj = json.loads(
resp.decode("utf-8", errors="ignore"))
result = read_obj.get("result", read_obj) if isinstance(
read_obj, dict) else {}
if result.get("success"):
data_obj = result.get("data", {})
contents = data_obj.get("contents") or ""
@ -222,13 +228,15 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
pass
if not read_success or contents is None:
stats["apply_errors"] = stats.get("apply_errors", 0) + 1
stats["apply_errors"] = stats.get(
"apply_errors", 0) + 1
await asyncio.sleep(0.5)
continue
# Compute SHA and EOF insertion point
import hashlib
sha = hashlib.sha256(contents.encode("utf-8")).hexdigest()
sha = hashlib.sha256(
contents.encode("utf-8")).hexdigest()
lines = contents.splitlines(keepends=True)
# Insert at true EOF (safe against header guards)
end_line = len(lines) + 1 # 1-based exclusive end
@ -237,7 +245,8 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
# Build a unique marker append; ensure it begins with a newline if needed
marker = f"// MCP_STRESS seq={seq} time={int(time.time())}"
seq += 1
insert_text = ("\n" if not contents.endswith("\n") else "") + marker + "\n"
insert_text = ("\n" if not contents.endswith(
"\n") else "") + marker + "\n"
# 2) Apply text edits with immediate refresh and precondition
apply_payload = {
@ -269,11 +278,14 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
await write_frame(writer, json.dumps(apply_payload).encode("utf-8"))
resp = await asyncio.wait_for(read_frame(reader), timeout=TIMEOUT)
try:
data = json.loads(resp.decode("utf-8", errors="ignore"))
result = data.get("result", data) if isinstance(data, dict) else {}
data = json.loads(resp.decode(
"utf-8", errors="ignore"))
result = data.get("result", data) if isinstance(
data, dict) else {}
ok = bool(result.get("success", False))
if ok:
stats["applies"] = stats.get("applies", 0) + 1
stats["applies"] = stats.get(
"applies", 0) + 1
apply_success = True
break
except Exception:
@ -290,7 +302,8 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
except Exception:
pass
if not apply_success:
stats["apply_errors"] = stats.get("apply_errors", 0) + 1
stats["apply_errors"] = stats.get(
"apply_errors", 0) + 1
except Exception:
pass
@ -298,13 +311,17 @@ async def reload_churn_task(project_path: str, stop_time: float, unity_file: str
async def main():
ap = argparse.ArgumentParser(description="Stress test the Unity MCP bridge with concurrent clients and reload churn")
ap = argparse.ArgumentParser(
description="Stress test the Unity MCP bridge with concurrent clients and reload churn")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--project", default=str(Path(__file__).resolve().parents[1] / "TestProjects" / "UnityMCPTests"))
ap.add_argument("--unity-file", default=str(Path(__file__).resolve().parents[1] / "TestProjects" / "UnityMCPTests" / "Assets" / "Scripts" / "LongUnityScriptClaudeTest.cs"))
ap.add_argument("--project", default=str(
Path(__file__).resolve().parents[1] / "TestProjects" / "UnityMCPTests"))
ap.add_argument("--unity-file", default=str(Path(__file__).resolve(
).parents[1] / "TestProjects" / "UnityMCPTests" / "Assets" / "Scripts" / "LongUnityScriptClaudeTest.cs"))
ap.add_argument("--clients", type=int, default=10)
ap.add_argument("--duration", type=int, default=60)
ap.add_argument("--storm-count", type=int, default=1, help="Number of scripts to touch each cycle")
ap.add_argument("--storm-count", type=int, default=1,
help="Number of scripts to touch each cycle")
args = ap.parse_args()
port = discover_port(args.project)
@ -315,10 +332,12 @@ async def main():
# Spawn clients
for i in range(max(1, args.clients)):
tasks.append(asyncio.create_task(client_loop(i, args.host, port, stop_time, stats)))
tasks.append(asyncio.create_task(
client_loop(i, args.host, port, stop_time, stats)))
# Spawn reload churn task
tasks.append(asyncio.create_task(reload_churn_task(args.project, stop_time, args.unity_file, args.host, port, stats, storm_count=args.storm_count)))
tasks.append(asyncio.create_task(reload_churn_task(args.project, stop_time,
args.unity_file, args.host, port, stats, storm_count=args.storm_count)))
await asyncio.gather(*tasks, return_exceptions=True)
print(json.dumps({"port": port, "stats": stats}, indent=2))
@ -329,5 +348,3 @@ if __name__ == "__main__":
asyncio.run(main())
except KeyboardInterrupt:
pass