Travis Vasceannie
3acb779569
- Deleted .env.example file as it is no longer needed. - Added .gitignore to manage ignored files and directories. - Introduced CLAUDE.md for AI provider integration documentation. - Created dev.sh for development setup and scripts. - Updated Dockerfile and Dockerfile.production for improved build processes. - Added multiple test files and directories for comprehensive testing. - Introduced new utility and service files for enhanced functionality. - Organized codebase with new directories and files for better maintainability.
826 lines
31 KiB
Python
826 lines
31 KiB
Python
"""
|
|
Comprehensive tests for utils.exceptions module
|
|
|
|
Tests standardized exception hierarchy including:
|
|
- BotBaseException base class
|
|
- Audio processing exceptions
|
|
- Permission exceptions
|
|
- Metrics exceptions
|
|
- Prompt building exceptions
|
|
- Discord API exceptions
|
|
- External service exceptions
|
|
- Resource exceptions
|
|
- Input validation exceptions
|
|
- Network exceptions
|
|
- Timeout exceptions
|
|
- Configuration exceptions
|
|
"""
|
|
|
|
import discord
|
|
import pytest
|
|
|
|
from utils.exceptions import ( # Base exception; Audio processing exceptions; Permission exceptions; Metrics exceptions; Prompt building exceptions; Discord API exceptions; External service exceptions; Resource exceptions; Input validation exceptions; Network exceptions; Timeout exceptions; Configuration exceptions
|
|
AIProviderError, AnthropicError, AudioCompressionError,
|
|
AudioConversionError, AudioFormatError, AudioProcessingError,
|
|
AudioValidationError, BotBaseException, BotPermissionError,
|
|
ConfigurationError, ConnectionRefusedError, ConnectionTimeoutError,
|
|
DatabaseConnectionError, DatabaseError, DatabaseQueryError,
|
|
DiscordAPIError, DiscordConnectionError, DiscordRateLimitError,
|
|
DiscordTimeoutError, ExternalServiceError, FFmpegError,
|
|
InsufficientPermissionsError, InvalidConfigurationError, InvalidInputError,
|
|
MetricsCollectionError, MetricsError, MetricsExportError,
|
|
MissingConfigurationError, MissingRequiredFieldError, NetworkError,
|
|
OpenAIError, OperationTimeoutError, PermissionError, PrometheusError,
|
|
PromptError, PromptFormattingError, PromptTemplateError,
|
|
PromptVariableError, ResourceError, ResourceExhaustedError,
|
|
ResourceNotFoundError, SystemMetricsError, TemporaryFileError,
|
|
TimeoutError, ValidationError, VoiceActivityDetectionError,
|
|
VoiceChannelPermissionError)
|
|
|
|
|
|
class TestBotBaseException:
|
|
"""Test BotBaseException base class."""
|
|
|
|
def test_bot_base_exception_creation(self):
|
|
"""Test BotBaseException creation with all parameters."""
|
|
metadata = {"key": "value", "number": 42}
|
|
exception = BotBaseException(
|
|
message="Test error",
|
|
component="test_component",
|
|
operation="test_operation",
|
|
metadata=metadata,
|
|
)
|
|
|
|
assert str(exception) == "Test error"
|
|
assert exception.message == "Test error"
|
|
assert exception.component == "test_component"
|
|
assert exception.operation == "test_operation"
|
|
assert exception.metadata == metadata
|
|
|
|
def test_bot_base_exception_minimal(self):
|
|
"""Test BotBaseException creation with minimal parameters."""
|
|
exception = BotBaseException("Simple error", "test_component")
|
|
|
|
assert str(exception) == "Simple error"
|
|
assert exception.component == "test_component"
|
|
assert exception.operation is None
|
|
assert exception.metadata == {}
|
|
|
|
def test_bot_base_exception_default_metadata(self):
|
|
"""Test BotBaseException with None metadata becomes empty dict."""
|
|
exception = BotBaseException("Test", "component", metadata=None)
|
|
|
|
assert exception.metadata == {}
|
|
|
|
def test_bot_base_exception_inheritance(self):
|
|
"""Test that BotBaseException properly inherits from Exception."""
|
|
exception = BotBaseException("Test", "component")
|
|
|
|
assert isinstance(exception, Exception)
|
|
assert isinstance(exception, BotBaseException)
|
|
|
|
|
|
class TestAudioProcessingExceptions:
|
|
"""Test audio processing exception hierarchy."""
|
|
|
|
def test_audio_processing_error_base(self):
|
|
"""Test AudioProcessingError as base for audio exceptions."""
|
|
exception = AudioProcessingError("Audio error", "audio_processor")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, AudioProcessingError)
|
|
|
|
def test_audio_format_error(self):
|
|
"""Test AudioFormatError specialization."""
|
|
exception = AudioFormatError("Invalid format", "audio_validator")
|
|
|
|
assert isinstance(exception, AudioProcessingError)
|
|
assert isinstance(exception, AudioFormatError)
|
|
|
|
def test_audio_conversion_error(self):
|
|
"""Test AudioConversionError specialization."""
|
|
exception = AudioConversionError("Conversion failed", "audio_converter")
|
|
|
|
assert isinstance(exception, AudioProcessingError)
|
|
assert isinstance(exception, AudioConversionError)
|
|
|
|
def test_audio_validation_error(self):
|
|
"""Test AudioValidationError specialization."""
|
|
exception = AudioValidationError("Validation failed", "audio_validator")
|
|
|
|
assert isinstance(exception, AudioProcessingError)
|
|
assert isinstance(exception, AudioValidationError)
|
|
|
|
def test_audio_compression_error(self):
|
|
"""Test AudioCompressionError specialization."""
|
|
exception = AudioCompressionError("Compression failed", "audio_compressor")
|
|
|
|
assert isinstance(exception, AudioProcessingError)
|
|
assert isinstance(exception, AudioCompressionError)
|
|
|
|
def test_ffmpeg_error(self):
|
|
"""Test FFmpegError specialization."""
|
|
exception = FFmpegError("FFmpeg failed", "audio_converter")
|
|
|
|
assert isinstance(exception, AudioProcessingError)
|
|
assert isinstance(exception, FFmpegError)
|
|
|
|
def test_voice_activity_detection_error(self):
|
|
"""Test VoiceActivityDetectionError specialization."""
|
|
exception = VoiceActivityDetectionError("VAD failed", "audio_preprocessor")
|
|
|
|
assert isinstance(exception, AudioProcessingError)
|
|
assert isinstance(exception, VoiceActivityDetectionError)
|
|
|
|
|
|
class TestPermissionExceptions:
|
|
"""Test permission exception hierarchy."""
|
|
|
|
def test_permission_error_base(self):
|
|
"""Test PermissionError as base for permission exceptions."""
|
|
exception = PermissionError("Permission denied", "permissions")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, PermissionError)
|
|
|
|
def test_insufficient_permissions_error(self):
|
|
"""Test InsufficientPermissionsError with full context."""
|
|
# Create mock Discord objects
|
|
mock_user = discord.Object(id=12345)
|
|
mock_guild = discord.Object(id=67890)
|
|
|
|
exception = InsufficientPermissionsError(
|
|
message="User lacks permissions",
|
|
required_permissions=["manage_messages", "kick_members"],
|
|
user=mock_user,
|
|
guild=mock_guild,
|
|
component="permissions",
|
|
operation="check_moderator",
|
|
)
|
|
|
|
assert isinstance(exception, PermissionError)
|
|
assert exception.required_permissions == ["manage_messages", "kick_members"]
|
|
assert exception.user == mock_user
|
|
assert exception.guild == mock_guild
|
|
assert exception.metadata["required_permissions"] == [
|
|
"manage_messages",
|
|
"kick_members",
|
|
]
|
|
assert exception.metadata["user_id"] == 12345
|
|
assert exception.metadata["guild_id"] == 67890
|
|
|
|
def test_insufficient_permissions_error_minimal(self):
|
|
"""Test InsufficientPermissionsError with minimal parameters."""
|
|
exception = InsufficientPermissionsError(
|
|
message="Permission denied", required_permissions=["admin"]
|
|
)
|
|
|
|
assert exception.required_permissions == ["admin"]
|
|
assert exception.user is None
|
|
assert exception.guild is None
|
|
assert exception.metadata["user_id"] is None
|
|
assert exception.metadata["guild_id"] is None
|
|
|
|
def test_bot_permission_error(self):
|
|
"""Test BotPermissionError specialization."""
|
|
mock_guild = discord.Object(id=67890)
|
|
|
|
exception = BotPermissionError(
|
|
message="Bot lacks permissions",
|
|
required_permissions=["send_messages", "embed_links"],
|
|
guild=mock_guild,
|
|
)
|
|
|
|
assert isinstance(exception, PermissionError)
|
|
assert exception.required_permissions == ["send_messages", "embed_links"]
|
|
assert exception.guild == mock_guild
|
|
assert exception.metadata["guild_id"] == 67890
|
|
|
|
def test_voice_channel_permission_error(self):
|
|
"""Test VoiceChannelPermissionError specialization."""
|
|
mock_channel = discord.Object(id=11111)
|
|
mock_user = discord.Object(id=22222)
|
|
|
|
exception = VoiceChannelPermissionError(
|
|
message="Voice channel access denied", channel=mock_channel, user=mock_user
|
|
)
|
|
|
|
assert isinstance(exception, PermissionError)
|
|
assert exception.channel == mock_channel
|
|
assert exception.user == mock_user
|
|
assert exception.metadata["channel_id"] == 11111
|
|
assert exception.metadata["user_id"] == 22222
|
|
|
|
|
|
class TestMetricsExceptions:
|
|
"""Test metrics exception hierarchy."""
|
|
|
|
def test_metrics_error_base(self):
|
|
"""Test MetricsError as base for metrics exceptions."""
|
|
exception = MetricsError("Metrics error", "metrics_collector")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, MetricsError)
|
|
|
|
def test_metrics_collection_error(self):
|
|
"""Test MetricsCollectionError specialization."""
|
|
exception = MetricsCollectionError("Collection failed", "metrics_collector")
|
|
|
|
assert isinstance(exception, MetricsError)
|
|
assert isinstance(exception, MetricsCollectionError)
|
|
|
|
def test_metrics_export_error(self):
|
|
"""Test MetricsExportError specialization."""
|
|
exception = MetricsExportError("Export failed", "metrics_collector")
|
|
|
|
assert isinstance(exception, MetricsError)
|
|
assert isinstance(exception, MetricsExportError)
|
|
|
|
def test_prometheus_error(self):
|
|
"""Test PrometheusError specialization."""
|
|
exception = PrometheusError("Prometheus failed", "metrics_collector")
|
|
|
|
assert isinstance(exception, MetricsError)
|
|
assert isinstance(exception, PrometheusError)
|
|
|
|
def test_system_metrics_error(self):
|
|
"""Test SystemMetricsError specialization."""
|
|
exception = SystemMetricsError("System metrics failed", "metrics_collector")
|
|
|
|
assert isinstance(exception, MetricsError)
|
|
assert isinstance(exception, SystemMetricsError)
|
|
|
|
|
|
class TestPromptExceptions:
|
|
"""Test prompt building exception hierarchy."""
|
|
|
|
def test_prompt_error_base(self):
|
|
"""Test PromptError as base for prompt exceptions."""
|
|
exception = PromptError("Prompt error", "prompt_builder")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, PromptError)
|
|
|
|
def test_prompt_template_error(self):
|
|
"""Test PromptTemplateError specialization."""
|
|
exception = PromptTemplateError("Template not found", "prompt_builder")
|
|
|
|
assert isinstance(exception, PromptError)
|
|
assert isinstance(exception, PromptTemplateError)
|
|
|
|
def test_prompt_variable_error(self):
|
|
"""Test PromptVariableError with missing variables."""
|
|
exception = PromptVariableError(
|
|
message="Missing variables",
|
|
missing_variables=["quote", "speaker_name"],
|
|
component="prompt_builder",
|
|
operation="build_analysis_prompt",
|
|
)
|
|
|
|
assert isinstance(exception, PromptError)
|
|
assert exception.missing_variables == ["quote", "speaker_name"]
|
|
assert exception.metadata["missing_variables"] == ["quote", "speaker_name"]
|
|
|
|
def test_prompt_formatting_error(self):
|
|
"""Test PromptFormattingError specialization."""
|
|
exception = PromptFormattingError("Format failed", "prompt_builder")
|
|
|
|
assert isinstance(exception, PromptError)
|
|
assert isinstance(exception, PromptFormattingError)
|
|
|
|
|
|
class TestDiscordAPIExceptions:
|
|
"""Test Discord API exception hierarchy."""
|
|
|
|
def test_discord_api_error_base(self):
|
|
"""Test DiscordAPIError as base for Discord exceptions."""
|
|
exception = DiscordAPIError("Discord API error", "discord_client")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, DiscordAPIError)
|
|
|
|
def test_discord_rate_limit_error(self):
|
|
"""Test DiscordRateLimitError with rate limit context."""
|
|
exception = DiscordRateLimitError(
|
|
message="Rate limited",
|
|
retry_after=30.5,
|
|
endpoint="/api/v9/channels/123/messages",
|
|
component="discord_client",
|
|
operation="send_message",
|
|
)
|
|
|
|
assert isinstance(exception, DiscordAPIError)
|
|
assert exception.retry_after == 30.5
|
|
assert exception.endpoint == "/api/v9/channels/123/messages"
|
|
assert exception.metadata["retry_after"] == 30.5
|
|
assert exception.metadata["endpoint"] == "/api/v9/channels/123/messages"
|
|
|
|
def test_discord_rate_limit_error_minimal(self):
|
|
"""Test DiscordRateLimitError with minimal parameters."""
|
|
exception = DiscordRateLimitError("Rate limited")
|
|
|
|
assert exception.retry_after is None
|
|
assert exception.endpoint is None
|
|
|
|
def test_discord_connection_error(self):
|
|
"""Test DiscordConnectionError specialization."""
|
|
exception = DiscordConnectionError("Connection lost", "discord_client")
|
|
|
|
assert isinstance(exception, DiscordAPIError)
|
|
assert isinstance(exception, DiscordConnectionError)
|
|
|
|
def test_discord_timeout_error(self):
|
|
"""Test DiscordTimeoutError specialization."""
|
|
exception = DiscordTimeoutError("Request timed out", "discord_client")
|
|
|
|
assert isinstance(exception, DiscordAPIError)
|
|
assert isinstance(exception, DiscordTimeoutError)
|
|
|
|
|
|
class TestExternalServiceExceptions:
|
|
"""Test external service exception hierarchy."""
|
|
|
|
def test_external_service_error_base(self):
|
|
"""Test ExternalServiceError as base for service exceptions."""
|
|
exception = ExternalServiceError("Service error", "external_client")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, ExternalServiceError)
|
|
|
|
def test_ai_provider_error_base(self):
|
|
"""Test AIProviderError as base for AI service exceptions."""
|
|
exception = AIProviderError("AI service error", "ai_manager")
|
|
|
|
assert isinstance(exception, ExternalServiceError)
|
|
assert isinstance(exception, AIProviderError)
|
|
|
|
def test_openai_error(self):
|
|
"""Test OpenAIError specialization."""
|
|
exception = OpenAIError("OpenAI API error", "ai_manager")
|
|
|
|
assert isinstance(exception, AIProviderError)
|
|
assert isinstance(exception, OpenAIError)
|
|
|
|
def test_anthropic_error(self):
|
|
"""Test AnthropicError specialization."""
|
|
exception = AnthropicError("Anthropic API error", "ai_manager")
|
|
|
|
assert isinstance(exception, AIProviderError)
|
|
assert isinstance(exception, AnthropicError)
|
|
|
|
def test_database_error_base(self):
|
|
"""Test DatabaseError as base for database exceptions."""
|
|
exception = DatabaseError("Database error", "database_manager")
|
|
|
|
assert isinstance(exception, ExternalServiceError)
|
|
assert isinstance(exception, DatabaseError)
|
|
|
|
def test_database_connection_error(self):
|
|
"""Test DatabaseConnectionError specialization."""
|
|
exception = DatabaseConnectionError("Connection failed", "database_manager")
|
|
|
|
assert isinstance(exception, DatabaseError)
|
|
assert isinstance(exception, DatabaseConnectionError)
|
|
|
|
def test_database_query_error(self):
|
|
"""Test DatabaseQueryError specialization."""
|
|
exception = DatabaseQueryError("Query failed", "database_manager")
|
|
|
|
assert isinstance(exception, DatabaseError)
|
|
assert isinstance(exception, DatabaseQueryError)
|
|
|
|
|
|
class TestResourceExceptions:
|
|
"""Test resource exception hierarchy."""
|
|
|
|
def test_resource_error_base(self):
|
|
"""Test ResourceError as base for resource exceptions."""
|
|
exception = ResourceError("Resource error", "resource_manager")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, ResourceError)
|
|
|
|
def test_resource_not_found_error(self):
|
|
"""Test ResourceNotFoundError specialization."""
|
|
exception = ResourceNotFoundError("Resource not found", "file_manager")
|
|
|
|
assert isinstance(exception, ResourceError)
|
|
assert isinstance(exception, ResourceNotFoundError)
|
|
|
|
def test_resource_exhausted_error(self):
|
|
"""Test ResourceExhaustedError specialization."""
|
|
exception = ResourceExhaustedError("Resources exhausted", "memory_manager")
|
|
|
|
assert isinstance(exception, ResourceError)
|
|
assert isinstance(exception, ResourceExhaustedError)
|
|
|
|
def test_temporary_file_error(self):
|
|
"""Test TemporaryFileError specialization."""
|
|
exception = TemporaryFileError("Temp file error", "file_processor")
|
|
|
|
assert isinstance(exception, ResourceError)
|
|
assert isinstance(exception, TemporaryFileError)
|
|
|
|
|
|
class TestValidationExceptions:
|
|
"""Test input validation exception hierarchy."""
|
|
|
|
def test_validation_error_base(self):
|
|
"""Test ValidationError as base for validation exceptions."""
|
|
exception = ValidationError("Validation failed", "validator")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, ValidationError)
|
|
|
|
def test_invalid_input_error(self):
|
|
"""Test InvalidInputError specialization."""
|
|
exception = InvalidInputError("Invalid input", "input_validator")
|
|
|
|
assert isinstance(exception, ValidationError)
|
|
assert isinstance(exception, InvalidInputError)
|
|
|
|
def test_missing_required_field_error(self):
|
|
"""Test MissingRequiredFieldError with field context."""
|
|
exception = MissingRequiredFieldError(
|
|
message="Field missing",
|
|
field_name="username",
|
|
component="form_validator",
|
|
operation="validate_user_data",
|
|
)
|
|
|
|
assert isinstance(exception, ValidationError)
|
|
assert exception.field_name == "username"
|
|
assert exception.metadata["field_name"] == "username"
|
|
|
|
|
|
class TestNetworkExceptions:
|
|
"""Test network exception hierarchy."""
|
|
|
|
def test_network_error_base(self):
|
|
"""Test NetworkError as base for network exceptions."""
|
|
exception = NetworkError("Network error", "network_client")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, NetworkError)
|
|
|
|
def test_connection_timeout_error(self):
|
|
"""Test ConnectionTimeoutError specialization."""
|
|
exception = ConnectionTimeoutError("Connection timed out", "http_client")
|
|
|
|
assert isinstance(exception, NetworkError)
|
|
assert isinstance(exception, ConnectionTimeoutError)
|
|
|
|
def test_connection_refused_error(self):
|
|
"""Test ConnectionRefusedError specialization."""
|
|
exception = ConnectionRefusedError("Connection refused", "tcp_client")
|
|
|
|
assert isinstance(exception, NetworkError)
|
|
assert isinstance(exception, ConnectionRefusedError)
|
|
|
|
|
|
class TestTimeoutExceptions:
|
|
"""Test timeout exception hierarchy."""
|
|
|
|
def test_timeout_error_base(self):
|
|
"""Test TimeoutError as base for timeout exceptions."""
|
|
exception = TimeoutError("Operation timed out", "processor")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, TimeoutError)
|
|
|
|
def test_operation_timeout_error(self):
|
|
"""Test OperationTimeoutError with timeout duration."""
|
|
exception = OperationTimeoutError(
|
|
message="Operation timeout",
|
|
timeout_duration=30.0,
|
|
component="audio_processor",
|
|
operation="process_audio",
|
|
)
|
|
|
|
assert isinstance(exception, TimeoutError)
|
|
assert exception.timeout_duration == 30.0
|
|
assert exception.metadata["timeout_duration"] == 30.0
|
|
|
|
|
|
class TestConfigurationExceptions:
|
|
"""Test configuration exception hierarchy."""
|
|
|
|
def test_configuration_error_base(self):
|
|
"""Test ConfigurationError as base for config exceptions."""
|
|
exception = ConfigurationError("Config error", "config_manager")
|
|
|
|
assert isinstance(exception, BotBaseException)
|
|
assert isinstance(exception, ConfigurationError)
|
|
|
|
def test_missing_configuration_error(self):
|
|
"""Test MissingConfigurationError specialization."""
|
|
exception = MissingConfigurationError("Config missing", "config_loader")
|
|
|
|
assert isinstance(exception, ConfigurationError)
|
|
assert isinstance(exception, MissingConfigurationError)
|
|
|
|
def test_invalid_configuration_error(self):
|
|
"""Test InvalidConfigurationError specialization."""
|
|
exception = InvalidConfigurationError("Config invalid", "config_validator")
|
|
|
|
assert isinstance(exception, ConfigurationError)
|
|
assert isinstance(exception, InvalidConfigurationError)
|
|
|
|
|
|
class TestExceptionInteroperability:
|
|
"""Test exception interoperability and edge cases."""
|
|
|
|
def test_exception_chaining(self):
|
|
"""Test that exceptions can be chained properly."""
|
|
original_error = ValueError("Original error")
|
|
|
|
chained_error = AudioConversionError(
|
|
"Conversion failed due to invalid input", "audio_converter"
|
|
)
|
|
|
|
# Test that we can chain exceptions
|
|
try:
|
|
raise chained_error from original_error
|
|
except AudioConversionError as e:
|
|
assert e.__cause__ == original_error
|
|
assert isinstance(e, AudioProcessingError)
|
|
assert isinstance(e, BotBaseException)
|
|
|
|
def test_exception_with_complex_metadata(self):
|
|
"""Test exceptions with complex metadata structures."""
|
|
complex_metadata = {
|
|
"nested_dict": {"key1": "value1", "key2": 42},
|
|
"list_data": [1, 2, 3, "string"],
|
|
"none_value": None,
|
|
"boolean": True,
|
|
"float": 3.14159,
|
|
}
|
|
|
|
exception = MetricsCollectionError(
|
|
"Complex metadata test",
|
|
"metrics_collector",
|
|
"test_operation",
|
|
complex_metadata,
|
|
)
|
|
|
|
assert exception.metadata == complex_metadata
|
|
assert exception.metadata["nested_dict"]["key1"] == "value1"
|
|
assert exception.metadata["list_data"][3] == "string"
|
|
assert exception.metadata["none_value"] is None
|
|
|
|
def test_exception_string_representation(self):
|
|
"""Test string representations of exceptions."""
|
|
exception = PromptVariableError(
|
|
"Missing required variables",
|
|
missing_variables=["quote", "speaker"],
|
|
component="prompt_builder",
|
|
)
|
|
|
|
str_repr = str(exception)
|
|
assert "Missing required variables" in str_repr
|
|
|
|
# Test that the base message is preserved
|
|
assert exception.message == "Missing required variables"
|
|
|
|
def test_exception_attribute_access(self):
|
|
"""Test that all exception attributes are accessible."""
|
|
exception = BotPermissionError(
|
|
"Bot permission test",
|
|
required_permissions=["admin", "moderator"],
|
|
guild=discord.Object(id=123),
|
|
)
|
|
|
|
# Test all expected attributes exist
|
|
assert hasattr(exception, "message")
|
|
assert hasattr(exception, "component")
|
|
assert hasattr(exception, "operation")
|
|
assert hasattr(exception, "metadata")
|
|
assert hasattr(exception, "required_permissions")
|
|
assert hasattr(exception, "guild")
|
|
|
|
# Test attribute values
|
|
assert exception.message == "Bot permission test"
|
|
assert exception.required_permissions == ["admin", "moderator"]
|
|
assert exception.guild.id == 123
|
|
|
|
def test_exception_inheritance_hierarchy(self):
|
|
"""Test that inheritance hierarchy is correct for all exceptions."""
|
|
# Test audio exceptions
|
|
assert issubclass(AudioFormatError, AudioProcessingError)
|
|
assert issubclass(AudioProcessingError, BotBaseException)
|
|
assert issubclass(BotBaseException, Exception)
|
|
|
|
# Test permission exceptions
|
|
assert issubclass(InsufficientPermissionsError, PermissionError)
|
|
assert issubclass(PermissionError, BotBaseException)
|
|
|
|
# Test AI provider exceptions
|
|
assert issubclass(OpenAIError, AIProviderError)
|
|
assert issubclass(AIProviderError, ExternalServiceError)
|
|
assert issubclass(ExternalServiceError, BotBaseException)
|
|
|
|
# Test database exceptions
|
|
assert issubclass(DatabaseConnectionError, DatabaseError)
|
|
assert issubclass(DatabaseError, ExternalServiceError)
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestExceptionIntegration:
|
|
"""Integration tests for exceptions with other systems."""
|
|
|
|
def test_exception_with_discord_objects(self):
|
|
"""Test exceptions work properly with Discord.py objects."""
|
|
# This tests integration with Discord.py library objects
|
|
discord.Object(id=123456789)
|
|
mock_user = discord.Object(id=987654321)
|
|
mock_channel = discord.Object(id=555666777)
|
|
|
|
exception = VoiceChannelPermissionError(
|
|
"Voice channel permission denied",
|
|
channel=mock_channel,
|
|
user=mock_user,
|
|
component="voice_permissions",
|
|
operation="check_voice_access",
|
|
)
|
|
|
|
assert exception.channel.id == 555666777
|
|
assert exception.user.id == 987654321
|
|
assert exception.metadata["channel_id"] == 555666777
|
|
assert exception.metadata["user_id"] == 987654321
|
|
|
|
def test_exception_metadata_serialization(self):
|
|
"""Test that exception metadata can be serialized (e.g., for logging)."""
|
|
import json
|
|
|
|
metadata = {
|
|
"string": "test",
|
|
"number": 42,
|
|
"float": 3.14,
|
|
"boolean": True,
|
|
"none": None,
|
|
"list": [1, 2, 3],
|
|
"dict": {"nested": "value"},
|
|
}
|
|
|
|
exception = MetricsError(
|
|
"Serialization test", "metrics_collector", metadata=metadata
|
|
)
|
|
|
|
# Should be able to serialize metadata to JSON
|
|
serialized = json.dumps(exception.metadata)
|
|
deserialized = json.loads(serialized)
|
|
|
|
assert deserialized == metadata
|
|
|
|
def test_exception_in_error_handling_context(self):
|
|
"""Test exceptions work in error handling contexts."""
|
|
|
|
def process_audio():
|
|
raise AudioConversionError(
|
|
"Conversion failed",
|
|
"audio_processor",
|
|
"convert_to_wav",
|
|
{"input_format": "mp3", "output_format": "wav"},
|
|
)
|
|
|
|
def handle_audio_error():
|
|
try:
|
|
process_audio()
|
|
except AudioProcessingError as e:
|
|
# Should be able to handle as base audio error
|
|
assert e.component == "audio_processor"
|
|
assert e.operation == "convert_to_wav"
|
|
assert e.metadata["input_format"] == "mp3"
|
|
raise PrometheusError(
|
|
"Metrics failed due to audio error", "metrics_collector"
|
|
) from e
|
|
|
|
# Test exception chaining works
|
|
with pytest.raises(PrometheusError) as exc_info:
|
|
handle_audio_error()
|
|
|
|
assert isinstance(exc_info.value.__cause__, AudioConversionError)
|
|
|
|
def test_exception_context_preservation(self):
|
|
"""Test that exception context is preserved across function calls."""
|
|
|
|
def inner_function():
|
|
return PromptTemplateError(
|
|
"Template not found",
|
|
"prompt_builder",
|
|
"load_template",
|
|
{
|
|
"template_name": "analysis_prompt",
|
|
"provider": "openai",
|
|
"search_paths": ["/templates", "/custom_templates"],
|
|
},
|
|
)
|
|
|
|
def middle_function():
|
|
error = inner_function()
|
|
# Add additional context
|
|
error.metadata["middleware_info"] = "processing in middle layer"
|
|
return error
|
|
|
|
def outer_function():
|
|
error = middle_function()
|
|
error.metadata["request_id"] = "req_12345"
|
|
raise error
|
|
|
|
with pytest.raises(PromptTemplateError) as exc_info:
|
|
outer_function()
|
|
|
|
exception = exc_info.value
|
|
assert exception.component == "prompt_builder"
|
|
assert exception.operation == "load_template"
|
|
assert exception.metadata["template_name"] == "analysis_prompt"
|
|
assert exception.metadata["middleware_info"] == "processing in middle layer"
|
|
assert exception.metadata["request_id"] == "req_12345"
|
|
|
|
|
|
@pytest.mark.performance
|
|
class TestExceptionPerformance:
|
|
"""Performance tests for exception creation and handling."""
|
|
|
|
def test_exception_creation_performance(self):
|
|
"""Test performance of exception creation."""
|
|
import time
|
|
|
|
start_time = time.time()
|
|
|
|
exceptions = []
|
|
for i in range(1000):
|
|
exception = AudioProcessingError(
|
|
f"Error {i}",
|
|
"audio_processor",
|
|
f"operation_{i}",
|
|
{"iteration": i, "timestamp": start_time + i},
|
|
)
|
|
exceptions.append(exception)
|
|
|
|
end_time = time.time()
|
|
duration = end_time - start_time
|
|
|
|
# Should be able to create 1000 exceptions quickly
|
|
assert duration < 0.5 # Less than 500ms
|
|
assert len(exceptions) == 1000
|
|
|
|
def test_exception_inheritance_check_performance(self):
|
|
"""Test performance of isinstance checks with exception hierarchy."""
|
|
import time
|
|
|
|
# Create various exception instances
|
|
exceptions = [
|
|
AudioFormatError("Format error", "audio"),
|
|
PromptVariableError("Variable error", ["var"], "prompt"),
|
|
MetricsCollectionError("Metrics error", "metrics"),
|
|
DatabaseConnectionError("DB error", "database"),
|
|
ValidationError("Validation error", "validator"),
|
|
]
|
|
|
|
start_time = time.time()
|
|
|
|
# Perform many isinstance checks
|
|
for _ in range(1000):
|
|
for exception in exceptions:
|
|
isinstance(exception, BotBaseException)
|
|
isinstance(exception, AudioProcessingError)
|
|
isinstance(exception, ValidationError)
|
|
isinstance(exception, ExternalServiceError)
|
|
|
|
end_time = time.time()
|
|
duration = end_time - start_time
|
|
|
|
# isinstance checks should be fast
|
|
assert duration < 0.1 # Less than 100ms
|
|
|
|
def test_exception_metadata_access_performance(self):
|
|
"""Test performance of metadata access."""
|
|
import time
|
|
|
|
large_metadata = {}
|
|
for i in range(100):
|
|
large_metadata[f"key_{i}"] = f"value_{i}" * 10
|
|
|
|
exception = MetricsError(
|
|
"Large metadata test", "metrics_collector", metadata=large_metadata
|
|
)
|
|
|
|
start_time = time.time()
|
|
|
|
# Access metadata many times
|
|
for _ in range(1000):
|
|
assert exception.metadata
|
|
assert exception.component
|
|
assert exception.message
|
|
# Access some keys
|
|
if "key_50" in exception.metadata:
|
|
value = exception.metadata["key_50"]
|
|
assert value
|
|
|
|
end_time = time.time()
|
|
duration = end_time - start_time
|
|
|
|
# Metadata access should be fast
|
|
assert duration < 0.1 # Less than 100ms
|