File size: 13,176 Bytes
dbb04e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 | """
Tests for MnemoCore Error Handling
===================================
Tests the exception hierarchy, error codes, and FastAPI integration.
"""
import pytest
import os
import sys
# Add parent to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from mnemocore.core.exceptions import (
# Base
MnemoCoreError,
RecoverableError,
IrrecoverableError,
ErrorCategory,
# Storage
StorageError,
StorageConnectionError,
StorageTimeoutError,
DataCorruptionError,
# Vector
VectorError,
DimensionMismatchError,
VectorOperationError,
# Config
ConfigurationError,
# Circuit Breaker
CircuitOpenError,
# Memory
MemoryOperationError,
# Validation
ValidationError,
MetadataValidationError,
AttributeValidationError,
# Not Found
NotFoundError,
AgentNotFoundError,
MemoryNotFoundError,
# Provider
ProviderError,
UnsupportedProviderError,
UnsupportedTransportError,
DependencyMissingError,
# Utilities
wrap_storage_exception,
is_debug_mode,
)
class TestExceptionHierarchy:
"""Test the exception inheritance hierarchy."""
def test_base_exception(self):
"""Test MnemoCoreError base class."""
exc = MnemoCoreError("Test error")
assert str(exc) == "Test error"
assert exc.message == "Test error"
assert exc.context == {}
assert exc.recoverable is True
assert exc.error_code == "MNEMO_CORE_ERROR"
def test_exception_with_context(self):
"""Test exception with context."""
exc = MnemoCoreError("Test error", context={"key": "value"})
assert exc.context == {"key": "value"}
assert "context=" in str(exc)
def test_exception_to_dict(self):
"""Test to_dict conversion."""
exc = ValidationError(
field="test_field",
reason="Invalid value",
value="bad_data"
)
d = exc.to_dict()
assert d["error"] == "Validation error for 'test_field': Invalid value"
assert d["code"] == "VALIDATION_ERROR"
assert d["recoverable"] is False
assert "traceback" not in d
def test_exception_to_dict_with_traceback(self):
"""Test to_dict with traceback in debug mode."""
exc = ValidationError(field="test", reason="test")
d = exc.to_dict(include_traceback=True)
assert "traceback" in d
class TestRecoverableErrors:
"""Test recoverable error classes."""
def test_storage_connection_error_is_recoverable(self):
"""Storage connection errors should be recoverable."""
exc = StorageConnectionError("redis", "Connection refused")
assert exc.recoverable is True
assert exc.error_code == "STORAGE_CONNECTION_ERROR"
assert exc.backend == "redis"
def test_storage_timeout_error_is_recoverable(self):
"""Storage timeout errors should be recoverable."""
exc = StorageTimeoutError("qdrant", "search", timeout_ms=5000)
assert exc.recoverable is True
assert exc.error_code == "STORAGE_TIMEOUT_ERROR"
assert exc.backend == "qdrant"
assert exc.operation == "search"
assert exc.context["timeout_ms"] == 5000
def test_circuit_open_error_is_recoverable(self):
"""Circuit breaker open errors should be recoverable."""
exc = CircuitOpenError("storage", failures=5)
assert exc.recoverable is True
assert exc.error_code == "CIRCUIT_OPEN_ERROR"
assert exc.breaker_name == "storage"
assert exc.failures == 5
class TestIrrecoverableErrors:
"""Test irrecoverable error classes."""
def test_validation_error_is_irrecoverable(self):
"""Validation errors should be irrecoverable."""
exc = ValidationError(field="content", reason="Cannot be empty")
assert exc.recoverable is False
assert exc.error_code == "VALIDATION_ERROR"
assert exc.field == "content"
def test_configuration_error_is_irrecoverable(self):
"""Configuration errors should be irrecoverable."""
exc = ConfigurationError("api_key", "Missing required key")
assert exc.recoverable is False
assert exc.error_code == "CONFIGURATION_ERROR"
assert exc.config_key == "api_key"
def test_data_corruption_error_is_irrecoverable(self):
"""Data corruption errors should be irrecoverable."""
exc = DataCorruptionError("mem_123", "Invalid checksum")
assert exc.recoverable is False
assert exc.error_code == "DATA_CORRUPTION_ERROR"
assert exc.resource_id == "mem_123"
def test_not_found_errors_are_irrecoverable(self):
"""Not found errors should be irrecoverable."""
exc = MemoryNotFoundError("mem_123")
assert exc.recoverable is False
assert exc.error_code == "MEMORY_NOT_FOUND_ERROR"
exc2 = AgentNotFoundError("agent_456")
assert exc2.recoverable is False
assert exc2.error_code == "AGENT_NOT_FOUND_ERROR"
def test_unsupported_provider_error_is_irrecoverable(self):
"""Unsupported provider errors should be irrecoverable."""
exc = UnsupportedProviderError("unknown", supported_providers=["openai", "anthropic"])
assert exc.recoverable is False
assert exc.error_code == "UNSUPPORTED_PROVIDER_ERROR"
assert exc.provider == "unknown"
assert "openai" in str(exc)
class TestVectorErrors:
"""Test vector-related errors."""
def test_dimension_mismatch_error(self):
"""Test dimension mismatch error."""
exc = DimensionMismatchError(expected=16384, actual=10000, operation="encode")
assert exc.recoverable is False
assert exc.error_code == "DIMENSION_MISMATCH_ERROR"
assert exc.expected == 16384
assert exc.actual == 10000
assert "16384" in str(exc)
assert "10000" in str(exc)
def test_vector_operation_error(self):
"""Test vector operation error."""
exc = VectorOperationError("bundle", "NaN detected")
assert exc.recoverable is False
assert exc.error_code == "VECTOR_OPERATION_ERROR"
assert exc.operation == "bundle"
class TestStorageErrorWrapper:
"""Test wrap_storage_exception utility."""
def test_wrap_timeout_exception(self):
"""Timeout exceptions should be wrapped as StorageTimeoutError."""
exc = Exception("Connection timeout after 5000ms")
wrapped = wrap_storage_exception("redis", "get", exc)
assert isinstance(wrapped, StorageTimeoutError)
assert wrapped.backend == "redis"
assert wrapped.operation == "get"
def test_wrap_connection_exception(self):
"""Connection exceptions should be wrapped as StorageConnectionError."""
# Create a mock exception with 'Connection' in the class name
class ConnectionRefusedError(Exception):
pass
exc = ConnectionRefusedError("Connection refused")
wrapped = wrap_storage_exception("qdrant", "search", exc)
assert isinstance(wrapped, StorageConnectionError)
assert wrapped.backend == "qdrant"
def test_wrap_generic_exception(self):
"""Generic exceptions should be wrapped as StorageError."""
exc = Exception("Unknown error")
wrapped = wrap_storage_exception("redis", "set", exc)
assert isinstance(wrapped, StorageError)
assert "redis" in str(wrapped)
assert "set" in str(wrapped)
class TestDebugMode:
"""Test debug mode detection."""
def test_debug_mode_off_by_default(self):
"""Debug mode should be off by default."""
# Save and clear env
old_val = os.environ.get("MNEMO_DEBUG")
if "MNEMO_DEBUG" in os.environ:
del os.environ["MNEMO_DEBUG"]
try:
assert is_debug_mode() is False
finally:
if old_val:
os.environ["MNEMO_DEBUG"] = old_val
def test_debug_mode_on_with_true(self):
"""Debug mode should be on when set to 'true'."""
old_val = os.environ.get("MNEMO_DEBUG")
os.environ["MNEMO_DEBUG"] = "true"
try:
assert is_debug_mode() is True
finally:
if old_val:
os.environ["MNEMO_DEBUG"] = old_val
else:
del os.environ["MNEMO_DEBUG"]
def test_debug_mode_on_with_1(self):
"""Debug mode should be on when set to '1'."""
old_val = os.environ.get("MNEMO_DEBUG")
os.environ["MNEMO_DEBUG"] = "1"
try:
assert is_debug_mode() is True
finally:
if old_val:
os.environ["MNEMO_DEBUG"] = old_val
else:
del os.environ["MNEMO_DEBUG"]
class TestErrorCategories:
"""Test error category classification."""
def test_storage_error_category(self):
"""Storage errors should have STORAGE category."""
exc = StorageError("test")
assert exc.category == ErrorCategory.STORAGE
def test_vector_error_category(self):
"""Vector errors should have VECTOR category."""
exc = VectorError("test")
assert exc.category == ErrorCategory.VECTOR
def test_config_error_category(self):
"""Config errors should have CONFIG category."""
exc = ConfigurationError("key", "reason")
assert exc.category == ErrorCategory.CONFIG
def test_validation_error_category(self):
"""Validation errors should have VALIDATION category."""
exc = ValidationError("field", "reason")
assert exc.category == ErrorCategory.VALIDATION
def test_memory_error_category(self):
"""Memory errors should have MEMORY category."""
exc = MemoryOperationError("store", "mem_1", "failed")
assert exc.category == ErrorCategory.MEMORY
def test_agent_error_category(self):
"""Agent errors should have AGENT category."""
exc = AgentNotFoundError("agent_1")
assert exc.category == ErrorCategory.AGENT
def test_provider_error_category(self):
"""Provider errors should have PROVIDER category."""
exc = UnsupportedProviderError("unknown")
assert exc.category == ErrorCategory.PROVIDER
class TestMetadataValidationErrors:
"""Test specialized validation errors."""
def test_metadata_validation_error(self):
"""Test metadata validation error."""
exc = MetadataValidationError("metadata", "Too many keys")
assert exc.error_code == "METADATA_VALIDATION_ERROR"
assert exc.recoverable is False
def test_attribute_validation_error(self):
"""Test attribute validation error."""
exc = AttributeValidationError("attributes", "Key too long")
assert exc.error_code == "ATTRIBUTE_VALIDATION_ERROR"
assert exc.recoverable is False
class TestUnsupportedTransportError:
"""Test unsupported transport error."""
def test_unsupported_transport_error(self):
"""Test unsupported transport error."""
exc = UnsupportedTransportError(
transport="websocket",
supported_transports=["stdio", "sse"]
)
assert exc.recoverable is False
assert exc.error_code == "UNSUPPORTED_TRANSPORT_ERROR"
assert exc.transport == "websocket"
assert "stdio" in str(exc)
assert "sse" in str(exc)
class TestDependencyMissingError:
"""Test dependency missing error."""
def test_dependency_missing_error(self):
"""Test dependency missing error."""
exc = DependencyMissingError(
dependency="mcp",
message="Install with: pip install mcp"
)
assert exc.recoverable is False
assert exc.error_code == "DEPENDENCY_MISSING_ERROR"
assert exc.dependency == "mcp"
assert "pip install mcp" in str(exc)
class TestErrorContext:
"""Test error context handling."""
def test_context_preserved_in_subclass(self):
"""Context should be preserved in subclasses."""
exc = StorageConnectionError(
backend="redis",
message="Connection failed",
context={"retry_count": 3, "last_error": "ECONNREFUSED"}
)
assert exc.context["retry_count"] == 3
assert exc.context["last_error"] == "ECONNREFUSED"
assert exc.context["backend"] == "redis"
def test_value_truncation_in_validation_error(self):
"""Large values should be truncated in validation error context."""
large_value = "x" * 200
exc = ValidationError("field", "too long", value=large_value)
assert len(exc.context["value"]) == 103 # 100 + "..."
# Run tests
if __name__ == "__main__":
pytest.main([__file__, "-v"])
|