934 lines
30 KiB
Python
934 lines
30 KiB
Python
"""
|
|
Characterization tests for Models & Data Structures domain.
|
|
|
|
Tests capture CURRENT behavior of models in:
|
|
- Server/src/models/models.py (MCPResponse, UnityInstanceInfo, ToolParameterModel, ToolDefinitionModel)
|
|
- Server/src/models/unity_response.py (normalize_unity_response function)
|
|
|
|
Domain Overview:
|
|
- Purpose: Request/response structures, configuration schemas
|
|
- Pattern: Shared data definitions across Python/C# with duplications noted
|
|
- Key Issue: Duplicate session models should be consolidated (PluginSession vs SessionDetails)
|
|
|
|
These tests verify:
|
|
- Model instantiation with valid/invalid data
|
|
- Serialization and deserialization
|
|
- Validation logic and error messages
|
|
- Default value application
|
|
- Schema consistency
|
|
- Request/response contract verification
|
|
|
|
DUPLICATION NOTES:
|
|
- NOTE: PluginSession (Python) and SessionDetails (C# likely) represent the same concept
|
|
These should be consolidated in refactor P1-4
|
|
- NOTE: McpClient (C#) has many configuration flags that could be simplified via builder pattern
|
|
This relates to refactor P2-3
|
|
"""
|
|
import json
|
|
import pytest
|
|
from datetime import datetime
|
|
from typing import Any, Dict
|
|
|
|
from models.models import (
|
|
MCPResponse,
|
|
UnityInstanceInfo,
|
|
ToolParameterModel,
|
|
ToolDefinitionModel,
|
|
)
|
|
from models.unity_response import normalize_unity_response
|
|
|
|
|
|
class TestMCPResponseModel:
|
|
"""Test MCPResponse model instantiation, validation, and serialization."""
|
|
|
|
def test_mcp_response_minimal_required_fields(self):
|
|
"""Test MCPResponse with only required field (success)."""
|
|
response = MCPResponse(success=True)
|
|
|
|
assert response.success is True
|
|
assert response.message is None
|
|
assert response.error is None
|
|
assert response.data is None
|
|
assert response.hint is None
|
|
|
|
def test_mcp_response_all_fields(self):
|
|
"""Test MCPResponse with all fields specified."""
|
|
response = MCPResponse(
|
|
success=True,
|
|
message="Operation completed successfully",
|
|
error=None,
|
|
data={"key": "value"},
|
|
hint="retry"
|
|
)
|
|
|
|
assert response.success is True
|
|
assert response.message == "Operation completed successfully"
|
|
assert response.error is None
|
|
assert response.data == {"key": "value"}
|
|
assert response.hint == "retry"
|
|
|
|
def test_mcp_response_success_false_with_error(self):
|
|
"""Test MCPResponse with success=False and error message."""
|
|
response = MCPResponse(
|
|
success=False,
|
|
message=None,
|
|
error="Failed to execute command",
|
|
data=None
|
|
)
|
|
|
|
assert response.success is False
|
|
assert response.error == "Failed to execute command"
|
|
assert response.message is None
|
|
|
|
def test_mcp_response_serialization_to_json(self):
|
|
"""Test MCPResponse can be serialized to JSON."""
|
|
response = MCPResponse(
|
|
success=True,
|
|
message="Success",
|
|
data={"count": 5}
|
|
)
|
|
|
|
json_str = response.model_dump_json()
|
|
assert isinstance(json_str, str)
|
|
|
|
data = json.loads(json_str)
|
|
assert data["success"] is True
|
|
assert data["message"] == "Success"
|
|
assert data["data"]["count"] == 5
|
|
|
|
def test_mcp_response_deserialization_from_json(self):
|
|
"""Test MCPResponse can be deserialized from JSON."""
|
|
json_str = json.dumps({
|
|
"success": True,
|
|
"message": "All good",
|
|
"error": None,
|
|
"data": {"result": "ok"}
|
|
})
|
|
|
|
response = MCPResponse.model_validate_json(json_str)
|
|
|
|
assert response.success is True
|
|
assert response.message == "All good"
|
|
assert response.data == {"result": "ok"}
|
|
|
|
def test_mcp_response_hint_values(self):
|
|
"""Test MCPResponse with various hint values."""
|
|
hints = ["retry", "other_hint", None]
|
|
|
|
for hint in hints:
|
|
response = MCPResponse(success=True, hint=hint)
|
|
assert response.hint == hint
|
|
|
|
def test_mcp_response_complex_data_structure(self):
|
|
"""Test MCPResponse with nested data structures."""
|
|
complex_data = {
|
|
"items": [
|
|
{"id": 1, "name": "Item 1"},
|
|
{"id": 2, "name": "Item 2"}
|
|
],
|
|
"metadata": {
|
|
"total": 2,
|
|
"page": 1,
|
|
"nested": {
|
|
"deep": {
|
|
"value": "here"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
response = MCPResponse(success=True, data=complex_data)
|
|
|
|
assert response.data == complex_data
|
|
json_str = response.model_dump_json()
|
|
restored = MCPResponse.model_validate_json(json_str)
|
|
assert restored.data == complex_data
|
|
|
|
@pytest.mark.parametrize("success,message,error", [
|
|
(True, "OK", None),
|
|
(False, None, "Error occurred"),
|
|
(True, "Completed", "Old error"),
|
|
(False, "Message", "Error"),
|
|
])
|
|
def test_mcp_response_various_combinations(self, success, message, error):
|
|
"""Parametrized test for various field combinations."""
|
|
response = MCPResponse(success=success, message=message, error=error)
|
|
|
|
assert response.success == success
|
|
assert response.message == message
|
|
assert response.error == error
|
|
|
|
# Round-trip through JSON
|
|
json_str = response.model_dump_json()
|
|
restored = MCPResponse.model_validate_json(json_str)
|
|
assert restored.success == success
|
|
|
|
|
|
class TestToolParameterModel:
|
|
"""Test ToolParameterModel for parameter schema validation."""
|
|
|
|
def test_tool_parameter_minimal(self):
|
|
"""Test ToolParameterModel with minimal required fields."""
|
|
param = ToolParameterModel(name="input")
|
|
|
|
assert param.name == "input"
|
|
assert param.description is None
|
|
assert param.type == "string"
|
|
assert param.required is True
|
|
assert param.default_value is None
|
|
|
|
def test_tool_parameter_full_specification(self):
|
|
"""Test ToolParameterModel with all fields specified."""
|
|
param = ToolParameterModel(
|
|
name="count",
|
|
description="Number of items",
|
|
type="integer",
|
|
required=False,
|
|
default_value="10"
|
|
)
|
|
|
|
assert param.name == "count"
|
|
assert param.description == "Number of items"
|
|
assert param.type == "integer"
|
|
assert param.required is False
|
|
assert param.default_value == "10"
|
|
|
|
def test_tool_parameter_type_defaults_to_string(self):
|
|
"""Test that parameter type defaults to 'string'."""
|
|
param = ToolParameterModel(name="text")
|
|
assert param.type == "string"
|
|
|
|
def test_tool_parameter_required_defaults_to_true(self):
|
|
"""Test that required defaults to True."""
|
|
param = ToolParameterModel(name="mandatory")
|
|
assert param.required is True
|
|
|
|
def test_tool_parameter_various_types(self):
|
|
"""Test ToolParameterModel with various type specifications."""
|
|
types = ["string", "integer", "float", "boolean", "array", "object"]
|
|
|
|
for param_type in types:
|
|
param = ToolParameterModel(name="test", type=param_type)
|
|
assert param.type == param_type
|
|
|
|
def test_tool_parameter_serialization(self):
|
|
"""Test ToolParameterModel serialization to JSON."""
|
|
param = ToolParameterModel(
|
|
name="search_term",
|
|
description="What to search for",
|
|
type="string",
|
|
required=True
|
|
)
|
|
|
|
json_str = param.model_dump_json()
|
|
data = json.loads(json_str)
|
|
|
|
assert data["name"] == "search_term"
|
|
assert data["description"] == "What to search for"
|
|
assert data["type"] == "string"
|
|
assert data["required"] is True
|
|
|
|
def test_tool_parameter_deserialization(self):
|
|
"""Test ToolParameterModel deserialization from JSON."""
|
|
json_str = json.dumps({
|
|
"name": "filepath",
|
|
"description": "Path to file",
|
|
"type": "string",
|
|
"required": True,
|
|
"default_value": None
|
|
})
|
|
|
|
param = ToolParameterModel.model_validate_json(json_str)
|
|
|
|
assert param.name == "filepath"
|
|
assert param.type == "string"
|
|
|
|
def test_tool_parameter_with_default_value(self):
|
|
"""Test ToolParameterModel with default values."""
|
|
param = ToolParameterModel(
|
|
name="timeout",
|
|
type="integer",
|
|
required=False,
|
|
default_value="30"
|
|
)
|
|
|
|
assert param.default_value == "30"
|
|
assert param.required is False
|
|
|
|
@pytest.mark.parametrize("name,param_type,required", [
|
|
("api_key", "string", True),
|
|
("limit", "integer", False),
|
|
("enabled", "boolean", True),
|
|
("data", "object", False),
|
|
("items", "array", True),
|
|
])
|
|
def test_tool_parameter_combinations(self, name, param_type, required):
|
|
"""Parametrized test for various parameter specifications."""
|
|
param = ToolParameterModel(
|
|
name=name,
|
|
type=param_type,
|
|
required=required
|
|
)
|
|
|
|
assert param.name == name
|
|
assert param.type == param_type
|
|
assert param.required == required
|
|
|
|
|
|
class TestToolDefinitionModel:
|
|
"""Test ToolDefinitionModel for tool schema validation."""
|
|
|
|
def test_tool_definition_minimal(self):
|
|
"""Test ToolDefinitionModel with minimal required fields."""
|
|
tool = ToolDefinitionModel(name="read_file")
|
|
|
|
assert tool.name == "read_file"
|
|
assert tool.description is None
|
|
assert tool.structured_output is True
|
|
assert tool.requires_polling is False
|
|
assert tool.poll_action == "status"
|
|
assert tool.parameters == []
|
|
|
|
def test_tool_definition_full_specification(self):
|
|
"""Test ToolDefinitionModel with all fields specified."""
|
|
params = [
|
|
ToolParameterModel(name="path", type="string", required=True),
|
|
ToolParameterModel(name="encoding", type="string", required=False, default_value="utf-8")
|
|
]
|
|
|
|
tool = ToolDefinitionModel(
|
|
name="read_file",
|
|
description="Read contents of a file",
|
|
structured_output=True,
|
|
requires_polling=False,
|
|
poll_action="status",
|
|
parameters=params
|
|
)
|
|
|
|
assert tool.name == "read_file"
|
|
assert tool.description == "Read contents of a file"
|
|
assert len(tool.parameters) == 2
|
|
assert tool.parameters[0].name == "path"
|
|
|
|
def test_tool_definition_defaults(self):
|
|
"""Test ToolDefinitionModel default values."""
|
|
tool = ToolDefinitionModel(name="test_tool")
|
|
|
|
assert tool.structured_output is True
|
|
assert tool.requires_polling is False
|
|
assert tool.poll_action == "status"
|
|
assert tool.parameters == []
|
|
|
|
def test_tool_definition_with_polling(self):
|
|
"""Test ToolDefinitionModel for tool requiring polling."""
|
|
tool = ToolDefinitionModel(
|
|
name="long_running_task",
|
|
requires_polling=True,
|
|
poll_action="check_progress"
|
|
)
|
|
|
|
assert tool.requires_polling is True
|
|
assert tool.poll_action == "check_progress"
|
|
|
|
def test_tool_definition_with_many_parameters(self):
|
|
"""Test ToolDefinitionModel with multiple parameters."""
|
|
params = [
|
|
ToolParameterModel(name=f"param_{i}", type="string")
|
|
for i in range(5)
|
|
]
|
|
|
|
tool = ToolDefinitionModel(name="complex_tool", parameters=params)
|
|
|
|
assert len(tool.parameters) == 5
|
|
assert all(p.name.startswith("param_") for p in tool.parameters)
|
|
|
|
def test_tool_definition_serialization(self):
|
|
"""Test ToolDefinitionModel serialization to JSON."""
|
|
params = [
|
|
ToolParameterModel(name="input", type="string", required=True),
|
|
ToolParameterModel(name="format", type="string", required=False, default_value="json")
|
|
]
|
|
|
|
tool = ToolDefinitionModel(
|
|
name="process_data",
|
|
description="Process input data",
|
|
parameters=params
|
|
)
|
|
|
|
json_str = tool.model_dump_json()
|
|
data = json.loads(json_str)
|
|
|
|
assert data["name"] == "process_data"
|
|
assert len(data["parameters"]) == 2
|
|
assert data["parameters"][0]["name"] == "input"
|
|
|
|
def test_tool_definition_deserialization(self):
|
|
"""Test ToolDefinitionModel deserialization from JSON."""
|
|
json_str = json.dumps({
|
|
"name": "analyze",
|
|
"description": "Analyze data",
|
|
"structured_output": True,
|
|
"requires_polling": False,
|
|
"poll_action": "status",
|
|
"parameters": [
|
|
{
|
|
"name": "data",
|
|
"type": "string",
|
|
"required": True,
|
|
"default_value": None,
|
|
"description": None
|
|
}
|
|
]
|
|
})
|
|
|
|
tool = ToolDefinitionModel.model_validate_json(json_str)
|
|
|
|
assert tool.name == "analyze"
|
|
assert len(tool.parameters) == 1
|
|
assert tool.parameters[0].name == "data"
|
|
|
|
@pytest.mark.parametrize("name,requires_polling,poll_action", [
|
|
("instant_tool", False, "status"),
|
|
("async_tool", True, "get_result"),
|
|
("check_tool", True, "check_status"),
|
|
("simple", False, "status"),
|
|
])
|
|
def test_tool_definition_polling_combinations(self, name, requires_polling, poll_action):
|
|
"""Parametrized test for polling configurations."""
|
|
tool = ToolDefinitionModel(
|
|
name=name,
|
|
requires_polling=requires_polling,
|
|
poll_action=poll_action
|
|
)
|
|
|
|
assert tool.requires_polling == requires_polling
|
|
assert tool.poll_action == poll_action
|
|
|
|
|
|
class TestUnityInstanceInfo:
|
|
"""Test UnityInstanceInfo model for instance data representation."""
|
|
|
|
def test_unity_instance_info_minimal(self):
|
|
"""Test UnityInstanceInfo with minimal required fields."""
|
|
instance = UnityInstanceInfo(
|
|
id="MyProject@abc123",
|
|
name="MyProject",
|
|
path="/path/to/project",
|
|
hash="abc123",
|
|
port=12345,
|
|
status="running"
|
|
)
|
|
|
|
assert instance.id == "MyProject@abc123"
|
|
assert instance.name == "MyProject"
|
|
assert instance.path == "/path/to/project"
|
|
assert instance.hash == "abc123"
|
|
assert instance.port == 12345
|
|
assert instance.status == "running"
|
|
assert instance.last_heartbeat is None
|
|
assert instance.unity_version is None
|
|
|
|
def test_unity_instance_info_full_fields(self):
|
|
"""Test UnityInstanceInfo with all fields."""
|
|
now = datetime.now()
|
|
instance = UnityInstanceInfo(
|
|
id="Project@hash",
|
|
name="Project",
|
|
path="/path",
|
|
hash="hash",
|
|
port=12345,
|
|
status="running",
|
|
last_heartbeat=now,
|
|
unity_version="2022.3.0f1"
|
|
)
|
|
|
|
assert instance.last_heartbeat == now
|
|
assert instance.unity_version == "2022.3.0f1"
|
|
|
|
def test_unity_instance_info_status_values(self):
|
|
"""Test UnityInstanceInfo with various status values."""
|
|
statuses = ["running", "reloading", "offline"]
|
|
|
|
for status in statuses:
|
|
instance = UnityInstanceInfo(
|
|
id="id",
|
|
name="name",
|
|
path="/path",
|
|
hash="hash",
|
|
port=12345,
|
|
status=status
|
|
)
|
|
assert instance.status == status
|
|
|
|
def test_unity_instance_info_to_dict(self):
|
|
"""Test UnityInstanceInfo.to_dict() method."""
|
|
instance = UnityInstanceInfo(
|
|
id="Project@hash",
|
|
name="Project",
|
|
path="/path/to/project",
|
|
hash="abc123",
|
|
port=8080,
|
|
status="running"
|
|
)
|
|
|
|
dict_repr = instance.to_dict()
|
|
|
|
assert isinstance(dict_repr, dict)
|
|
assert dict_repr["id"] == "Project@hash"
|
|
assert dict_repr["name"] == "Project"
|
|
assert dict_repr["path"] == "/path/to/project"
|
|
assert dict_repr["hash"] == "abc123"
|
|
assert dict_repr["port"] == 8080
|
|
assert dict_repr["status"] == "running"
|
|
assert dict_repr["last_heartbeat"] is None
|
|
assert dict_repr["unity_version"] is None
|
|
|
|
def test_unity_instance_info_to_dict_with_heartbeat(self):
|
|
"""Test UnityInstanceInfo.to_dict() with heartbeat datetime."""
|
|
now = datetime(2024, 1, 15, 10, 30, 45)
|
|
instance = UnityInstanceInfo(
|
|
id="id",
|
|
name="name",
|
|
path="/path",
|
|
hash="hash",
|
|
port=12345,
|
|
status="running",
|
|
last_heartbeat=now
|
|
)
|
|
|
|
dict_repr = instance.to_dict()
|
|
|
|
# Should be ISO format string
|
|
assert dict_repr["last_heartbeat"] == "2024-01-15T10:30:45"
|
|
|
|
def test_unity_instance_info_serialization_to_json(self):
|
|
"""Test UnityInstanceInfo serialization to JSON."""
|
|
instance = UnityInstanceInfo(
|
|
id="MyProject@abc",
|
|
name="MyProject",
|
|
path="/path/to/project",
|
|
hash="abc",
|
|
port=8888,
|
|
status="running"
|
|
)
|
|
|
|
json_str = instance.model_dump_json()
|
|
data = json.loads(json_str)
|
|
|
|
assert data["id"] == "MyProject@abc"
|
|
assert data["port"] == 8888
|
|
|
|
def test_unity_instance_info_deserialization_from_json(self):
|
|
"""Test UnityInstanceInfo deserialization from JSON."""
|
|
json_str = json.dumps({
|
|
"id": "Project@hash123",
|
|
"name": "MyProject",
|
|
"path": "/home/user/unity/project",
|
|
"hash": "hash123",
|
|
"port": 9999,
|
|
"status": "reloading",
|
|
"last_heartbeat": "2024-01-15T10:30:45",
|
|
"unity_version": "2023.2.0f1"
|
|
})
|
|
|
|
instance = UnityInstanceInfo.model_validate_json(json_str)
|
|
|
|
assert instance.id == "Project@hash123"
|
|
assert instance.port == 9999
|
|
assert instance.status == "reloading"
|
|
assert instance.unity_version == "2023.2.0f1"
|
|
|
|
def test_unity_instance_info_round_trip_json(self):
|
|
"""Test round-trip serialization/deserialization for UnityInstanceInfo."""
|
|
original = UnityInstanceInfo(
|
|
id="TestProject@xyz789",
|
|
name="TestProject",
|
|
path="/test/path",
|
|
hash="xyz789",
|
|
port=5555,
|
|
status="offline",
|
|
unity_version="2021.3.0f1"
|
|
)
|
|
|
|
json_str = original.model_dump_json()
|
|
restored = UnityInstanceInfo.model_validate_json(json_str)
|
|
|
|
assert restored.id == original.id
|
|
assert restored.name == original.name
|
|
assert restored.path == original.path
|
|
assert restored.hash == original.hash
|
|
assert restored.port == original.port
|
|
assert restored.status == original.status
|
|
assert restored.unity_version == original.unity_version
|
|
|
|
@pytest.mark.parametrize("port,status", [
|
|
(8000, "running"),
|
|
(9000, "reloading"),
|
|
(10000, "offline"),
|
|
(65535, "running"),
|
|
(1234, "offline"),
|
|
])
|
|
def test_unity_instance_info_port_status_combinations(self, port, status):
|
|
"""Parametrized test for port and status combinations."""
|
|
instance = UnityInstanceInfo(
|
|
id="id",
|
|
name="name",
|
|
path="/path",
|
|
hash="hash",
|
|
port=port,
|
|
status=status
|
|
)
|
|
|
|
assert instance.port == port
|
|
assert instance.status == status
|
|
|
|
|
|
class TestNormalizeUnityResponse:
|
|
"""Test normalize_unity_response function for response normalization."""
|
|
|
|
def test_normalize_empty_dict(self):
|
|
"""Test normalizing empty dictionary."""
|
|
result = normalize_unity_response({})
|
|
|
|
assert result == {}
|
|
|
|
def test_normalize_already_normalized_response(self):
|
|
"""Test normalizing already MCPResponse-shaped response."""
|
|
response = {
|
|
"success": True,
|
|
"message": "OK",
|
|
"error": None,
|
|
"data": None
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result == response
|
|
assert result["success"] is True
|
|
|
|
def test_normalize_status_success_response(self):
|
|
"""Test normalizing status='success' response."""
|
|
response = {
|
|
"status": "success",
|
|
"result": {
|
|
"message": "Operation succeeded"
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is True
|
|
assert result["message"] == "Operation succeeded"
|
|
|
|
def test_normalize_status_error_response(self):
|
|
"""Test normalizing status='error' response."""
|
|
response = {
|
|
"status": "error",
|
|
"result": {
|
|
"error": "Something went wrong"
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"] == "Something went wrong"
|
|
|
|
def test_normalize_with_data_payload(self):
|
|
"""Test normalizing response with data in result."""
|
|
response = {
|
|
"status": "success",
|
|
"result": {
|
|
"message": "Retrieved data",
|
|
"data": {"id": 1, "name": "Test"}
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is True
|
|
assert result["data"]["id"] == 1
|
|
|
|
def test_normalize_non_dict_response(self):
|
|
"""Test normalizing non-dict response (should pass through)."""
|
|
response = "plain string response"
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result == response
|
|
|
|
def test_normalize_none_response(self):
|
|
"""Test normalizing None response."""
|
|
result = normalize_unity_response(None)
|
|
assert result is None
|
|
|
|
def test_normalize_list_response(self):
|
|
"""Test normalizing list response (should pass through)."""
|
|
response = [1, 2, 3]
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result == response
|
|
|
|
def test_normalize_result_with_nested_dict(self):
|
|
"""Test normalizing result field containing nested dict."""
|
|
response = {
|
|
"status": "success",
|
|
"result": {
|
|
"message": "Complex result",
|
|
"nested": {
|
|
"deep": {
|
|
"value": "found"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is True
|
|
assert result["data"]["nested"]["deep"]["value"] == "found"
|
|
|
|
def test_normalize_no_status_no_success_field(self):
|
|
"""Test normalizing response with neither status nor success field."""
|
|
response = {
|
|
"id": 123,
|
|
"name": "Some response"
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
# Should pass through unchanged
|
|
assert result == response
|
|
|
|
def test_normalize_result_field_as_string(self):
|
|
"""Test normalizing when result field is a string."""
|
|
response = {
|
|
"status": "success",
|
|
"result": "simple string result",
|
|
"message": "Operation complete"
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is True
|
|
assert result["message"] == "Operation complete"
|
|
|
|
def test_normalize_error_message_fallback(self):
|
|
"""Test error message falls back to message field."""
|
|
response = {
|
|
"status": "error",
|
|
"message": "Command failed",
|
|
"result": {}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"] == "Command failed"
|
|
|
|
def test_normalize_unknown_status(self):
|
|
"""Test normalizing response with unknown status."""
|
|
response = {
|
|
"status": "unknown_status",
|
|
"message": "Unclear what happened"
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
# Unknown status != "success" so should be failure
|
|
assert result["success"] is False
|
|
|
|
def test_normalize_result_none_value(self):
|
|
"""Test normalizing when result field is None."""
|
|
response = {
|
|
"status": "success",
|
|
"result": None,
|
|
"message": "OK but no data"
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is True
|
|
assert result["data"] is None
|
|
|
|
def test_normalize_nested_success_in_result(self):
|
|
"""Test normalizing when result itself contains 'success' field."""
|
|
response = {
|
|
"status": "pending",
|
|
"result": {
|
|
"success": True,
|
|
"message": "Inner success",
|
|
"data": {"value": 42}
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
# Should extract the inner response
|
|
assert result["success"] is True
|
|
assert result["message"] == "Inner success"
|
|
|
|
@pytest.mark.parametrize("status,expected_success", [
|
|
("success", True),
|
|
("error", False),
|
|
("failed", False),
|
|
("pending", False),
|
|
("completed", False),
|
|
])
|
|
def test_normalize_status_to_success_mapping(self, status, expected_success):
|
|
"""Parametrized test for status to success field mapping."""
|
|
response = {
|
|
"status": status,
|
|
"result": {"message": f"Status is {status}"}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] == expected_success
|
|
|
|
def test_normalize_preserves_extra_fields_in_result(self):
|
|
"""Test that extra fields in result are included in data."""
|
|
response = {
|
|
"status": "success",
|
|
"result": {
|
|
"message": "Done",
|
|
"field1": "value1",
|
|
"field2": 123,
|
|
"field3": True
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
# Extra fields should be in data
|
|
assert result["data"]["field1"] == "value1"
|
|
assert result["data"]["field2"] == 123
|
|
assert result["data"]["field3"] is True
|
|
|
|
def test_normalize_empty_result_dict(self):
|
|
"""Test normalizing response with empty result dict."""
|
|
response = {
|
|
"status": "success",
|
|
"result": {}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
assert result["success"] is True
|
|
assert result["data"] is None
|
|
|
|
def test_normalize_status_code_excluded_from_data(self):
|
|
"""Test that 'code' and 'status' fields are filtered from data."""
|
|
response = {
|
|
"status": "success",
|
|
"result": {
|
|
"message": "OK",
|
|
"code": 200,
|
|
"status": "ok",
|
|
"data": {"actual": "data"}
|
|
}
|
|
}
|
|
|
|
result = normalize_unity_response(response)
|
|
|
|
# code and status should not appear in data
|
|
assert "code" not in result["data"]
|
|
assert "status" not in result["data"]
|
|
assert result["data"]["actual"] == "data"
|
|
|
|
|
|
class TestModelValidation:
|
|
"""Test model validation and error handling."""
|
|
|
|
def test_mcp_response_missing_success_field_required(self):
|
|
"""Test that MCPResponse requires success field."""
|
|
with pytest.raises(Exception): # Pydantic ValidationError
|
|
MCPResponse.model_validate({})
|
|
|
|
def test_tool_parameter_missing_name_required(self):
|
|
"""Test that ToolParameterModel requires name field."""
|
|
with pytest.raises(Exception):
|
|
ToolParameterModel.model_validate({})
|
|
|
|
def test_tool_definition_missing_name_required(self):
|
|
"""Test that ToolDefinitionModel requires name field."""
|
|
with pytest.raises(Exception):
|
|
ToolDefinitionModel.model_validate({})
|
|
|
|
def test_unity_instance_info_missing_required_fields(self):
|
|
"""Test that UnityInstanceInfo requires all core fields."""
|
|
with pytest.raises(Exception):
|
|
UnityInstanceInfo.model_validate({})
|
|
|
|
def test_unity_instance_info_missing_single_field(self):
|
|
"""Test UnityInstanceInfo with one missing required field."""
|
|
incomplete_data = {
|
|
"id": "id",
|
|
"name": "name",
|
|
"path": "/path",
|
|
"hash": "hash",
|
|
# Missing port
|
|
"status": "running"
|
|
}
|
|
|
|
with pytest.raises(Exception):
|
|
UnityInstanceInfo.model_validate(incomplete_data)
|
|
|
|
|
|
class TestSchemaConsistency:
|
|
"""Test schema consistency and inter-model contracts."""
|
|
|
|
def test_mcp_response_with_tool_definition_as_data(self):
|
|
"""Test MCPResponse containing ToolDefinitionModel as data."""
|
|
tool = ToolDefinitionModel(
|
|
name="test_tool",
|
|
description="A test tool"
|
|
)
|
|
|
|
response = MCPResponse(
|
|
success=True,
|
|
data={
|
|
"tool": tool.model_dump()
|
|
}
|
|
)
|
|
|
|
assert response.data["tool"]["name"] == "test_tool"
|
|
|
|
def test_tool_definition_with_all_parameter_types(self):
|
|
"""Test ToolDefinitionModel can represent all parameter types."""
|
|
param_types = ["string", "integer", "float", "boolean", "array", "object"]
|
|
|
|
params = [
|
|
ToolParameterModel(name=f"param_{i}", type=ptype)
|
|
for i, ptype in enumerate(param_types)
|
|
]
|
|
|
|
tool = ToolDefinitionModel(name="multi_type_tool", parameters=params)
|
|
|
|
for i, param in enumerate(tool.parameters):
|
|
assert param.type == param_types[i]
|
|
|
|
def test_unity_instance_info_to_dict_json_roundtrip(self):
|
|
"""Test UnityInstanceInfo can be converted via to_dict() and back."""
|
|
original = UnityInstanceInfo(
|
|
id="Test@id",
|
|
name="Test",
|
|
path="/test",
|
|
hash="id",
|
|
port=9876,
|
|
status="running",
|
|
unity_version="2023.1.0f1"
|
|
)
|
|
|
|
dict_repr = original.to_dict()
|
|
json_str = json.dumps(dict_repr, default=str)
|
|
|
|
restored_dict = json.loads(json_str)
|
|
restored = UnityInstanceInfo.model_validate(restored_dict)
|
|
|
|
assert restored.id == original.id
|
|
assert restored.port == original.port
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|