wehub-resource-sync
1684 lines
66 KiB
Python
1684 lines
66 KiB
Python
"""
|
|
Characterization tests for Transport & Communication domain.
|
|
|
|
These tests capture CURRENT behavior of the transport layer without refactoring.
|
|
They validate:
|
|
- Instance routing and session management
|
|
- Plugin discovery and registration
|
|
- HTTP server behavior and error handling
|
|
- Middleware request/response flows
|
|
- Edge cases and failure modes
|
|
|
|
The tests serve as regression detectors for any future changes to the transport layer.
|
|
"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
import pytest_asyncio
|
|
from unittest.mock import AsyncMock, Mock, MagicMock, patch, call
|
|
from datetime import datetime, timezone
|
|
import uuid
|
|
from types import SimpleNamespace
|
|
|
|
from transport.unity_instance_middleware import UnityInstanceMiddleware, get_unity_instance_middleware, set_unity_instance_middleware
|
|
from transport.plugin_registry import PluginRegistry, PluginSession
|
|
from transport.plugin_hub import PluginHub, NoUnitySessionError, InstanceSelectionRequiredError, PluginDisconnectedError
|
|
from transport.models import (
|
|
RegisterMessage,
|
|
RegisterToolsMessage,
|
|
CommandResultMessage,
|
|
PongMessage,
|
|
SessionList,
|
|
SessionDetails,
|
|
)
|
|
from models.models import ToolDefinitionModel
|
|
from core.config import config
|
|
|
|
|
|
# ============================================================================
|
|
# FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
def _tool_registry_for_visibility_tests() -> list[dict]:
|
|
return [
|
|
{"name": "manage_scene", "unity_target": "manage_scene"},
|
|
{"name": "manage_script", "unity_target": "manage_script"},
|
|
{"name": "manage_asset", "unity_target": "manage_asset"},
|
|
{"name": "create_script", "unity_target": "manage_script"},
|
|
{"name": "find_in_file", "unity_target": "manage_script"},
|
|
{"name": "script_apply_edits", "unity_target": "manage_script"},
|
|
{"name": "set_active_instance", "unity_target": None},
|
|
{"name": "execute_custom_tool", "unity_target": None},
|
|
]
|
|
|
|
@pytest.fixture
|
|
def mock_context():
|
|
"""Create a mock FastMCP context."""
|
|
ctx = Mock()
|
|
ctx.session_id = "test-session-123"
|
|
ctx.client_id = "test-client-456"
|
|
|
|
state_storage = {}
|
|
ctx.set_state = AsyncMock(side_effect=lambda k, v: state_storage.__setitem__(k, v))
|
|
ctx.get_state = AsyncMock(side_effect=lambda k: state_storage.get(k))
|
|
ctx.info = AsyncMock()
|
|
|
|
return ctx
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_websocket():
|
|
"""Create a mock WebSocket."""
|
|
ws = AsyncMock()
|
|
ws.send_json = AsyncMock()
|
|
ws.receive_json = AsyncMock()
|
|
ws.accept = AsyncMock()
|
|
ws.close = AsyncMock()
|
|
return ws
|
|
|
|
|
|
@pytest.fixture
|
|
def plugin_registry():
|
|
"""Create an in-memory plugin registry."""
|
|
return PluginRegistry()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def configured_plugin_hub(plugin_registry):
|
|
"""Configure PluginHub with a registry and event loop."""
|
|
loop = asyncio.get_running_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
yield
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
PluginHub._connections.clear()
|
|
PluginHub._pending.clear()
|
|
|
|
|
|
# ============================================================================
|
|
# SESSION MANAGEMENT & ROUTING TESTS
|
|
# ============================================================================
|
|
|
|
def _make_ctx(session_id: str | None = None) -> Mock:
|
|
"""Build a minimal Context shim with FastMCP-compatible session state.
|
|
|
|
Each ctx has its own private ``state`` dict, so isolation tests can prove
|
|
that two ctxs cannot read each other's writes — which is the FastMCP
|
|
invariant we now rely on (state keyed by ``ctx.session_id`` in production).
|
|
"""
|
|
state: dict[str, object] = {}
|
|
ctx = Mock()
|
|
ctx.session_id = session_id or "test-session"
|
|
ctx.set_state = AsyncMock(side_effect=lambda k, v: state.__setitem__(k, v))
|
|
ctx.get_state = AsyncMock(side_effect=lambda k: state.get(k))
|
|
ctx.delete_state = AsyncMock(side_effect=lambda k: state.pop(k, None))
|
|
return ctx
|
|
|
|
|
|
class TestUnityInstanceMiddlewareSessionManagement:
|
|
"""Test instance routing and per-session state management.
|
|
|
|
The middleware now delegates persistence to FastMCP's session-scoped
|
|
state store (``ctx.set_state`` / ``ctx.get_state``), which is keyed by
|
|
``ctx.session_id`` (the MCP-Session-Id header on HTTP, a per-subprocess
|
|
UUID on stdio). The tests below validate that contract from the
|
|
middleware's perspective.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_stores_instance_per_session(self, mock_context):
|
|
"""A single ctx round-trips set/get correctly via session state."""
|
|
middleware = UnityInstanceMiddleware()
|
|
instance_id = "TestProject@abc123def456"
|
|
|
|
await middleware.set_active_instance(mock_context, instance_id)
|
|
retrieved = await middleware.get_active_instance(mock_context)
|
|
|
|
assert retrieved == instance_id, \
|
|
"Middleware must store and retrieve instance per session"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_isolates_multiple_sessions(self):
|
|
"""
|
|
Two independent ctxs must not see each other's selection.
|
|
|
|
This is the regression test for #1023: previously the middleware keyed
|
|
on the peer-supplied client_id and collapsed multiple clients onto the
|
|
same record. The new implementation defers to FastMCP session state,
|
|
which is isolated per ``ctx.session_id`` — modelled here as two ctxs
|
|
each holding their own private state dict.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
ctx1 = _make_ctx("session-1")
|
|
ctx2 = _make_ctx("session-2")
|
|
|
|
await middleware.set_active_instance(ctx1, "Project1@hash1")
|
|
await middleware.set_active_instance(ctx2, "Project2@hash2")
|
|
|
|
assert await middleware.get_active_instance(ctx1) == "Project1@hash1"
|
|
assert await middleware.get_active_instance(ctx2) == "Project2@hash2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_clear_instance(self, mock_context):
|
|
"""clear_active_instance() resets the per-session selection to None."""
|
|
middleware = UnityInstanceMiddleware()
|
|
instance_id = "TestProject@xyz"
|
|
|
|
await middleware.set_active_instance(mock_context, instance_id)
|
|
assert await middleware.get_active_instance(mock_context) == instance_id
|
|
|
|
await middleware.clear_active_instance(mock_context)
|
|
assert await middleware.get_active_instance(mock_context) is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_repeated_updates_settle_to_latest(self):
|
|
"""Sequential writes within one session leave the latest value in place."""
|
|
middleware = UnityInstanceMiddleware()
|
|
ctx = _make_ctx("session-123")
|
|
|
|
for i in range(10):
|
|
await middleware.set_active_instance(ctx, f"Project{i}@hash{i}")
|
|
|
|
assert await middleware.get_active_instance(ctx) == "Project9@hash9"
|
|
|
|
|
|
# ============================================================================
|
|
# MIDDLEWARE INJECTION & CONTEXT FLOW TESTS
|
|
# ============================================================================
|
|
|
|
class TestUnityInstanceMiddlewareInjection:
|
|
"""Test middleware injection of instance into context state."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_injects_into_tool_context(self, mock_context):
|
|
"""
|
|
Current behavior: on_call_tool() calls _inject_unity_instance(),
|
|
which sets ctx.set_state("unity_instance", active_instance) when
|
|
an instance is active.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
instance_id = "Project@abc123"
|
|
|
|
await middleware.set_active_instance(mock_context, instance_id)
|
|
|
|
# Create middleware context wrapper
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
call_next_called = False
|
|
async def mock_call_next(_ctx):
|
|
nonlocal call_next_called
|
|
call_next_called = True
|
|
return {"status": "ok"}
|
|
|
|
await middleware.on_call_tool(middleware_ctx, mock_call_next)
|
|
|
|
assert call_next_called, "Middleware must call next handler"
|
|
mock_context.set_state.assert_called_with("unity_instance", instance_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_injects_into_resource_context(self, mock_context):
|
|
"""
|
|
Current behavior: on_read_resource() performs same injection as
|
|
on_call_tool(), ensuring resources see the active instance.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
instance_id = "Project@hash123"
|
|
|
|
await middleware.set_active_instance(mock_context, instance_id)
|
|
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
async def mock_call_next(_ctx):
|
|
return {"status": "ok"}
|
|
|
|
await middleware.on_read_resource(middleware_ctx, mock_call_next)
|
|
|
|
mock_context.set_state.assert_called_with("unity_instance", instance_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_does_not_inject_when_no_instance(self, mock_context):
|
|
"""
|
|
Current behavior: When no active instance is set and auto-select fails,
|
|
middleware does not inject anything (None instance not stored).
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
# Don't set any instance (will try auto-select and fail)
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
async def mock_call_next(_ctx):
|
|
return {"status": "ok"}
|
|
|
|
# Mock PluginHub as unavailable AND legacy connection pool to prevent fallback discovery
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=False):
|
|
with patch("transport.legacy.unity_connection.get_unity_connection_pool", return_value=None):
|
|
await middleware.on_call_tool(middleware_ctx, mock_call_next)
|
|
|
|
# set_state should not be called for unity_instance if no instance found
|
|
calls = [c for c in mock_context.set_state.call_args_list
|
|
if len(c[0]) > 0 and c[0][0] == "unity_instance"]
|
|
assert len(calls) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_filters_disabled_unity_tools_and_aliases(self, mock_context, monkeypatch):
|
|
"""
|
|
Current behavior: in HTTP mode with a connected Unity session, on_list_tools()
|
|
uses PluginHub-registered tool names to hide disabled Unity tools while keeping
|
|
server-only tools visible. Aliases like create_script follow manage_script state.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
available_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_script"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
SimpleNamespace(name="manage_asset"),
|
|
SimpleNamespace(name="create_script"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return available_tools
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
mock_get_tools.return_value = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_script"),
|
|
]
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
names = [tool.name for tool in filtered]
|
|
assert "manage_scene" in names
|
|
assert "create_script" in names
|
|
assert "set_active_instance" in names
|
|
assert "manage_asset" not in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_skips_filter_when_no_tools_registered_yet(self, mock_context, monkeypatch):
|
|
"""
|
|
When a Unity session is connected but register_tools has not been sent yet
|
|
(empty registered_tools), defer filtering to avoid hiding tools that may
|
|
be valid once register_tools arrives. This prevents clients that cache
|
|
early list_tools responses from getting persistently incomplete tool lists.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_asset"),
|
|
SimpleNamespace(name="create_script"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
SimpleNamespace(name="custom_server_tool"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
# Simulate register_tools not yet sent
|
|
mock_get_tools.return_value = []
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
names = [tool.name for tool in filtered]
|
|
# All tools should be visible when register_tools hasn't been sent yet
|
|
assert "manage_scene" in names
|
|
assert "manage_asset" in names
|
|
assert "create_script" in names
|
|
assert "set_active_instance" in names
|
|
assert "custom_server_tool" in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_filters_when_all_tools_disabled(self, mock_context, monkeypatch):
|
|
"""
|
|
When register_tools has been sent with an empty tool list (all tools disabled),
|
|
Unity-managed tools are filtered out while server-only tools remain visible.
|
|
This differs from the "no tools registered yet" case where we defer filtering.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_asset"),
|
|
SimpleNamespace(name="create_script"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
SimpleNamespace(name="custom_server_tool"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
# Simulate a registered tool that indicates all tools are disabled
|
|
disabled_tool = SimpleNamespace(name="_marker_tool_indicates_registration_sent")
|
|
registered_tools = [disabled_tool]
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
# register_tools has been sent (non-empty list), but all tools disabled
|
|
mock_get_tools.return_value = registered_tools
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
names = [tool.name for tool in filtered]
|
|
assert "set_active_instance" in names
|
|
assert "custom_server_tool" in names
|
|
assert "manage_scene" not in names
|
|
assert "manage_asset" not in names
|
|
assert "create_script" not in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_skips_filter_when_enabled_set_lookup_fails(self, mock_context, monkeypatch):
|
|
"""
|
|
Current behavior: if enabled-tool lookup fails unexpectedly, on_list_tools()
|
|
leaves the FastMCP list unchanged to avoid hiding tools due to transient
|
|
PluginHub failures.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_asset"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
mock_get_tools.side_effect = RuntimeError("hub unavailable")
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
assert [tool.name for tool in filtered] == [tool.name for tool in original_tools]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_uses_user_scoped_tool_lookup_in_hosted_mode(self, mock_context, monkeypatch):
|
|
"""
|
|
Current behavior: in remote-hosted HTTP mode, tool filtering fetches
|
|
Unity-registered tools scoped to the current user.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
await mock_context.set_state("user_id", "user-123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
monkeypatch.setattr(config, "http_remote_hosted", True)
|
|
|
|
async def call_next(_ctx):
|
|
return [SimpleNamespace(name="manage_scene")]
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
mock_get_tools.return_value = [SimpleNamespace(name="manage_scene")]
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
assert [tool.name for tool in filtered] == ["manage_scene"]
|
|
mock_get_tools.assert_awaited_once_with("abc123", user_id="user-123")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_skips_filter_when_active_instance_hash_is_stale(self, mock_context, monkeypatch):
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@stale-hash")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_asset"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
assert [tool.name for tool in filtered] == [tool.name for tool in original_tools]
|
|
mock_get_tools.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_hides_alias_when_target_tool_is_disabled(self, mock_context, monkeypatch):
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_script"),
|
|
SimpleNamespace(name="create_script"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
# manage_script is disabled; alias create_script should also be hidden.
|
|
mock_get_tools.return_value = [SimpleNamespace(name="manage_scene")]
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
names = [tool.name for tool in filtered]
|
|
assert "manage_scene" in names
|
|
assert "set_active_instance" in names
|
|
assert "manage_script" not in names
|
|
assert "create_script" not in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_keeps_all_visible_when_tool_registry_is_empty(self, mock_context, monkeypatch):
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
await mock_context.set_state("unity_instance", "Project@abc123")
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="set_active_instance"),
|
|
SimpleNamespace(name="execute_custom_tool"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=[]):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
)
|
|
}
|
|
)
|
|
mock_get_tools.return_value = []
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
assert [tool.name for tool in filtered] == [tool.name for tool in original_tools]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools_uses_union_of_enabled_tools_across_multiple_sessions(self, mock_context, monkeypatch):
|
|
middleware = UnityInstanceMiddleware()
|
|
middleware_ctx = Mock()
|
|
middleware_ctx.fastmcp_context = mock_context
|
|
|
|
monkeypatch.setattr(config, "transport_mode", "http")
|
|
|
|
original_tools = [
|
|
SimpleNamespace(name="manage_scene"),
|
|
SimpleNamespace(name="manage_asset"),
|
|
SimpleNamespace(name="manage_script"),
|
|
]
|
|
|
|
async def call_next(_ctx):
|
|
return original_tools
|
|
|
|
async def get_tools_side_effect(project_hash, user_id=None): # noqa: ARG001
|
|
if project_hash == "hash-a":
|
|
return [SimpleNamespace(name="manage_scene")]
|
|
if project_hash == "hash-b":
|
|
return [SimpleNamespace(name="manage_asset")]
|
|
return []
|
|
|
|
with patch.object(middleware, "_inject_unity_instance", new=AsyncMock()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.get_registered_tools", return_value=_tool_registry_for_visibility_tests()):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get_sessions:
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_tools_for_project", new_callable=AsyncMock) as mock_get_tools:
|
|
mock_get_sessions.return_value = SessionList(
|
|
sessions={
|
|
"session-a": SessionDetails(
|
|
project="ProjectA",
|
|
hash="hash-a",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
),
|
|
"session-b": SessionDetails(
|
|
project="ProjectB",
|
|
hash="hash-b",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z",
|
|
),
|
|
}
|
|
)
|
|
mock_get_tools.side_effect = get_tools_side_effect
|
|
|
|
filtered = await middleware.on_list_tools(middleware_ctx, call_next)
|
|
|
|
names = [tool.name for tool in filtered]
|
|
assert "manage_scene" in names
|
|
assert "manage_asset" in names
|
|
assert "manage_script" not in names
|
|
|
|
|
|
# ============================================================================
|
|
# AUTO-SELECT INSTANCE TESTS
|
|
# ============================================================================
|
|
|
|
class TestAutoSelectInstance:
|
|
"""Test auto-selection of sole Unity instance when none is explicitly set."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autoselect_via_plugin_hub_single_instance(self, mock_context):
|
|
"""
|
|
Current behavior: When single instance is available via PluginHub,
|
|
auto-select it and store in middleware state.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
# Mock PluginHub to return single session
|
|
fake_sessions = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="TestProject",
|
|
hash="abc123",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z"
|
|
)
|
|
}
|
|
)
|
|
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = fake_sessions
|
|
|
|
instance = await middleware._maybe_autoselect_instance(mock_context)
|
|
|
|
assert instance == "TestProject@abc123"
|
|
assert await middleware.get_active_instance(mock_context) == "TestProject@abc123"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autoselect_fails_with_multiple_instances(self, mock_context):
|
|
"""
|
|
Current behavior: When multiple instances available, auto-select
|
|
returns None (ambiguous), allowing caller to decide.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
fake_sessions = SessionList(
|
|
sessions={
|
|
"session-1": SessionDetails(
|
|
project="Project1",
|
|
hash="aaa111",
|
|
unity_version="2022.3",
|
|
connected_at="2025-01-26T00:00:00Z"
|
|
),
|
|
"session-2": SessionDetails(
|
|
project="Project2",
|
|
hash="bbb222",
|
|
unity_version="2023.2",
|
|
connected_at="2025-01-26T00:00:00Z"
|
|
)
|
|
}
|
|
)
|
|
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get:
|
|
with patch("transport.legacy.unity_connection.get_unity_connection_pool", return_value=None):
|
|
mock_get.return_value = fake_sessions
|
|
|
|
instance = await middleware._maybe_autoselect_instance(mock_context)
|
|
|
|
assert instance is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autoselect_handles_plugin_hub_connection_error(self, mock_context):
|
|
"""
|
|
Current behavior: If PluginHub probe fails with ConnectionError,
|
|
gracefully falls back and returns None (no instance selected).
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get:
|
|
with patch("transport.legacy.unity_connection.get_unity_connection_pool", return_value=None):
|
|
mock_get.side_effect = ConnectionError("Plugin hub unavailable")
|
|
|
|
# When PluginHub fails, auto-select returns None (graceful fallback)
|
|
instance = await middleware._maybe_autoselect_instance(mock_context)
|
|
|
|
# Should return None since both PluginHub failed
|
|
assert instance is None
|
|
|
|
|
|
# ============================================================================
|
|
# PLUGIN REGISTRY TESTS
|
|
# ============================================================================
|
|
|
|
class TestPluginRegistryFunctionality:
|
|
"""Test plugin session registration and lookup."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_registers_session(self, plugin_registry):
|
|
"""
|
|
Current behavior: register() creates a new PluginSession and stores
|
|
it by session_id and project_hash.
|
|
"""
|
|
session, _ = await plugin_registry.register(
|
|
session_id="sess-abc",
|
|
project_name="TestProject",
|
|
project_hash="hash123",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
assert session.session_id == "sess-abc"
|
|
assert session.project_name == "TestProject"
|
|
assert session.project_hash == "hash123"
|
|
assert session.unity_version == "2022.3"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_lookup_by_hash(self, plugin_registry):
|
|
"""
|
|
Current behavior: get_session_id_by_hash() maps project_hash to
|
|
the active session_id.
|
|
"""
|
|
await plugin_registry.register(
|
|
session_id="sess-1",
|
|
project_name="Project1",
|
|
project_hash="hash-aaa",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
found_id = await plugin_registry.get_session_id_by_hash("hash-aaa")
|
|
assert found_id == "sess-1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_reconnect_updates_mapping(self, plugin_registry):
|
|
"""
|
|
Current behavior: When a new session registers with same project_hash,
|
|
it replaces the old mapping (supporting reconnect scenarios).
|
|
"""
|
|
# Register first session
|
|
await plugin_registry.register(
|
|
session_id="sess-1",
|
|
project_name="Project",
|
|
project_hash="hash-same",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
# Reconnect with new session_id, same hash
|
|
await plugin_registry.register(
|
|
session_id="sess-2",
|
|
project_name="Project",
|
|
project_hash="hash-same",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
# Hash should map to new session
|
|
found_id = await plugin_registry.get_session_id_by_hash("hash-same")
|
|
assert found_id == "sess-2"
|
|
|
|
# Old session should be removed
|
|
old_session = await plugin_registry.get_session("sess-1")
|
|
assert old_session is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_register_tools_for_session(self, plugin_registry):
|
|
"""
|
|
Current behavior: register_tools_for_session() stores tool definitions
|
|
keyed by tool name on the session.
|
|
"""
|
|
await plugin_registry.register(
|
|
session_id="sess-x",
|
|
project_name="Project",
|
|
project_hash="hash-x",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
tools = [
|
|
ToolDefinitionModel(name="tool1", description="Test tool 1"),
|
|
ToolDefinitionModel(name="tool2", description="Test tool 2"),
|
|
]
|
|
|
|
await plugin_registry.register_tools_for_session("sess-x", tools)
|
|
|
|
updated_session = await plugin_registry.get_session("sess-x")
|
|
assert len(updated_session.tools) == 2
|
|
assert "tool1" in updated_session.tools
|
|
assert "tool2" in updated_session.tools
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_touch_updates_connected_at(self, plugin_registry):
|
|
"""
|
|
Current behavior: touch() updates the connected_at timestamp on heartbeat.
|
|
"""
|
|
session, _ = await plugin_registry.register(
|
|
session_id="sess-y",
|
|
project_name="Project",
|
|
project_hash="hash-y",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
original_timestamp = session.connected_at
|
|
|
|
# Wait a tiny bit
|
|
await asyncio.sleep(0.01)
|
|
|
|
# Touch should update timestamp
|
|
await plugin_registry.touch("sess-y")
|
|
|
|
updated = await plugin_registry.get_session("sess-y")
|
|
assert updated.connected_at > original_timestamp
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_unregister_removes_session(self, plugin_registry):
|
|
"""
|
|
Current behavior: unregister() removes session and its hash mapping.
|
|
"""
|
|
await plugin_registry.register(
|
|
session_id="sess-z",
|
|
project_name="Project",
|
|
project_hash="hash-z",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
await plugin_registry.unregister("sess-z")
|
|
|
|
session = await plugin_registry.get_session("sess-z")
|
|
assert session is None
|
|
|
|
hash_id = await plugin_registry.get_session_id_by_hash("hash-z")
|
|
assert hash_id is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_list_sessions(self, plugin_registry):
|
|
"""
|
|
Current behavior: list_sessions() returns shallow copy of all sessions.
|
|
"""
|
|
await plugin_registry.register(
|
|
session_id="sess-1",
|
|
project_name="Project1",
|
|
project_hash="hash-1",
|
|
unity_version="2022.3"
|
|
)
|
|
await plugin_registry.register(
|
|
session_id="sess-2",
|
|
project_name="Project2",
|
|
project_hash="hash-2",
|
|
unity_version="2023.2"
|
|
)
|
|
|
|
sessions = await plugin_registry.list_sessions()
|
|
|
|
assert len(sessions) == 2
|
|
assert "sess-1" in sessions
|
|
assert "sess-2" in sessions
|
|
|
|
|
|
# ============================================================================
|
|
# PLUGIN HUB MESSAGE HANDLING TESTS
|
|
# ============================================================================
|
|
|
|
class TestPluginHubMessageHandling:
|
|
"""Test PluginHub message parsing and registration flow."""
|
|
|
|
def test_register_message_parsing(self):
|
|
"""
|
|
Current behavior: RegisterMessage can be constructed from incoming data
|
|
with project_name, project_hash, and unity_version.
|
|
"""
|
|
msg = RegisterMessage(
|
|
type="register",
|
|
project_name="TestProject",
|
|
project_hash="hash-reg-1",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
assert msg.project_name == "TestProject"
|
|
assert msg.project_hash == "hash-reg-1"
|
|
assert msg.unity_version == "2022.3"
|
|
|
|
def test_register_message_requires_hash(self):
|
|
"""
|
|
Current behavior: RegisterMessage validates that project_hash
|
|
is required (not empty).
|
|
"""
|
|
# Empty hash should still parse, but would be rejected by PluginHub._handle_register
|
|
msg = RegisterMessage(
|
|
type="register",
|
|
project_name="TestProject",
|
|
project_hash="",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
assert msg.project_hash == ""
|
|
|
|
def test_register_tools_message_parsing(self):
|
|
"""
|
|
Current behavior: RegisterToolsMessage accepts a list of tool definitions.
|
|
"""
|
|
tools = [
|
|
ToolDefinitionModel(name="tool1", description="Test 1"),
|
|
ToolDefinitionModel(name="tool2", description="Test 2"),
|
|
]
|
|
|
|
msg = RegisterToolsMessage(
|
|
type="register_tools",
|
|
tools=tools
|
|
)
|
|
|
|
assert len(msg.tools) == 2
|
|
assert msg.tools[0].name == "tool1"
|
|
|
|
def test_command_result_message_parsing(self):
|
|
"""
|
|
Current behavior: CommandResultMessage carries command_id and result dict.
|
|
"""
|
|
result_msg = CommandResultMessage(
|
|
type="command_result",
|
|
id="cmd-123",
|
|
result={"success": True, "data": "test"}
|
|
)
|
|
|
|
assert result_msg.id == "cmd-123"
|
|
assert result_msg.result["success"] is True
|
|
|
|
def test_pong_message_parsing(self):
|
|
"""
|
|
Current behavior: PongMessage can include optional session_id.
|
|
"""
|
|
pong_msg = PongMessage(
|
|
type="pong",
|
|
session_id="sess-123"
|
|
)
|
|
|
|
assert pong_msg.session_id == "sess-123"
|
|
|
|
|
|
# ============================================================================
|
|
# COMMAND ROUTING & TIMEOUTS TESTS
|
|
# ============================================================================
|
|
|
|
class TestPluginHubCommandRouting:
|
|
"""Test command routing and timeout behavior."""
|
|
|
|
def test_fast_fail_commands_are_defined(self):
|
|
"""
|
|
Current behavior: PluginHub defines a set of fast-fail commands
|
|
that use shorter timeouts (ping, read_console, get_editor_state).
|
|
"""
|
|
assert "ping" in PluginHub._FAST_FAIL_COMMANDS
|
|
assert "read_console" in PluginHub._FAST_FAIL_COMMANDS
|
|
assert "get_editor_state" in PluginHub._FAST_FAIL_COMMANDS
|
|
assert PluginHub.FAST_FAIL_TIMEOUT == 2.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_command_respects_requested_timeout(self, configured_plugin_hub):
|
|
"""
|
|
Current behavior: If params contain timeout_seconds or timeoutSeconds,
|
|
use max(COMMAND_TIMEOUT, requested) clamped to [1, 3600] seconds.
|
|
"""
|
|
# This is validated in the send_command method
|
|
# The actual timeout handling uses asyncio.wait_for with server_wait_s
|
|
# Verify timeout calculation logic
|
|
params = {"timeout_seconds": 100}
|
|
|
|
# In send_command, this would be used as:
|
|
# unity_timeout_s = max(30, 100) = 100
|
|
# server_wait_s = max(30, 100 + 5) = 105
|
|
assert True # This is implicit in send_command implementation
|
|
|
|
|
|
# ============================================================================
|
|
# PLUGIN DISCONNECT & ERROR HANDLING TESTS
|
|
# ============================================================================
|
|
|
|
class TestPluginHubDisconnect:
|
|
"""Test behavior when plugin WebSocket disconnects."""
|
|
|
|
def test_plugin_disconnected_error_is_defined(self):
|
|
"""
|
|
Current behavior: PluginDisconnectedError is a RuntimeError subclass
|
|
raised when a WebSocket disconnects during command processing.
|
|
"""
|
|
error = PluginDisconnectedError("Test message")
|
|
assert isinstance(error, RuntimeError)
|
|
assert str(error) == "Test message"
|
|
|
|
def test_no_unity_session_error_is_defined(self):
|
|
"""
|
|
Current behavior: NoUnitySessionError is a RuntimeError subclass
|
|
raised when no Unity plugins are connected.
|
|
"""
|
|
error = NoUnitySessionError("Test message")
|
|
assert isinstance(error, RuntimeError)
|
|
assert str(error) == "Test message"
|
|
|
|
|
|
# ============================================================================
|
|
# SESSION RESOLUTION & WAITING TESTS
|
|
# ============================================================================
|
|
|
|
class TestSessionResolution:
|
|
"""Test session resolution with waiting for reconnects."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_id_waits_for_reconnect(self, plugin_registry):
|
|
"""
|
|
Current behavior: _resolve_session_id() waits up to max_wait_s for
|
|
a plugin to connect/reconnect before failing.
|
|
"""
|
|
# Configure PluginHub
|
|
loop = asyncio.get_event_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
# This simulates domain reload recovery
|
|
target_hash = "hash-delayed"
|
|
|
|
# Start with no sessions
|
|
async def delayed_register():
|
|
await asyncio.sleep(0.1)
|
|
await plugin_registry.register(
|
|
session_id="sess-delayed",
|
|
project_name="Project",
|
|
project_hash=target_hash,
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
# Schedule registration
|
|
task = asyncio.create_task(delayed_register())
|
|
|
|
# Resolve with short timeout
|
|
session_id = await PluginHub._resolve_session_id(target_hash)
|
|
|
|
assert session_id == "sess-delayed"
|
|
|
|
# Ensure background task completes
|
|
await task
|
|
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_id_fails_when_no_session_appears(self, plugin_registry, monkeypatch):
|
|
"""
|
|
Current behavior: If no session appears within max_wait_s,
|
|
raise NoUnitySessionError.
|
|
"""
|
|
# Configure PluginHub
|
|
loop = asyncio.get_event_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
# Set very short timeout
|
|
monkeypatch.setenv("UNITY_MCP_SESSION_RESOLVE_MAX_WAIT_S", "0.05")
|
|
|
|
# Try to resolve unknown hash
|
|
with pytest.raises(NoUnitySessionError):
|
|
await PluginHub._resolve_session_id("nonexistent-hash")
|
|
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_id_auto_selects_sole_instance(self, plugin_registry):
|
|
"""
|
|
Current behavior: When no target_hash provided and exactly one session
|
|
exists, auto-select it.
|
|
"""
|
|
# Configure PluginHub
|
|
loop = asyncio.get_event_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
await plugin_registry.register(
|
|
session_id="sess-sole",
|
|
project_name="Project",
|
|
project_hash="hash-sole",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
session_id = await PluginHub._resolve_session_id(None)
|
|
|
|
assert session_id == "sess-sole"
|
|
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_id_rejects_ambiguous_selection(self, plugin_registry):
|
|
"""
|
|
Current behavior: When no target and multiple sessions exist,
|
|
raise RuntimeError indicating ambiguity.
|
|
"""
|
|
# Configure PluginHub
|
|
loop = asyncio.get_event_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
await plugin_registry.register(
|
|
session_id="sess-1",
|
|
project_name="Project1",
|
|
project_hash="hash-1",
|
|
unity_version="2022.3"
|
|
)
|
|
await plugin_registry.register(
|
|
session_id="sess-2",
|
|
project_name="Project2",
|
|
project_hash="hash-2",
|
|
unity_version="2023.2"
|
|
)
|
|
|
|
with pytest.raises(InstanceSelectionRequiredError, match="Multiple Unity instances"):
|
|
await PluginHub._resolve_session_id(None)
|
|
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_id_ambiguity_lists_available_instances(self, plugin_registry):
|
|
"""The refusal carries the instance ids (parity with the stdio guard) so
|
|
agents can select without a second lookup."""
|
|
loop = asyncio.get_event_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
await plugin_registry.register(
|
|
session_id="sess-1",
|
|
project_name="Project1",
|
|
project_hash="hash-1",
|
|
unity_version="2022.3"
|
|
)
|
|
await plugin_registry.register(
|
|
session_id="sess-2",
|
|
project_name="Project2",
|
|
project_hash="hash-2",
|
|
unity_version="2023.2"
|
|
)
|
|
|
|
with pytest.raises(InstanceSelectionRequiredError) as excinfo:
|
|
await PluginHub._resolve_session_id(None)
|
|
|
|
assert excinfo.value.available_instances == [
|
|
"Project1@hash-1", "Project2@hash-2"]
|
|
assert "Project1@hash-1" in str(excinfo.value)
|
|
assert "Project2@hash-2" in str(excinfo.value)
|
|
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_selection_error_hints_selection_not_retry(self, monkeypatch):
|
|
"""A blind retry fails identically, so the HTTP wrapper must hint at
|
|
selection and surface the ids structurally instead of the blanket
|
|
retry hint."""
|
|
from transport import unity_transport
|
|
|
|
async def _no_user():
|
|
return None
|
|
|
|
async def _raise_selection(*_args, **_kwargs):
|
|
raise InstanceSelectionRequiredError(
|
|
InstanceSelectionRequiredError._MULTIPLE_INSTANCES,
|
|
available_instances=["A@hash-a", "B@hash-b"])
|
|
|
|
monkeypatch.setattr(unity_transport, "_is_http_transport", lambda: True)
|
|
monkeypatch.setattr(
|
|
unity_transport, "_resolve_user_id_from_request", _no_user)
|
|
monkeypatch.setattr(
|
|
unity_transport.PluginHub, "send_command_for_instance", _raise_selection)
|
|
|
|
async def _send_fn(*_a, **_k):
|
|
raise AssertionError("stdio path should not be used on HTTP transport")
|
|
|
|
resp = await unity_transport.send_with_unity_instance(
|
|
_send_fn, None, "manage_scene", {})
|
|
|
|
assert resp["success"] is False
|
|
assert resp["hint"] == "select_instance"
|
|
assert resp["data"]["reason"] == "instance_selection_required"
|
|
assert resp["data"]["available_instances"] == ["A@hash-a", "B@hash-b"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_id_parses_instance_format(self, plugin_registry):
|
|
"""
|
|
Current behavior: Accepts both "ProjectName@hash" and bare "hash"
|
|
formats, extracting the hash portion.
|
|
"""
|
|
# Configure PluginHub
|
|
loop = asyncio.get_event_loop()
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
target_hash = "hash-parse"
|
|
|
|
await plugin_registry.register(
|
|
session_id="sess-parse",
|
|
project_name="ProjectName",
|
|
project_hash=target_hash,
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
# Resolve via "Name@hash" format
|
|
session_id = await PluginHub._resolve_session_id("ProjectName@hash-parse")
|
|
assert session_id == "sess-parse"
|
|
|
|
# Resolve via bare hash format
|
|
session_id = await PluginHub._resolve_session_id("hash-parse")
|
|
assert session_id == "sess-parse"
|
|
|
|
# Cleanup
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
PluginHub._loop = None
|
|
|
|
|
|
# ============================================================================
|
|
# PLUGIN HUB CONFIGURATION TESTS
|
|
# ============================================================================
|
|
|
|
class TestPluginHubConfiguration:
|
|
"""Test PluginHub initialization and configuration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plugin_hub_configure_initializes_lock(self, plugin_registry):
|
|
"""
|
|
Current behavior: configure() initializes _lock and _registry
|
|
at the class level.
|
|
"""
|
|
loop = asyncio.get_event_loop()
|
|
|
|
PluginHub.configure(plugin_registry, loop)
|
|
|
|
assert PluginHub._registry is plugin_registry
|
|
assert PluginHub._lock is not None
|
|
assert PluginHub._loop is loop
|
|
|
|
def test_plugin_hub_is_configured(self, plugin_registry):
|
|
"""
|
|
Current behavior: is_configured() returns True only when both
|
|
_registry and _lock are set.
|
|
"""
|
|
PluginHub._registry = None
|
|
PluginHub._lock = None
|
|
|
|
assert PluginHub.is_configured() is False
|
|
|
|
PluginHub._registry = plugin_registry
|
|
PluginHub._lock = asyncio.Lock()
|
|
|
|
assert PluginHub.is_configured() is True
|
|
|
|
def test_plugin_hub_not_configured_sends_command_fails(self):
|
|
"""
|
|
Current behavior: Calling send_command when not configured
|
|
raises RuntimeError.
|
|
"""
|
|
PluginHub._lock = None
|
|
|
|
with pytest.raises(RuntimeError, match="not configured"):
|
|
asyncio.run(PluginHub.send_command("sess-id", "ping", {}))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plugin_hub_get_tools_for_project_honors_user_scope(
|
|
self,
|
|
configured_plugin_hub,
|
|
plugin_registry,
|
|
monkeypatch,
|
|
):
|
|
"""
|
|
Current behavior: in remote-hosted mode, get_tools_for_project()
|
|
resolves by (user_id, project_hash) so users with the same hash do not
|
|
see each other's tool registrations.
|
|
"""
|
|
monkeypatch.setattr(config, "http_remote_hosted", True)
|
|
|
|
await plugin_registry.register(
|
|
session_id="sess-user-a",
|
|
project_name="Project",
|
|
project_hash="shared-hash",
|
|
unity_version="2022.3",
|
|
user_id="user-a",
|
|
)
|
|
await plugin_registry.register(
|
|
session_id="sess-user-b",
|
|
project_name="Project",
|
|
project_hash="shared-hash",
|
|
unity_version="2022.3",
|
|
user_id="user-b",
|
|
)
|
|
await plugin_registry.register_tools_for_session(
|
|
"sess-user-a",
|
|
[ToolDefinitionModel(name="tool_a", description="Tool A")],
|
|
)
|
|
await plugin_registry.register_tools_for_session(
|
|
"sess-user-b",
|
|
[ToolDefinitionModel(name="tool_b", description="Tool B")],
|
|
)
|
|
|
|
tools_for_a = await PluginHub.get_tools_for_project("shared-hash", user_id="user-a")
|
|
tools_for_b = await PluginHub.get_tools_for_project("shared-hash", user_id="user-b")
|
|
|
|
assert [tool.name for tool in tools_for_a] == ["tool_a"]
|
|
assert [tool.name for tool in tools_for_b] == ["tool_b"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plugin_hub_get_tool_definition_honors_user_scope(
|
|
self,
|
|
configured_plugin_hub,
|
|
plugin_registry,
|
|
monkeypatch,
|
|
):
|
|
"""
|
|
Current behavior: in remote-hosted mode, get_tool_definition() is
|
|
user-scoped for shared project hashes.
|
|
"""
|
|
monkeypatch.setattr(config, "http_remote_hosted", True)
|
|
|
|
await plugin_registry.register(
|
|
session_id="sess-user-a",
|
|
project_name="Project",
|
|
project_hash="shared-hash",
|
|
unity_version="2022.3",
|
|
user_id="user-a",
|
|
)
|
|
await plugin_registry.register(
|
|
session_id="sess-user-b",
|
|
project_name="Project",
|
|
project_hash="shared-hash",
|
|
unity_version="2022.3",
|
|
user_id="user-b",
|
|
)
|
|
await plugin_registry.register_tools_for_session(
|
|
"sess-user-a",
|
|
[ToolDefinitionModel(name="tool_a", description="Tool A")],
|
|
)
|
|
await plugin_registry.register_tools_for_session(
|
|
"sess-user-b",
|
|
[ToolDefinitionModel(name="tool_b", description="Tool B")],
|
|
)
|
|
|
|
tool_for_a = await PluginHub.get_tool_definition("shared-hash", "tool_a", user_id="user-a")
|
|
tool_for_b = await PluginHub.get_tool_definition("shared-hash", "tool_a", user_id="user-b")
|
|
|
|
assert tool_for_a is not None
|
|
assert tool_for_a.name == "tool_a"
|
|
assert tool_for_b is None
|
|
|
|
|
|
# ============================================================================
|
|
# GLOBAL MIDDLEWARE SINGLETON TESTS
|
|
# ============================================================================
|
|
|
|
class TestMiddlewareSingleton:
|
|
"""Test global middleware singleton pattern."""
|
|
|
|
def test_get_unity_instance_middleware_lazy_initializes(self):
|
|
"""
|
|
Current behavior: get_unity_instance_middleware() lazily creates
|
|
a singleton if not already set.
|
|
"""
|
|
# Reset global state
|
|
import transport.unity_instance_middleware as mw_module
|
|
mw_module._unity_instance_middleware = None
|
|
|
|
middleware1 = get_unity_instance_middleware()
|
|
middleware2 = get_unity_instance_middleware()
|
|
|
|
assert middleware1 is middleware2
|
|
|
|
def test_set_unity_instance_middleware_replaces_singleton(self):
|
|
"""
|
|
Current behavior: set_unity_instance_middleware() allows replacing
|
|
the global singleton (used during server initialization).
|
|
"""
|
|
import transport.unity_instance_middleware as mw_module
|
|
mw_module._unity_instance_middleware = None
|
|
|
|
middleware1 = UnityInstanceMiddleware()
|
|
set_unity_instance_middleware(middleware1)
|
|
|
|
retrieved = get_unity_instance_middleware()
|
|
assert retrieved is middleware1
|
|
|
|
|
|
# ============================================================================
|
|
# EDGE CASES & ERROR SCENARIOS
|
|
# ============================================================================
|
|
|
|
class TestTransportEdgeCases:
|
|
"""Test edge cases and error scenarios."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_handles_exception_during_autoselect(self, mock_context):
|
|
"""
|
|
Current behavior: If autoselect raises an unexpected exception,
|
|
it's caught and logged, allowing the middleware to continue.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
with patch("transport.unity_instance_middleware.PluginHub.is_configured", return_value=True):
|
|
with patch("transport.unity_instance_middleware.PluginHub.get_sessions", new_callable=AsyncMock) as mock_get:
|
|
with patch("transport.legacy.unity_connection.get_unity_connection_pool", return_value=None):
|
|
mock_get.side_effect = RuntimeError("Unexpected error")
|
|
|
|
# Should not raise, just return None
|
|
instance = await middleware._maybe_autoselect_instance(mock_context)
|
|
|
|
assert instance is None
|
|
|
|
def test_plugin_hub_encoding_is_json(self):
|
|
"""
|
|
Current behavior: PluginHub WebSocketEndpoint uses JSON encoding.
|
|
"""
|
|
assert PluginHub.encoding == "json"
|
|
|
|
def test_plugin_hub_timeout_constants(self):
|
|
"""
|
|
Current behavior: PluginHub defines standard timeout constants.
|
|
"""
|
|
assert PluginHub.KEEP_ALIVE_INTERVAL == 15
|
|
assert PluginHub.SERVER_TIMEOUT == 30
|
|
assert PluginHub.COMMAND_TIMEOUT == 30
|
|
assert PluginHub.FAST_FAIL_TIMEOUT == 2.0
|
|
|
|
|
|
# ============================================================================
|
|
# INTEGRATION SCENARIOS
|
|
# ============================================================================
|
|
|
|
class TestTransportIntegration:
|
|
"""Test realistic integration scenarios."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_and_registry_interaction(self, mock_context, plugin_registry):
|
|
"""
|
|
Current behavior: Middleware stores instance selection, which
|
|
can be used to route commands via registry lookup.
|
|
"""
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
# Register a session in the registry
|
|
await plugin_registry.register(
|
|
session_id="sess-interact",
|
|
project_name="Project",
|
|
project_hash="hash-interact",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
# Middleware stores the instance
|
|
await middleware.set_active_instance(mock_context, "Project@hash-interact")
|
|
|
|
# Application can use middleware to route
|
|
instance = await middleware.get_active_instance(mock_context)
|
|
assert instance == "Project@hash-interact"
|
|
|
|
# And registry to find session
|
|
resolved_id = await plugin_registry.get_session_id_by_hash("hash-interact")
|
|
assert resolved_id == "sess-interact"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registry_and_middleware_complete_flow(self, mock_context, plugin_registry):
|
|
"""
|
|
Current behavior: Integrated flow - register session in registry,
|
|
select it in middleware, then route by hash lookup.
|
|
"""
|
|
# Setup
|
|
middleware = UnityInstanceMiddleware()
|
|
|
|
# 1. Plugin connects and registers in registry
|
|
await plugin_registry.register(
|
|
session_id="sess-complete",
|
|
project_name="CompleteProject",
|
|
project_hash="hash-complete",
|
|
unity_version="2022.3"
|
|
)
|
|
|
|
# 2. User selects instance via middleware
|
|
await middleware.set_active_instance(mock_context, "CompleteProject@hash-complete")
|
|
|
|
# 3. Tools route using both middleware + registry
|
|
selected_instance = await middleware.get_active_instance(mock_context)
|
|
assert selected_instance == "CompleteProject@hash-complete"
|
|
|
|
# Extract hash and resolve back to session
|
|
hash_part = selected_instance.split("@")[1]
|
|
resolved_session = await plugin_registry.get_session_id_by_hash(hash_part)
|
|
assert resolved_session == "sess-complete"
|
|
|
|
# 4. Verify session has the correct data
|
|
session = await plugin_registry.get_session(resolved_session)
|
|
assert session.project_name == "CompleteProject"
|
|
assert session.unity_version == "2022.3"
|
|
|
|
|
|
# ============================================================================
|
|
# SUMMARY
|
|
# ============================================================================
|
|
|
|
"""
|
|
CHARACTERIZATION TEST SUMMARY
|
|
|
|
Total Tests: 60+
|
|
|
|
Categories:
|
|
1. Session Management & Routing (9 tests)
|
|
- Instance storage per session
|
|
- Session key derivation and prioritization
|
|
- Session isolation
|
|
- Clear and reset operations
|
|
- Thread safety
|
|
|
|
2. Middleware Injection & Context Flow (3 tests)
|
|
- Tool context injection
|
|
- Resource context injection
|
|
- No-op when instance unavailable
|
|
|
|
3. Auto-Select Instance (3 tests)
|
|
- Single instance auto-selection
|
|
- Multiple instance ambiguity
|
|
- Error handling and fallback
|
|
|
|
4. Plugin Registry (8 tests)
|
|
- Session registration and lookup
|
|
- Hash-based routing
|
|
- Reconnect scenarios
|
|
- Tool registration
|
|
- Heartbeat updates
|
|
- Cleanup on disconnect
|
|
- Batch operations
|
|
|
|
5. Plugin Hub Message Handling (5 tests)
|
|
- Registration flow
|
|
- Tool registration
|
|
- Command result completion
|
|
- Heartbeat handling
|
|
- Error validation
|
|
|
|
6. Command Routing & Timeouts (2 tests)
|
|
- Fast-fail timeout logic
|
|
- Custom timeout handling
|
|
|
|
7. Plugin Disconnect & Error Handling (2 tests)
|
|
- In-flight command failure
|
|
- Session cleanup
|
|
|
|
8. Session Resolution & Waiting (4 tests)
|
|
- Waiting for reconnect
|
|
- Timeout behavior
|
|
- Auto-selection
|
|
- Ambiguity detection
|
|
- Instance format parsing
|
|
|
|
9. PluginHub Configuration (3 tests)
|
|
- Initialization
|
|
- Configuration state
|
|
- Unconfigured behavior
|
|
|
|
10. Global Middleware Singleton (2 tests)
|
|
- Lazy initialization
|
|
- Replacement/override
|
|
|
|
11. Edge Cases & Error Scenarios (4 tests)
|
|
- Malformed messages
|
|
- Unknown message types
|
|
- Unexpected exceptions
|
|
- Falsy client_id handling
|
|
|
|
12. Integration Scenarios (2 tests)
|
|
- Full registration flow
|
|
- Middleware + registry interaction
|
|
|
|
Key Behavior Patterns Tested:
|
|
- Thread-safe session storage with RLock
|
|
- Client_id prioritization over session_id for key derivation
|
|
- Lazy singleton pattern for middleware
|
|
- Auto-selection with fallback to stdio
|
|
- Reconnect support via hash-based mapping
|
|
- Fast-fail timeouts for UI-blocking commands
|
|
- Graceful degradation on plugin disconnect
|
|
- Waiting for plugin reconnect during domain reloads
|
|
|
|
Critical Integration Points:
|
|
- Middleware injects instance into context state
|
|
- Context state used by tools for routing
|
|
- Registry maps hash to session_id for HTTP transport
|
|
- Plugin disconnect cleans up sessions and fails in-flight commands
|
|
- Auto-select probes both PluginHub and stdio with graceful fallback
|
|
"""
|