diff --git a/tests/test_logging_stdout.py b/tests/test_logging_stdout.py new file mode 100644 index 00000000..d6e728b7 --- /dev/null +++ b/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) + ) 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..39e84afd --- /dev/null +++ b/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