c9ca1b538d
- Remove sys.path manipulation from kb_workflow.py - Update CLI imports to use kb_workflow.pipeline prefix - Update all test file imports and patch paths Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
219 lines
7.3 KiB
Python
219 lines
7.3 KiB
Python
"""Tests for error handling module
|
|
|
|
Tests for KBError, NetworkError, ValidationError, SystemError, AuthError and ErrorType.
|
|
"""
|
|
|
|
import pytest
|
|
from kb_workflow.pipeline.errors import (
|
|
KBError,
|
|
NetworkError,
|
|
ValidationError,
|
|
SystemError,
|
|
AuthError,
|
|
ErrorType,
|
|
)
|
|
|
|
|
|
class TestErrorType:
|
|
"""Tests for ErrorType enum"""
|
|
|
|
def test_error_type_has_network(self):
|
|
"""Test ErrorType contains NETWORK"""
|
|
assert hasattr(ErrorType, "NETWORK")
|
|
assert ErrorType.NETWORK.value == "network"
|
|
|
|
def test_error_type_has_validation(self):
|
|
"""Test ErrorType contains VALIDATION"""
|
|
assert hasattr(ErrorType, "VALIDATION")
|
|
assert ErrorType.VALIDATION.value == "validation"
|
|
|
|
def test_error_type_has_system(self):
|
|
"""Test ErrorType contains SYSTEM"""
|
|
assert hasattr(ErrorType, "SYSTEM")
|
|
assert ErrorType.SYSTEM.value == "system"
|
|
|
|
def test_error_type_has_auth(self):
|
|
"""Test ErrorType contains AUTH"""
|
|
assert hasattr(ErrorType, "AUTH")
|
|
assert ErrorType.AUTH.value == "auth"
|
|
|
|
|
|
class TestKBError:
|
|
"""Tests for KBError base class"""
|
|
|
|
def test_kberror_contains_all_fields(self):
|
|
"""Test KBError contains all required fields"""
|
|
error = KBError(
|
|
message="Test error",
|
|
error_type=ErrorType.NETWORK,
|
|
step_name="test_step",
|
|
input_summary="input_id=123",
|
|
suggestion="Check your network",
|
|
)
|
|
assert error.step_name == "test_step"
|
|
assert error.input_summary == "input_id=123"
|
|
assert error.suggestion == "Check your network"
|
|
assert error.error_type == ErrorType.NETWORK
|
|
|
|
def test_kberror_to_dict_returns_correct_structure(self):
|
|
"""Test KBError.to_dict() returns correct dict"""
|
|
error = KBError(
|
|
message="Test error message",
|
|
error_type=ErrorType.NETWORK,
|
|
step_name="fetch",
|
|
input_summary="url=http://example.com",
|
|
suggestion="Check your network",
|
|
)
|
|
result = error.to_dict()
|
|
assert "step_name" in result
|
|
assert "error_type" in result
|
|
assert "suggestion" in result
|
|
assert "message" in result
|
|
assert "input_summary" in result
|
|
assert result["step_name"] == "fetch"
|
|
assert result["error_type"] == "network"
|
|
|
|
def test_kberror_with_cause(self):
|
|
"""Test KBError stores cause exception"""
|
|
original_error = ValueError("Original error")
|
|
error = KBError(
|
|
message="Wrapped error",
|
|
error_type=ErrorType.SYSTEM,
|
|
step_name="process",
|
|
input_summary="file=test.txt",
|
|
suggestion="Check the file",
|
|
cause=original_error,
|
|
)
|
|
assert error.cause is original_error
|
|
|
|
|
|
class TestNetworkError:
|
|
"""Tests for NetworkError"""
|
|
|
|
def test_network_error_includes_status_code(self):
|
|
"""Test NetworkError includes status_code"""
|
|
error = NetworkError(
|
|
step_name="fetch",
|
|
input_summary="url=http://example.com",
|
|
status_code=403,
|
|
)
|
|
assert error.status_code == 403
|
|
assert "403" in error.suggestion
|
|
|
|
def test_network_error_without_status_code(self):
|
|
"""Test NetworkError without status_code"""
|
|
error = NetworkError(
|
|
step_name="fetch",
|
|
input_summary="url=http://example.com",
|
|
)
|
|
assert error.status_code is None
|
|
assert "网络" in error.suggestion or "network" in error.suggestion.lower()
|
|
|
|
def test_network_error_is_kberror(self):
|
|
"""Test NetworkError is subclass of KBError"""
|
|
assert issubclass(NetworkError, KBError)
|
|
|
|
def test_network_error_has_correct_error_type(self):
|
|
"""Test NetworkError has NETWORK error type"""
|
|
error = NetworkError(
|
|
step_name="fetch",
|
|
input_summary="url=http://example.com",
|
|
)
|
|
assert error.error_type == ErrorType.NETWORK
|
|
|
|
|
|
class TestValidationError:
|
|
"""Tests for ValidationError"""
|
|
|
|
def test_validation_error_stores_validation_errors(self):
|
|
"""Test ValidationError stores validation_errors list"""
|
|
error = ValidationError(
|
|
step_name="validate",
|
|
input_summary="data={'key': 'value'}",
|
|
validation_errors=["Missing field: id", "Invalid type for: name"],
|
|
)
|
|
assert error.validation_errors == ["Missing field: id", "Invalid type for: name"]
|
|
|
|
def test_validation_error_default_empty_validation_errors(self):
|
|
"""Test ValidationError has empty list by default"""
|
|
error = ValidationError(
|
|
step_name="validate",
|
|
input_summary="data={'key': 'value'}",
|
|
)
|
|
assert error.validation_errors == []
|
|
|
|
def test_validation_error_is_kberror(self):
|
|
"""Test ValidationError is subclass of KBError"""
|
|
assert issubclass(ValidationError, KBError)
|
|
|
|
def test_validation_error_has_correct_error_type(self):
|
|
"""Test ValidationError has VALIDATION error type"""
|
|
error = ValidationError(
|
|
step_name="validate",
|
|
input_summary="data={'key': 'value'}",
|
|
)
|
|
assert error.error_type == ErrorType.VALIDATION
|
|
|
|
|
|
class TestSystemError:
|
|
"""Tests for SystemError"""
|
|
|
|
def test_system_error_with_system_message(self):
|
|
"""Test SystemError with system_message"""
|
|
error = SystemError(
|
|
step_name="file_read",
|
|
input_summary="path=/tmp/file.txt",
|
|
system_message="File not found",
|
|
)
|
|
assert "File not found" in str(error)
|
|
assert error.error_type == ErrorType.SYSTEM
|
|
|
|
def test_system_error_is_kberror(self):
|
|
"""Test SystemError is subclass of KBError"""
|
|
assert issubclass(SystemError, KBError)
|
|
|
|
|
|
class TestAuthError:
|
|
"""Tests for AuthError"""
|
|
|
|
def test_auth_error_with_auth_type(self):
|
|
"""Test AuthError with auth_type"""
|
|
error = AuthError(
|
|
step_name="api_call",
|
|
input_summary="endpoint=/api/data",
|
|
auth_type="API Token",
|
|
)
|
|
assert error.auth_type == "API Token"
|
|
assert "API Token" in str(error) or "Token" in error.suggestion
|
|
|
|
def test_auth_error_is_kberror(self):
|
|
"""Test AuthError is subclass of KBError"""
|
|
assert issubclass(AuthError, KBError)
|
|
|
|
def test_auth_error_has_correct_error_type(self):
|
|
"""Test AuthError has AUTH error type"""
|
|
error = AuthError(
|
|
step_name="api_call",
|
|
input_summary="endpoint=/api/data",
|
|
)
|
|
assert error.error_type == ErrorType.AUTH
|
|
|
|
|
|
class TestErrorHierarchy:
|
|
"""Tests for error class hierarchy"""
|
|
|
|
def test_all_errors_inherit_from_kberror(self):
|
|
"""Test all error classes inherit from KBError"""
|
|
assert issubclass(NetworkError, KBError)
|
|
assert issubclass(ValidationError, KBError)
|
|
assert issubclass(SystemError, KBError)
|
|
assert issubclass(AuthError, KBError)
|
|
|
|
def test_all_errors_are_exceptions(self):
|
|
"""Test all error classes are Exception subclasses"""
|
|
assert issubclass(KBError, Exception)
|
|
assert issubclass(NetworkError, Exception)
|
|
assert issubclass(ValidationError, Exception)
|
|
assert issubclass(SystemError, Exception)
|
|
assert issubclass(AuthError, Exception)
|