From 200483e826d00e25337138abc1bdd58441b3bb7d Mon Sep 17 00:00:00 2001 From: dsarno Date: Sun, 17 Aug 2025 08:59:42 -0700 Subject: [PATCH 1/6] Add initial transport handshake tests with plan placeholders --- tests/test_logging_stdout.py | 11 ++++ tests/test_resources_api.py | 11 ++++ tests/test_script_editing.py | 36 +++++++++++ tests/test_transport_framing.py | 102 ++++++++++++++++++++++++++++++++ 4 files changed, 160 insertions(+) create mode 100644 tests/test_logging_stdout.py create mode 100644 tests/test_resources_api.py create mode 100644 tests/test_script_editing.py create mode 100644 tests/test_transport_framing.py diff --git a/tests/test_logging_stdout.py b/tests/test_logging_stdout.py new file mode 100644 index 00000000..98dc23f4 --- /dev/null +++ b/tests/test_logging_stdout.py @@ -0,0 +1,11 @@ +import pytest + + +@pytest.mark.skip(reason="TODO: ensure server logs only to stderr and rotating file") +def test_no_stdout_output_from_tools(): + pass + + +@pytest.mark.skip(reason="TODO: sweep for accidental print statements in codebase") +def test_no_print_statements_in_codebase(): + pass diff --git a/tests/test_resources_api.py b/tests/test_resources_api.py new file mode 100644 index 00000000..bdcd7290 --- /dev/null +++ b/tests/test_resources_api.py @@ -0,0 +1,11 @@ +import pytest + + +@pytest.mark.skip(reason="TODO: resource.list returns only Assets/**/*.cs and rejects traversal") +def test_resource_list_filters_and_rejects_traversal(): + pass + + +@pytest.mark.skip(reason="TODO: resource.list rejects file:// paths outside project, including drive letters and symlinks") +def test_resource_list_rejects_outside_paths(): + pass diff --git a/tests/test_script_editing.py b/tests/test_script_editing.py new file mode 100644 index 00000000..e0b3705b --- /dev/null +++ b/tests/test_script_editing.py @@ -0,0 +1,36 @@ +import pytest + + +@pytest.mark.skip(reason="TODO: create new script, validate, apply edits, build and compile scene") +def test_script_edit_happy_path(): + pass + + +@pytest.mark.skip(reason="TODO: multiple micro-edits debounce to single compilation") +def test_micro_edits_debounce(): + pass + + +@pytest.mark.skip(reason="TODO: line ending variations handled correctly") +def test_line_endings_and_columns(): + pass + + +@pytest.mark.skip(reason="TODO: regex_replace no-op with allow_noop honored") +def test_regex_replace_noop_allowed(): + pass + + +@pytest.mark.skip(reason="TODO: large edit size boundaries and overflow protection") +def test_large_edit_size_and_overflow(): + pass + + +@pytest.mark.skip(reason="TODO: symlink and junction protections on edits") +def test_symlink_and_junction_protection(): + pass + + +@pytest.mark.skip(reason="TODO: atomic write guarantees") +def test_atomic_write_guarantees(): + pass diff --git a/tests/test_transport_framing.py b/tests/test_transport_framing.py new file mode 100644 index 00000000..1c3d02fa --- /dev/null +++ b/tests/test_transport_framing.py @@ -0,0 +1,102 @@ +import sys +import json +import struct +import socket +import threading +import time +from pathlib import Path + +import pytest + +# add server src to path +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "UnityMcpBridge" / "UnityMcpServer~" / "src" +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.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + sock.listen(1) + port = sock.getsockname()[1] + + def _run(): + conn, _ = sock.accept() + if greeting: + conn.sendall(greeting) + if respond_ping: + try: + header = conn.recv(8) + if len(header) == 8: + length = struct.unpack(">Q", header)[0] + payload = b"" + while len(payload) < length: + chunk = conn.recv(length - len(payload)) + if not chunk: + break + payload += chunk + if payload == b'{"type":"ping"}': + resp = b'{"type":"pong"}' + conn.sendall(struct.pack(">Q", len(resp)) + resp) + except Exception: + pass + time.sleep(0.1) + try: + conn.close() + finally: + sock.close() + + threading.Thread(target=_run, daemon=True).start() + return port + + +def test_handshake_requires_framing(): + port = start_dummy_server(b"MCP/0.1\n") + conn = UnityConnection(host="127.0.0.1", port=port) + assert conn.connect() is False + assert conn.sock is None + + +def test_small_frame_ping_pong(): + port = start_dummy_server(b"MCP/0.1 FRAMING=1\n", respond_ping=True) + conn = UnityConnection(host="127.0.0.1", port=port) + assert conn.connect() is True + assert conn.use_framing is True + payload = b'{"type":"ping"}' + conn.sock.sendall(struct.pack(">Q", len(payload)) + payload) + resp = conn.receive_full_response(conn.sock) + assert json.loads(resp.decode("utf-8"))["type"] == "pong" + conn.disconnect() + + +@pytest.mark.skip(reason="TODO: unframed data before reading greeting should disconnect") +def test_unframed_data_disconnect(): + pass + + +@pytest.mark.skip(reason="TODO: zero-length payload should raise error") +def test_zero_length_payload_error(): + pass + + +@pytest.mark.skip(reason="TODO: oversized payload should disconnect") +def test_oversized_payload_rejected(): + pass + + +@pytest.mark.skip(reason="TODO: partial header/payload triggers timeout and disconnect") +def test_partial_frame_timeout(): + pass + + +@pytest.mark.skip(reason="TODO: concurrency test with parallel tool invocations") +def test_parallel_invocations_no_interleaving(): + pass + + +@pytest.mark.skip(reason="TODO: reconnection after drop mid-command") +def test_reconnect_mid_command(): + pass From a3c81d657d5a2333a9e985643b0721cfe22dc6c5 Mon Sep 17 00:00:00 2001 From: dsarno Date: Sun, 17 Aug 2025 16:29:04 -0700 Subject: [PATCH 2/6] Fix dummy server startup and cleanup in transport tests --- tests/test_transport_framing.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_transport_framing.py b/tests/test_transport_framing.py index 1c3d02fa..50483f48 100644 --- a/tests/test_transport_framing.py +++ b/tests/test_transport_framing.py @@ -11,6 +11,8 @@ # add server src to path ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "UnityMcpBridge" / "UnityMcpServer~" / "src" +if not SRC.exists(): + raise FileNotFoundError(f"Server source directory not found: {SRC}") sys.path.insert(0, str(SRC)) from unity_connection import UnityConnection @@ -22,8 +24,10 @@ def start_dummy_server(greeting: bytes, respond_ping: bool = False): sock.bind(("127.0.0.1", 0)) sock.listen(1) port = sock.getsockname()[1] + ready = threading.Event() def _run(): + ready.set() conn, _ = sock.accept() if greeting: conn.sendall(greeting) @@ -46,10 +50,13 @@ def _run(): time.sleep(0.1) try: conn.close() + except Exception: + pass finally: sock.close() threading.Thread(target=_run, daemon=True).start() + ready.wait() return port From b01978c59e3e3a2589378a738fb6b764b05196fd Mon Sep 17 00:00:00 2001 From: dsarno Date: Sun, 17 Aug 2025 16:38:28 -0700 Subject: [PATCH 3/6] test: enforce no prints and handshake preamble --- README.md | 12 ++++++++++ tests/test_logging_stdout.py | 14 +++++++++-- tests/test_transport_framing.py | 42 +++++++++++++++++++++++++++++++-- 3 files changed, 64 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index d3c5c111..bb4dd965 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,18 @@ Unity MCP connects your tools using two components: --- +### Transport framing + +Unity MCP requires explicit framing negotiation. After connecting, the server +sends a `MCP/0.1` greeting. Clients must respond with `FRAMING=1`, and all +subsequent messages are sent as 8-byte big-endian length-prefixed JSON frames. + +### Resource URIs + +Assets are addressed using `unity://` URIs relative to the project root. For +example, `unity://path/Assets/Scripts/Foo.cs` refers to the file +`Assets/Scripts/Foo.cs` inside the Unity project. + ## Installation ⚙️ > **Note:** The setup is constantly improving as we update the package. Check back if you randomly start to run into issues. diff --git a/tests/test_logging_stdout.py b/tests/test_logging_stdout.py index 98dc23f4..d4389818 100644 --- a/tests/test_logging_stdout.py +++ b/tests/test_logging_stdout.py @@ -1,3 +1,6 @@ +import re +from pathlib import Path + import pytest @@ -6,6 +9,13 @@ def test_no_stdout_output_from_tools(): pass -@pytest.mark.skip(reason="TODO: sweep for accidental print statements in codebase") def test_no_print_statements_in_codebase(): - pass + """Ensure no stray print statements remain in server source.""" + src = Path(__file__).resolve().parents[1] / "UnityMcpBridge" / "UnityMcpServer~" / "src" + assert src.exists(), f"Server source directory not found: {src}" + offenders = [] + for py_file in src.rglob("*.py"): + text = py_file.read_text(encoding="utf-8") + if re.search(r"^\s*print\(", text, re.MULTILINE): + offenders.append(py_file.relative_to(src)) + assert not offenders, f"print statements found in: {offenders}" diff --git a/tests/test_transport_framing.py b/tests/test_transport_framing.py index 50483f48..602cb312 100644 --- a/tests/test_transport_framing.py +++ b/tests/test_transport_framing.py @@ -4,6 +4,7 @@ import socket import threading import time +import select from pathlib import Path import pytest @@ -60,6 +61,33 @@ def _run(): return port +def start_handshake_enforcing_server(): + """Server that drops connection if client sends data before handshake.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + sock.listen(1) + port = sock.getsockname()[1] + ready = threading.Event() + + def _run(): + ready.set() + conn, _ = sock.accept() + # if client sends any data before greeting, disconnect + r, _, _ = select.select([conn], [], [], 0.1) + if r: + conn.close() + sock.close() + return + conn.sendall(b"MCP/0.1 FRAMING=1\n") + time.sleep(0.1) + conn.close() + sock.close() + + threading.Thread(target=_run, daemon=True).start() + ready.wait() + return port + + def test_handshake_requires_framing(): port = start_dummy_server(b"MCP/0.1\n") conn = UnityConnection(host="127.0.0.1", port=port) @@ -79,9 +107,19 @@ def test_small_frame_ping_pong(): conn.disconnect() -@pytest.mark.skip(reason="TODO: unframed data before reading greeting should disconnect") def test_unframed_data_disconnect(): - pass + port = start_handshake_enforcing_server() + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect(("127.0.0.1", port)) + sock.sendall(b"BAD") + time.sleep(0.1) + try: + data = sock.recv(1024) + assert data == b"" + except ConnectionError: + pass + finally: + sock.close() @pytest.mark.skip(reason="TODO: zero-length payload should raise error") From 555d96510bd077e3a1dc8789a2426943e2e55ff4 Mon Sep 17 00:00:00 2001 From: dsarno Date: Sun, 17 Aug 2025 16:39:02 -0700 Subject: [PATCH 4/6] feat: add defensive server path resolution in tests --- tests/test_transport_framing.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/test_transport_framing.py b/tests/test_transport_framing.py index 602cb312..46601c3a 100644 --- a/tests/test_transport_framing.py +++ b/tests/test_transport_framing.py @@ -9,11 +9,18 @@ import pytest -# add server src to path +# locate server src dynamically to avoid hardcoded layout assumptions ROOT = Path(__file__).resolve().parents[1] -SRC = ROOT / "UnityMcpBridge" / "UnityMcpServer~" / "src" -if not SRC.exists(): - raise FileNotFoundError(f"Server source directory not found: {SRC}") +candidates = [ + ROOT / "UnityMcpBridge" / "UnityMcpServer~" / "src", + ROOT / "UnityMcpServer~" / "src", +] +SRC = next((p for p in candidates if p.exists()), None) +if SRC is None: + searched = "\n".join(str(p) for p in candidates) + raise FileNotFoundError( + "Unity MCP server source not found. Tried:\n" + searched + ) sys.path.insert(0, str(SRC)) from unity_connection import UnityConnection From e4544f68c3e0d271e9d9f48e5575b092e920fc0a Mon Sep 17 00:00:00 2001 From: dsarno Date: Sun, 17 Aug 2025 16:49:52 -0700 Subject: [PATCH 5/6] Refine server source path lookup --- README.md | 12 ------------ tests/test_logging_stdout.py | 20 ++++++++++++++++---- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index bb4dd965..d3c5c111 100644 --- a/README.md +++ b/README.md @@ -58,18 +58,6 @@ Unity MCP connects your tools using two components: --- -### Transport framing - -Unity MCP requires explicit framing negotiation. After connecting, the server -sends a `MCP/0.1` greeting. Clients must respond with `FRAMING=1`, and all -subsequent messages are sent as 8-byte big-endian length-prefixed JSON frames. - -### Resource URIs - -Assets are addressed using `unity://` URIs relative to the project root. For -example, `unity://path/Assets/Scripts/Foo.cs` refers to the file -`Assets/Scripts/Foo.cs` inside the Unity project. - ## Installation ⚙️ > **Note:** The setup is constantly improving as we update the package. Check back if you randomly start to run into issues. diff --git a/tests/test_logging_stdout.py b/tests/test_logging_stdout.py index d4389818..38e55d20 100644 --- a/tests/test_logging_stdout.py +++ b/tests/test_logging_stdout.py @@ -4,6 +4,20 @@ import pytest +# locate server src dynamically to avoid hardcoded layout assumptions +ROOT = Path(__file__).resolve().parents[1] +candidates = [ + ROOT / "UnityMcpBridge" / "UnityMcpServer~" / "src", + ROOT / "UnityMcpServer~" / "src", +] +SRC = next((p for p in candidates if p.exists()), None) +if SRC is None: + searched = "\n".join(str(p) for p in candidates) + raise FileNotFoundError( + "Unity MCP server source not found. Tried:\n" + searched + ) + + @pytest.mark.skip(reason="TODO: ensure server logs only to stderr and rotating file") def test_no_stdout_output_from_tools(): pass @@ -11,11 +25,9 @@ def test_no_stdout_output_from_tools(): def test_no_print_statements_in_codebase(): """Ensure no stray print statements remain in server source.""" - src = Path(__file__).resolve().parents[1] / "UnityMcpBridge" / "UnityMcpServer~" / "src" - assert src.exists(), f"Server source directory not found: {src}" offenders = [] - for py_file in src.rglob("*.py"): + for py_file in SRC.rglob("*.py"): text = py_file.read_text(encoding="utf-8") if re.search(r"^\s*print\(", text, re.MULTILINE): - offenders.append(py_file.relative_to(src)) + offenders.append(py_file.relative_to(SRC)) assert not offenders, f"print statements found in: {offenders}" From 9dbb4ffbcb5cb23a61f39d499b461d8fb82ad353 Mon Sep 17 00:00:00 2001 From: dsarno Date: Sun, 17 Aug 2025 17:02:36 -0700 Subject: [PATCH 6/6] Refine handshake tests and stdout hygiene --- tests/test_logging_stdout.py | 8 ++++++-- tests/test_transport_framing.py | 23 ++++++++++++++--------- 2 files changed, 20 insertions(+), 11 deletions(-) diff --git a/tests/test_logging_stdout.py b/tests/test_logging_stdout.py index 38e55d20..d6e728b7 100644 --- a/tests/test_logging_stdout.py +++ b/tests/test_logging_stdout.py @@ -28,6 +28,10 @@ def test_no_print_statements_in_codebase(): offenders = [] for py_file in SRC.rglob("*.py"): text = py_file.read_text(encoding="utf-8") - if re.search(r"^\s*print\(", text, re.MULTILINE): + if re.search(r"^\s*print\(", text, re.MULTILINE) or re.search( + r"sys\.stdout\.write\(", text + ): offenders.append(py_file.relative_to(SRC)) - assert not offenders, f"print statements found in: {offenders}" + assert not offenders, ( + "stdout writes found in: " + ", ".join(str(o) for o in offenders) + ) diff --git a/tests/test_transport_framing.py b/tests/test_transport_framing.py index 46601c3a..39e84afd 100644 --- a/tests/test_transport_framing.py +++ b/tests/test_transport_framing.py @@ -80,7 +80,8 @@ def _run(): ready.set() conn, _ = sock.accept() # if client sends any data before greeting, disconnect - r, _, _ = select.select([conn], [], [], 0.1) + # give clients a bit more time to send pre-handshake data before we greet + r, _, _ = select.select([conn], [], [], 0.2) if r: conn.close() sock.close() @@ -105,13 +106,15 @@ def test_handshake_requires_framing(): def test_small_frame_ping_pong(): port = start_dummy_server(b"MCP/0.1 FRAMING=1\n", respond_ping=True) conn = UnityConnection(host="127.0.0.1", port=port) - assert conn.connect() is True - assert conn.use_framing is True - payload = b'{"type":"ping"}' - conn.sock.sendall(struct.pack(">Q", len(payload)) + payload) - resp = conn.receive_full_response(conn.sock) - assert json.loads(resp.decode("utf-8"))["type"] == "pong" - conn.disconnect() + try: + assert conn.connect() is True + assert conn.use_framing is True + payload = b'{"type":"ping"}' + conn.sock.sendall(struct.pack(">Q", len(payload)) + payload) + resp = conn.receive_full_response(conn.sock) + assert json.loads(resp.decode("utf-8"))["type"] == "pong" + finally: + conn.disconnect() def test_unframed_data_disconnect(): @@ -123,7 +126,9 @@ def test_unframed_data_disconnect(): try: data = sock.recv(1024) assert data == b"" - except ConnectionError: + except (ConnectionResetError, ConnectionAbortedError): + # Some platforms raise instead of returning empty bytes when the + # server closes the connection after detecting pre-handshake data. pass finally: sock.close()