import pytest
import subprocess
import sys
import os
import time
import socket
import json
from pathlib import Path
import threading
# Path to the helper script
HELPER_SCRIPT = Path(__file__).parent / "start_blender_server.py"
@pytest.fixture(scope="session")
def blender_executable():
"""
Finds the Blender executable.
"""
# Hardcoded for this environment based on user input/discovery
# In a real project, this might search PATH or use env vars
candidates = [
r"C:\Program Files (x86)\Steam\steamapps\common\Blender\blender.exe",
r"C:\Program Files\Blender Foundation\Blender 4.0\blender.exe",
"blender"
]
for path in candidates:
if os.path.exists(path) or (path == "blender" and shutil.which("blender")):
return path
pytest.skip("Blender executable not found")
@pytest.fixture(scope="session")
def blender_server(blender_executable):
"""
Starts Blender in background mode with the addon enabled.
Yields the (host, port) tuple.
"""
host = "localhost"
port = 9876
# Command to run Blender
cmd = [
blender_executable,
"--background",
"--factory-startup",
"--python", str(HELPER_SCRIPT)
]
print(f"Starting Blender: {' '.join(cmd)}")
# Start the process
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
universal_newlines=True,
encoding='utf-8',
errors='replace'
)
# Wait for the server to be ready
server_ready = False
start_time = time.time()
timeout = 30 # seconds
def monitor_output(proc):
while True:
line = proc.stdout.readline()
if not line:
break
print(f"[Blender] {line.strip()}")
if "BLENDER_MCP_SERVER_READY" in line:
nonlocal server_ready
server_ready = True
# Start monitoring thread
t = threading.Thread(target=monitor_output, args=(process,), daemon=True)
t.start()
# Wait loop
while not server_ready:
if time.time() - start_time > timeout:
process.terminate()
raise TimeoutError("Blender server failed to start within timeout")
if process.poll() is not None:
stderr = process.stderr.read()
raise RuntimeError(f"Blender process exited unexpectedly. Stderr: {stderr}")
time.sleep(0.1)
print("Blender server is ready!")
yield host, port
# Teardown
print("Stopping Blender server...")
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
@pytest.fixture
def real_blender_connection(blender_server):
"""
Provides a socket connection to the real Blender instance.
"""
host, port = blender_server
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10.0)
try:
sock.connect((host, port))
except ConnectionRefusedError:
pytest.fail(f"Could not connect to Blender server at {host}:{port}")
yield sock
sock.close()
def send_command(sock, command_type, params=None):
"""Helper to send a command and receive response"""
if params is None:
params = {}
cmd = {
"type": command_type,
"params": params
}
sock.sendall(json.dumps(cmd).encode('utf-8'))
# Simple response reading (might need buffering for large responses)
# In a real scenario, we'd use a proper framing protocol or read until valid JSON
# For now, a large buffer is usually enough for simple tests
print(f"Waiting for response to {command_type}...")
data = sock.recv(32768)
print(f"Received {len(data)} bytes")
return json.loads(data.decode('utf-8'))
