Add initial transport handshake tests with plan placeholders

tests/test_logging_stdout.py

@@ -0,0 +1,37 @@
+import re
+from pathlib import Path
+
+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
+
+
+def test_no_print_statements_in_codebase():
+    """Ensure no stray print statements remain in server source."""
+    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) or re.search(
+            r"sys\.stdout\.write\(", text
+        ):
+            offenders.append(py_file.relative_to(SRC))
+    assert not offenders, (
+        "stdout writes found in: " + ", ".join(str(o) for o in offenders)
+    )

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

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

tests/test_transport_framing.py

@@ -0,0 +1,159 @@
+import sys
+import json
+import struct
+import socket
+import threading
+import time
+import select
+from pathlib import Path
+
+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
+    )
+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]
+    ready = threading.Event()
+
+    def _run():
+        ready.set()
+        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()
+        except Exception:
+            pass
+        finally:
+            sock.close()
+
+    threading.Thread(target=_run, daemon=True).start()
+    ready.wait()
+    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
+        # 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()
+            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)
+    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)
+    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():
+    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 (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()
+
+
+@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