""" 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"])