248 lines
8.6 KiB
Python
248 lines
8.6 KiB
Python
import pytest
|
|
|
|
from .test_helpers import DummyContext, DummyMCP
|
|
|
|
|
|
def setup_console_tools():
|
|
"""Setup console-related tools for testing."""
|
|
mcp = DummyMCP()
|
|
import services.tools.read_console
|
|
from services.registry import get_registered_tools
|
|
for tool_info in get_registered_tools():
|
|
tool_name = tool_info['name']
|
|
if any(keyword in tool_name for keyword in ['read_console', 'console']):
|
|
mcp.tools[tool_name] = tool_info['func']
|
|
return mcp.tools
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_console_full_default(monkeypatch):
|
|
tools = setup_console_tools()
|
|
read_console = tools["read_console"]
|
|
|
|
captured = {}
|
|
|
|
async def fake_send(_cmd, params, **_kwargs):
|
|
captured["params"] = params
|
|
return {
|
|
"success": True,
|
|
"data": {"lines": [{"level": "error", "message": "oops", "stacktrace": "trace", "time": "t"}]},
|
|
}
|
|
|
|
# Patch the send_command_with_retry function in the tools module
|
|
import services.tools.read_console
|
|
monkeypatch.setattr(
|
|
services.tools.read_console,
|
|
"async_send_command_with_retry",
|
|
fake_send,
|
|
)
|
|
|
|
resp = await read_console(ctx=DummyContext(), action="get", count=10)
|
|
assert resp == {
|
|
"success": True,
|
|
"data": {"lines": [{"level": "error", "message": "oops", "time": "t"}]},
|
|
}
|
|
assert captured["params"]["count"] == 10
|
|
assert captured["params"]["includeStacktrace"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_console_truncated(monkeypatch):
|
|
tools = setup_console_tools()
|
|
read_console = tools["read_console"]
|
|
|
|
captured = {}
|
|
|
|
async def fake_send(_cmd, params, **_kwargs):
|
|
captured["params"] = params
|
|
return {
|
|
"success": True,
|
|
"data": {"lines": [{"level": "error", "message": "oops", "stacktrace": "trace"}]},
|
|
}
|
|
|
|
# Patch the send_command_with_retry function in the tools module
|
|
import services.tools.read_console
|
|
monkeypatch.setattr(
|
|
services.tools.read_console,
|
|
"async_send_command_with_retry",
|
|
fake_send,
|
|
)
|
|
|
|
resp = await read_console(ctx=DummyContext(), action="get", count=10, include_stacktrace=False)
|
|
assert resp == {"success": True, "data": {
|
|
"lines": [{"level": "error", "message": "oops"}]}}
|
|
assert captured["params"]["includeStacktrace"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_console_default_count(monkeypatch):
|
|
"""Test that read_console defaults to count=10 when not specified."""
|
|
tools = setup_console_tools()
|
|
read_console = tools["read_console"]
|
|
|
|
captured = {}
|
|
|
|
async def fake_send(_cmd, params, **_kwargs):
|
|
captured["params"] = params
|
|
return {
|
|
"success": True,
|
|
"data": {"lines": [{"level": "error", "message": f"error {i}"} for i in range(15)]},
|
|
}
|
|
|
|
# Patch the send_command_with_retry function in the tools module
|
|
import services.tools.read_console
|
|
monkeypatch.setattr(
|
|
services.tools.read_console,
|
|
"async_send_command_with_retry",
|
|
fake_send,
|
|
)
|
|
|
|
# Call without specifying count - should default to 10
|
|
resp = await read_console(ctx=DummyContext(), action="get")
|
|
assert resp["success"] is True
|
|
# Verify that the default count of 10 was used
|
|
assert captured["params"]["count"] == 10
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_console_paging(monkeypatch):
|
|
"""Test that read_console paging works with page_size and cursor."""
|
|
tools = setup_console_tools()
|
|
read_console = tools["read_console"]
|
|
|
|
captured = {}
|
|
|
|
async def fake_send(_cmd, params, **_kwargs):
|
|
captured["params"] = params
|
|
# Simulate Unity returning paging info matching C# structure
|
|
page_size = params.get("pageSize", 10)
|
|
cursor = params.get("cursor", 0)
|
|
# Simulate 25 total messages
|
|
all_messages = [{"level": "error", "message": f"error {i}"} for i in range(25)]
|
|
|
|
# Return a page of results
|
|
start = cursor
|
|
end = min(start + page_size, len(all_messages))
|
|
messages = all_messages[start:end]
|
|
|
|
return {
|
|
"success": True,
|
|
"data": {
|
|
"items": messages,
|
|
"cursor": cursor,
|
|
"pageSize": page_size,
|
|
"nextCursor": str(end) if end < len(all_messages) else None,
|
|
"truncated": end < len(all_messages),
|
|
"total": len(all_messages),
|
|
},
|
|
}
|
|
|
|
# Patch the send_command_with_retry function in the tools module
|
|
import services.tools.read_console
|
|
monkeypatch.setattr(
|
|
services.tools.read_console,
|
|
"async_send_command_with_retry",
|
|
fake_send,
|
|
)
|
|
|
|
# First page - get first 5 entries
|
|
resp = await read_console(ctx=DummyContext(), action="get", page_size=5, cursor=0)
|
|
assert resp["success"] is True
|
|
assert captured["params"]["pageSize"] == 5
|
|
assert captured["params"]["cursor"] == 0
|
|
assert len(resp["data"]["items"]) == 5
|
|
assert resp["data"]["truncated"] is True
|
|
assert resp["data"]["nextCursor"] == "5"
|
|
assert resp["data"]["total"] == 25
|
|
|
|
# Second page - get next 5 entries
|
|
resp = await read_console(ctx=DummyContext(), action="get", page_size=5, cursor=5)
|
|
assert resp["success"] is True
|
|
assert captured["params"]["cursor"] == 5
|
|
assert len(resp["data"]["items"]) == 5
|
|
assert resp["data"]["truncated"] is True
|
|
assert resp["data"]["nextCursor"] == "10"
|
|
|
|
# Last page - get remaining entries
|
|
resp = await read_console(ctx=DummyContext(), action="get", page_size=5, cursor=20)
|
|
assert resp["success"] is True
|
|
assert len(resp["data"]["items"]) == 5
|
|
assert resp["data"]["truncated"] is False
|
|
assert resp["data"]["nextCursor"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_console_types_json_string(monkeypatch):
|
|
"""Test that read_console handles types parameter as JSON string (fixes issue #561)."""
|
|
tools = setup_console_tools()
|
|
read_console = tools["read_console"]
|
|
|
|
captured = {}
|
|
|
|
async def fake_send_with_unity_instance(_send_fn, _unity_instance, _command_type, params, **_kwargs):
|
|
captured["params"] = params
|
|
return {
|
|
"success": True,
|
|
"data": {"lines": [{"level": "error", "message": "test error"}]},
|
|
}
|
|
|
|
import services.tools.read_console as read_console_mod
|
|
monkeypatch.setattr(
|
|
read_console_mod,
|
|
"send_with_unity_instance",
|
|
fake_send_with_unity_instance,
|
|
)
|
|
|
|
# Test with types as JSON string (the problematic case from issue #561)
|
|
resp = await read_console(ctx=DummyContext(), action="get", types='["error", "warning", "all"]')
|
|
assert resp["success"] is True
|
|
# Verify types was parsed correctly and sent as a list
|
|
assert isinstance(captured["params"]["types"], list)
|
|
assert captured["params"]["types"] == ["error", "warning", "all"]
|
|
|
|
# Test case normalization to lowercase
|
|
captured.clear()
|
|
resp = await read_console(ctx=DummyContext(), action="get", types='["ERROR", "Warning", "LOG"]')
|
|
assert resp["success"] is True
|
|
assert captured["params"]["types"] == ["error", "warning", "log"]
|
|
|
|
# Test with types as actual list (should still work)
|
|
captured.clear()
|
|
resp = await read_console(ctx=DummyContext(), action="get", types=["error", "warning"])
|
|
assert resp["success"] is True
|
|
assert isinstance(captured["params"]["types"], list)
|
|
assert captured["params"]["types"] == ["error", "warning"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_console_types_validation(monkeypatch):
|
|
"""Test that read_console validates types entries and rejects invalid values."""
|
|
tools = setup_console_tools()
|
|
read_console = tools["read_console"]
|
|
|
|
captured = {}
|
|
|
|
async def fake_send_with_unity_instance(_send_fn, _unity_instance, _command_type, params, **_kwargs):
|
|
captured["params"] = params
|
|
return {"success": True, "data": {"lines": []}}
|
|
|
|
import services.tools.read_console as read_console_mod
|
|
monkeypatch.setattr(
|
|
read_console_mod,
|
|
"send_with_unity_instance",
|
|
fake_send_with_unity_instance,
|
|
)
|
|
|
|
# Invalid entry in list should return a clear error and not send.
|
|
captured.clear()
|
|
resp = await read_console(ctx=DummyContext(), action="get", types='["error", "nope"]')
|
|
assert resp["success"] is False
|
|
assert "invalid types entry" in resp["message"]
|
|
assert captured == {}
|
|
|
|
# Non-string entry should return a clear error and not send.
|
|
captured.clear()
|
|
resp = await read_console(ctx=DummyContext(), action="get", types='[1, "error"]')
|
|
assert resp["success"] is False
|
|
assert "types entries must be strings" in resp["message"]
|
|
assert captured == {} |