Convert skip decorators to xfail and harden framing tests

UnityMcpBridge/UnityMcpServer~/src/unity_connection.py

@@ -94,6 +94,8 @@ def receive_full_response(self, sock, buffer_size=config.buffer_size) -> bytes:
             try:
                 header = self._read_exact(sock, 8)
                 payload_len = struct.unpack('>Q', header)[0]
+                if payload_len == 0:
+                    raise Exception("Invalid framed length: 0")
                 if payload_len > (64 * 1024 * 1024):
                     raise Exception(f"Invalid framed length: {payload_len}")
                 payload = self._read_exact(sock, payload_len)

tests/test_logging_stdout.py

@@ -1,4 +1,4 @@
-import re
+import ast
 from pathlib import Path
 
 import pytest
@@ -13,8 +13,9 @@
 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.skip(
+        "Unity MCP server source not found. Tried:\n" + searched,
+        allow_module_level=True,
     )
 
 
@@ -24,14 +25,41 @@ def test_no_stdout_output_from_tools():
 
 
 def test_no_print_statements_in_codebase():
-    """Ensure no stray print statements remain in server source."""
+    """Ensure no stray print/sys.stdout writes remain in server source."""
     offenders = []
+    syntax_errors = []
     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
-        ):
+        try:
+            text = py_file.read_text(encoding="utf-8", errors="strict")
+        except UnicodeDecodeError:
+            # Be tolerant of encoding edge cases in source tree
+            text = py_file.read_text(encoding="utf-8", errors="ignore")
+        try:
+            tree = ast.parse(text, filename=str(py_file))
+        except SyntaxError:
+            syntax_errors.append(py_file.relative_to(SRC))
+            continue
+
+        class StdoutVisitor(ast.NodeVisitor):
+            def __init__(self):
+                self.hit = False
+
+            def visit_Call(self, node: ast.Call):
+                # print(...)
+                if isinstance(node.func, ast.Name) and node.func.id == "print":
+                    self.hit = True
+                # sys.stdout.write(...)
+                if isinstance(node.func, ast.Attribute) and node.func.attr == "write":
+                    val = node.func.value
+                    if isinstance(val, ast.Attribute) and val.attr == "stdout":
+                        if isinstance(val.value, ast.Name) and val.value.id == "sys":
+                            self.hit = True
+                self.generic_visit(node)
+
+        v = StdoutVisitor()
+        v.visit(tree)
+        if v.hit:
             offenders.append(py_file.relative_to(SRC))
-    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)

tests/test_resources_api.py

@@ -1,11 +1,11 @@
 import pytest
 
 
-@pytest.mark.skip(reason="TODO: resource.list returns only Assets/**/*.cs and rejects traversal")
+@pytest.mark.xfail(strict=False, reason="resource.list should return only Assets/**/*.cs and reject 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")
+@pytest.mark.xfail(strict=False, reason="resource.list should reject outside paths including drive letters and symlinks")
 def test_resource_list_rejects_outside_paths():
     pass

tests/test_script_editing.py

@@ -1,36 +1,36 @@
 import pytest
 
 
-@pytest.mark.skip(reason="TODO: create new script, validate, apply edits, build and compile scene")
+@pytest.mark.xfail(strict=False, reason="pending: 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")
+@pytest.mark.xfail(strict=False, reason="pending: multiple micro-edits debounce to single compilation")
 def test_micro_edits_debounce():
     pass
 
 
-@pytest.mark.skip(reason="TODO: line ending variations handled correctly")
+@pytest.mark.xfail(strict=False, reason="pending: 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")
+@pytest.mark.xfail(strict=False, reason="pending: 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")
+@pytest.mark.xfail(strict=False, reason="pending: 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")
+@pytest.mark.xfail(strict=False, reason="pending: symlink and junction protections on edits")
 def test_symlink_and_junction_protection():
     pass
 
 
-@pytest.mark.skip(reason="TODO: atomic write guarantees")
+@pytest.mark.xfail(strict=False, reason="pending: atomic write guarantees")
 def test_atomic_write_guarantees():
     pass

tests/test_transport_framing.py

@@ -18,8 +18,9 @@
 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.skip(
+        "Unity MCP server source not found. Tried:\n" + searched,
+        allow_module_level=True,
     )
 sys.path.insert(0, str(SRC))
 
@@ -37,19 +38,25 @@ def start_dummy_server(greeting: bytes, respond_ping: bool = False):
     def _run():
         ready.set()
         conn, _ = sock.accept()
+        conn.settimeout(1.0)
         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))
+                # Read exactly n bytes helper
+                def _read_exact(n: int) -> bytes:
+                    buf = b""
+                    while len(buf) < n:
+                        chunk = conn.recv(n - len(buf))
                         if not chunk:
                             break
-                        payload += chunk
+                        buf += chunk
+                    return buf
+
+                header = _read_exact(8)
+                if len(header) == 8:
+                    length = struct.unpack(">Q", header)[0]
+                    payload = _read_exact(length)
                     if payload == b'{"type":"ping"}':
                         resp = b'{"type":"pong"}'
                         conn.sendall(struct.pack(">Q", len(resp)) + resp)
@@ -79,13 +86,14 @@ def start_handshake_enforcing_server():
     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
+        # If client sends any data before greeting, disconnect (poll briefly)
+        deadline = time.time() + 0.5
+        while time.time() < deadline:
+            r, _, _ = select.select([conn], [], [], 0.05)
+            if r:
+                conn.close()
+                sock.close()
+                return
         conn.sendall(b"MCP/0.1 FRAMING=1\n")
         time.sleep(0.1)
         conn.close()
@@ -122,7 +130,7 @@ def test_unframed_data_disconnect():
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     sock.connect(("127.0.0.1", port))
     sock.sendall(b"BAD")
-    time.sleep(0.1)
+    time.sleep(0.4)
     try:
         data = sock.recv(1024)
         assert data == b""