From d0a1673965cc5191b091e7bbedb03aeac5c6e7e4 Mon Sep 17 00:00:00 2001 From: Clemens Schwaighofer Date: Fri, 24 Oct 2025 18:33:25 +0900 Subject: [PATCH] Add pytest for logging --- src/corelibs/logging_handling/log.py | 2 +- ...g_1_settings_parsing_spacers_parameters.py | 186 +++++++ .../log_testing/test_log_2_basic_handling.py | 436 +++++++++++++++ .../test_log_3_custom_console_formatter.py | 141 +++++ .../test_log_4_custom_handler_filter.py | 122 +++++ .../test_log_5_handler_management.py | 108 ++++ .../log_testing/test_log_6_logger.py | 92 ++++ .../log_testing/test_log_7_edge_cases.py | 113 ++++ .../log_testing/test_log_99_queue_listener.py | 140 +++++ .../logging_handling/test_error_handling.py | 503 ++++++++++++++++++ 10 files changed, 1842 insertions(+), 1 deletion(-) create mode 100644 tests/unit/logging_handling/log_testing/test_log_1_settings_parsing_spacers_parameters.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_2_basic_handling.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_3_custom_console_formatter.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_4_custom_handler_filter.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_5_handler_management.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_6_logger.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_7_edge_cases.py create mode 100644 tests/unit/logging_handling/log_testing/test_log_99_queue_listener.py create mode 100644 tests/unit/logging_handling/test_error_handling.py diff --git a/src/corelibs/logging_handling/log.py b/src/corelibs/logging_handling/log.py index 9710ed9..6dd75db 100644 --- a/src/corelibs/logging_handling/log.py +++ b/src/corelibs/logging_handling/log.py @@ -481,7 +481,7 @@ class Log(LogParent): """ Call when class is destroyed, make sure the listender is closed or else we throw a thread error """ - if self.log_settings['add_end_info']: + if hasattr(self, 'log_settings') and self.log_settings.get('add_end_info'): self.break_line('END') self.stop_listener() diff --git a/tests/unit/logging_handling/log_testing/test_log_1_settings_parsing_spacers_parameters.py b/tests/unit/logging_handling/log_testing/test_log_1_settings_parsing_spacers_parameters.py new file mode 100644 index 0000000..b29fdf2 --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_1_settings_parsing_spacers_parameters.py @@ -0,0 +1,186 @@ +""" +Unit tests for log settings parsing and spacer constants in Log class. +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +from pathlib import Path +from typing import Any +import pytest +from corelibs.logging_handling.log import ( + Log, + LogParent, + LogSettings, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test Log Settings Parsing +class TestLogSettingsParsing: + """Test cases for log settings parsing""" + + def test_parse_with_string_log_levels(self, tmp_log_path: Path): + """Test parsing with string log levels""" + settings: dict[str, Any] = { + "log_level_console": "ERROR", + "log_level_file": "INFO", + } + log = Log( + log_path=tmp_log_path, + log_name="test", + log_settings=settings # type: ignore + ) + + assert log.log_settings["log_level_console"] == LoggingLevel.ERROR + assert log.log_settings["log_level_file"] == LoggingLevel.INFO + + def test_parse_with_int_log_levels(self, tmp_log_path: Path): + """Test parsing with integer log levels""" + settings: dict[str, Any] = { + "log_level_console": 40, # ERROR + "log_level_file": 20, # INFO + } + log = Log( + log_path=tmp_log_path, + log_name="test", + log_settings=settings # type: ignore + ) + + assert log.log_settings["log_level_console"] == LoggingLevel.ERROR + assert log.log_settings["log_level_file"] == LoggingLevel.INFO + + def test_parse_with_invalid_bool_settings(self, tmp_log_path: Path): + """Test parsing with invalid bool settings""" + settings: dict[str, Any] = { + "console_enabled": "not_a_bool", + "per_run_log": 123, + } + log = Log( + log_path=tmp_log_path, + log_name="test", + log_settings=settings # type: ignore + ) + + # Should fall back to defaults + assert log.log_settings["console_enabled"] == Log.DEFAULT_LOG_SETTINGS["console_enabled"] + assert log.log_settings["per_run_log"] == Log.DEFAULT_LOG_SETTINGS["per_run_log"] + + +# MARK: Test Spacer Constants +class TestSpacerConstants: + """Test cases for spacer constants""" + + def test_spacer_char_constant(self): + """Test SPACER_CHAR constant""" + assert Log.SPACER_CHAR == '=' + assert LogParent.SPACER_CHAR == '=' + + def test_spacer_length_constant(self): + """Test SPACER_LENGTH constant""" + assert Log.SPACER_LENGTH == 32 + assert LogParent.SPACER_LENGTH == 32 + + +# MARK: Parametrized Tests +class TestParametrized: + """Parametrized tests for comprehensive coverage""" + + @pytest.mark.parametrize("log_level,expected", [ + (LoggingLevel.DEBUG, 10), + (LoggingLevel.INFO, 20), + (LoggingLevel.WARNING, 30), + (LoggingLevel.ERROR, 40), + (LoggingLevel.CRITICAL, 50), + (LoggingLevel.ALERT, 55), + (LoggingLevel.EMERGENCY, 60), + (LoggingLevel.EXCEPTION, 70), + ]) + def test_log_level_values(self, log_level: LoggingLevel, expected: int): + """Test log level values""" + assert log_level.value == expected + + @pytest.mark.parametrize("method_name,level_name", [ + ("debug", "DEBUG"), + ("info", "INFO"), + ("warning", "WARNING"), + ("error", "ERROR"), + ("critical", "CRITICAL"), + ]) + def test_logging_methods_write_correct_level( + self, + log_instance: Log, + tmp_log_path: Path, + method_name: str, + level_name: str + ): + """Test each logging method writes correct level""" + method = getattr(log_instance, method_name) + method(f"Test {level_name} message") + + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert level_name in content + assert f"Test {level_name} message" in content + + @pytest.mark.parametrize("setting_key,valid_value,invalid_value", [ + ("per_run_log", True, "not_bool"), + ("console_enabled", False, 123), + ("console_color_output_enabled", True, None), + ("add_start_info", False, []), + ("add_end_info", True, {}), + ]) + def test_bool_setting_validation( + self, + tmp_log_path: Path, + setting_key: str, + valid_value: bool, + invalid_value: Any + ): + """Test bool setting validation and fallback""" + # Test with valid value + settings_valid: dict[str, Any] = {setting_key: valid_value} + log_valid = Log(tmp_log_path, "test_valid", settings_valid) # type: ignore + assert log_valid.log_settings[setting_key] == valid_value + + # Test with invalid value (should fall back to default) + settings_invalid: dict[str, Any] = {setting_key: invalid_value} + log_invalid = Log(tmp_log_path, "test_invalid", settings_invalid) # type: ignore + assert log_invalid.log_settings[setting_key] == Log.DEFAULT_LOG_SETTINGS.get( + setting_key, True + ) + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_2_basic_handling.py b/tests/unit/logging_handling/log_testing/test_log_2_basic_handling.py new file mode 100644 index 0000000..7d12168 --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_2_basic_handling.py @@ -0,0 +1,436 @@ +""" +Unit tests for basic Log handling functionality. +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +import logging +from pathlib import Path +from typing import Any +import pytest +from corelibs.logging_handling.log import ( + Log, + LogParent, + LogSettings, + CustomConsoleFormatter, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test LogParent +class TestLogParent: + """Test cases for LogParent class""" + + def test_validate_log_level_valid(self): + """Test validate_log_level with valid levels""" + assert LogParent.validate_log_level(LoggingLevel.DEBUG) is True + assert LogParent.validate_log_level(10) is True + assert LogParent.validate_log_level("INFO") is True + assert LogParent.validate_log_level("warning") is True + + def test_validate_log_level_invalid(self): + """Test validate_log_level with invalid levels""" + assert LogParent.validate_log_level("INVALID") is False + assert LogParent.validate_log_level(999) is False + + def test_get_log_level_int_valid(self): + """Test get_log_level_int with valid levels""" + assert LogParent.get_log_level_int(LoggingLevel.DEBUG) == 10 + assert LogParent.get_log_level_int(20) == 20 + assert LogParent.get_log_level_int("ERROR") == 40 + + def test_get_log_level_int_invalid(self): + """Test get_log_level_int with invalid level returns default""" + result = LogParent.get_log_level_int("INVALID") + assert result == LoggingLevel.WARNING.value + + def test_debug_without_logger_raises(self): + """Test debug method raises when logger not initialized""" + parent = LogParent() + with pytest.raises(ValueError, match="Logger is not yet initialized"): + parent.debug("Test message") + + def test_info_without_logger_raises(self): + """Test info method raises when logger not initialized""" + parent = LogParent() + with pytest.raises(ValueError, match="Logger is not yet initialized"): + parent.info("Test message") + + def test_warning_without_logger_raises(self): + """Test warning method raises when logger not initialized""" + parent = LogParent() + with pytest.raises(ValueError, match="Logger is not yet initialized"): + parent.warning("Test message") + + def test_error_without_logger_raises(self): + """Test error method raises when logger not initialized""" + parent = LogParent() + with pytest.raises(ValueError, match="Logger is not yet initialized"): + parent.error("Test message") + + def test_critical_without_logger_raises(self): + """Test critical method raises when logger not initialized""" + parent = LogParent() + with pytest.raises(ValueError, match="Logger is not yet initialized"): + parent.critical("Test message") + + def test_flush_without_queue_returns_false(self, log_instance: Log): + """Test flush returns False when no queue""" + result = log_instance.flush() + assert result is False + + def test_cleanup_without_queue(self, log_instance: Log): + """Test cleanup does nothing when no queue""" + log_instance.cleanup() # Should not raise + + +# MARK: Test Log Initialization +class TestLogInitialization: + """Test cases for Log class initialization""" + + def test_init_basic(self, tmp_log_path: Path, basic_log_settings: LogSettings): + """Test basic Log initialization""" + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + assert log.log_name == "test_log" + assert log.logger is not None + assert isinstance(log.logger, logging.Logger) + assert "file_handler" in log.handlers + assert "stream_handler" in log.handlers + + def test_init_with_log_extension(self, tmp_log_path: Path, basic_log_settings: LogSettings): + """Test initialization with .log extension in name""" + log = Log( + log_path=tmp_log_path, + log_name="test_log.log", + log_settings=basic_log_settings + ) + + # When log_name ends with .log, the code strips it but the logic keeps it + # Based on code: if not log_name.endswith('.log'): log_name = Path(log_name).stem + # So if it DOES end with .log, it keeps the original name + assert log.log_name == "test_log.log" + + def test_init_with_file_path(self, tmp_log_path: Path, basic_log_settings: LogSettings): + """Test initialization with file path instead of directory""" + log_file = tmp_log_path / "custom.log" + log = Log( + log_path=log_file, + log_name="test", + log_settings=basic_log_settings + ) + + assert log.logger is not None + assert log.log_name == "test" + + def test_init_console_disabled(self, tmp_log_path: Path): + """Test initialization with console disabled""" + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": False, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=settings + ) + + assert "stream_handler" not in log.handlers + assert "file_handler" in log.handlers + + def test_init_per_run_log(self, tmp_log_path: Path): + """Test initialization with per_run_log enabled""" + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": True, + "console_enabled": False, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=settings + ) + + assert log.logger is not None + # Check that a timestamped log file was created + # Files are created in parent directory with sanitized name + log_files = list(tmp_log_path.glob("testlog.*.log")) + assert len(log_files) > 0 + + def test_init_with_none_settings(self, tmp_log_path: Path): + """Test initialization with None settings uses defaults""" + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=None + ) + + assert log.log_settings == Log.DEFAULT_LOG_SETTINGS + assert log.logger is not None + + def test_init_with_partial_settings(self, tmp_log_path: Path): + """Test initialization with partial settings""" + settings: dict[str, Any] = { + "log_level_console": LoggingLevel.ERROR, + "console_enabled": True, + } + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=settings # type: ignore + ) + + assert log.log_settings["log_level_console"] == LoggingLevel.ERROR + # Other settings should use defaults + assert log.log_settings["log_level_file"] == Log.DEFAULT_LOG_LEVEL_FILE + + def test_init_with_invalid_log_level(self, tmp_log_path: Path): + """Test initialization with invalid log level falls back to default""" + settings: dict[str, Any] = { + "log_level_console": "INVALID_LEVEL", + } + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=settings # type: ignore + ) + + # Invalid log levels are reset to the default for that specific entry + # Since INVALID_LEVEL fails validation, it uses DEFAULT_LOG_SETTINGS value + assert log.log_settings["log_level_console"] == Log.DEFAULT_LOG_SETTINGS["log_level_console"] + + def test_init_with_color_output(self, tmp_log_path: Path): + """Test initialization with color output enabled""" + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": True, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=settings + ) + + console_handler = log.handlers["stream_handler"] + assert isinstance(console_handler.formatter, CustomConsoleFormatter) + + def test_init_with_other_handlers(self, tmp_log_path: Path, basic_log_settings: LogSettings): + """Test initialization with additional custom handlers""" + custom_handler = logging.StreamHandler() + custom_handler.set_name("custom_handler") + + log = Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings, + other_handlers={"custom": custom_handler} + ) + + assert "custom" in log.handlers + assert log.handlers["custom"] == custom_handler + + +# MARK: Test Log Methods +class TestLogMethods: + """Test cases for Log logging methods""" + + def test_debug_logging(self, log_instance: Log, tmp_log_path: Path): + """Test debug level logging""" + log_instance.debug("Debug message") + # Verify log file contains the message + # Log file is created with sanitized name (testlog.log) + log_file = tmp_log_path / "testlog.log" + assert log_file.exists() + content = log_file.read_text() + assert "Debug message" in content + assert "DEBUG" in content + + def test_info_logging(self, log_instance: Log, tmp_log_path: Path): + """Test info level logging""" + log_instance.info("Info message") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Info message" in content + assert "INFO" in content + + def test_warning_logging(self, log_instance: Log, tmp_log_path: Path): + """Test warning level logging""" + log_instance.warning("Warning message") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Warning message" in content + assert "WARNING" in content + + def test_error_logging(self, log_instance: Log, tmp_log_path: Path): + """Test error level logging""" + log_instance.error("Error message") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Error message" in content + assert "ERROR" in content + + def test_critical_logging(self, log_instance: Log, tmp_log_path: Path): + """Test critical level logging""" + log_instance.critical("Critical message") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Critical message" in content + assert "CRITICAL" in content + + def test_alert_logging(self, log_instance: Log, tmp_log_path: Path): + """Test alert level logging""" + log_instance.alert("Alert message") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Alert message" in content + assert "ALERT" in content + + def test_emergency_logging(self, log_instance: Log, tmp_log_path: Path): + """Test emergency level logging""" + log_instance.emergency("Emergency message") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Emergency message" in content + assert "EMERGENCY" in content + + def test_exception_logging(self, log_instance: Log, tmp_log_path: Path): + """Test exception level logging""" + try: + raise ValueError("Test exception") + except ValueError: + log_instance.exception("Exception occurred") + + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Exception occurred" in content + assert "EXCEPTION" in content + assert "ValueError" in content + + def test_exception_logging_without_error(self, log_instance: Log, tmp_log_path: Path): + """Test exception logging with log_error=False""" + try: + raise ValueError("Test exception") + except ValueError: + log_instance.exception("Exception occurred", log_error=False) + + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Exception occurred" in content + # Should not have the ERROR level entry + assert "<=EXCEPTION=" not in content + + def test_log_with_extra(self, log_instance: Log, tmp_log_path: Path): + """Test logging with extra parameters""" + extra: dict[str, object] = {"custom_field": "custom_value"} + log_instance.info("Info with extra", extra=extra) + + log_file = tmp_log_path / "testlog.log" + assert log_file.exists() + content = log_file.read_text() + assert "Info with extra" in content + + def test_break_line(self, log_instance: Log, tmp_log_path: Path): + """Test break_line method""" + log_instance.break_line("TEST") + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "[TEST]" in content + assert "=" in content + + def test_break_line_default(self, log_instance: Log, tmp_log_path: Path): + """Test break_line with default parameter""" + log_instance.break_line() + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "[BREAK]" in content + + +# MARK: Test Log Level Handling +class TestLogLevelHandling: + """Test cases for log level handling""" + + def test_set_log_level_file_handler(self, log_instance: Log): + """Test setting log level for file handler""" + result = log_instance.set_log_level("file_handler", LoggingLevel.ERROR) + assert result is True + assert log_instance.get_log_level("file_handler") == LoggingLevel.ERROR + + def test_set_log_level_console_handler(self, log_instance: Log): + """Test setting log level for console handler""" + result = log_instance.set_log_level("stream_handler", LoggingLevel.CRITICAL) + assert result is True + assert log_instance.get_log_level("stream_handler") == LoggingLevel.CRITICAL + + def test_set_log_level_invalid_handler(self, log_instance: Log): + """Test setting log level for non-existent handler raises KeyError""" + # The actual implementation uses dict access which raises KeyError, not IndexError + with pytest.raises(KeyError): + log_instance.set_log_level("nonexistent", LoggingLevel.DEBUG) + + def test_get_log_level_invalid_handler(self, log_instance: Log): + """Test getting log level for non-existent handler raises KeyError""" + # The actual implementation uses dict access which raises KeyError, not IndexError + with pytest.raises(KeyError): + log_instance.get_log_level("nonexistent") + + def test_get_log_level(self, log_instance: Log): + """Test getting current log level""" + level = log_instance.get_log_level("file_handler") + assert level == LoggingLevel.DEBUG + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_3_custom_console_formatter.py b/tests/unit/logging_handling/log_testing/test_log_3_custom_console_formatter.py new file mode 100644 index 0000000..bc6dd00 --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_3_custom_console_formatter.py @@ -0,0 +1,141 @@ +""" +Unit tests for CustomConsoleFormatter in logging handling +""" + +# pylint: disable=protected-access,redefined-outer-name + +import logging +from pathlib import Path +import pytest +from corelibs.logging_handling.log import ( + Log, + LogSettings, + CustomConsoleFormatter, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test CustomConsoleFormatter +class TestCustomConsoleFormatter: + """Test cases for CustomConsoleFormatter""" + + def test_format_debug_level(self): + """Test formatting DEBUG level message""" + formatter = CustomConsoleFormatter('[%(levelname)s] %(message)s') + record = logging.LogRecord( + name="test", + level=logging.DEBUG, + pathname="test.py", + lineno=1, + msg="Debug message", + args=(), + exc_info=None + ) + + result = formatter.format(record) + assert "Debug message" in result + assert "DEBUG" in result + + def test_format_info_level(self): + """Test formatting INFO level message""" + formatter = CustomConsoleFormatter('[%(levelname)s] %(message)s') + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="test.py", + lineno=1, + msg="Info message", + args=(), + exc_info=None + ) + + result = formatter.format(record) + assert "Info message" in result + assert "INFO" in result + + def test_format_warning_level(self): + """Test formatting WARNING level message""" + formatter = CustomConsoleFormatter('[%(levelname)s] %(message)s') + record = logging.LogRecord( + name="test", + level=logging.WARNING, + pathname="test.py", + lineno=1, + msg="Warning message", + args=(), + exc_info=None + ) + + result = formatter.format(record) + assert "Warning message" in result + assert "WARNING" in result + + def test_format_error_level(self): + """Test formatting ERROR level message""" + formatter = CustomConsoleFormatter('[%(levelname)s] %(message)s') + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname="test.py", + lineno=1, + msg="Error message", + args=(), + exc_info=None + ) + + result = formatter.format(record) + assert "Error message" in result + assert "ERROR" in result + + def test_format_critical_level(self): + """Test formatting CRITICAL level message""" + formatter = CustomConsoleFormatter('[%(levelname)s] %(message)s') + record = logging.LogRecord( + name="test", + level=logging.CRITICAL, + pathname="test.py", + lineno=1, + msg="Critical message", + args=(), + exc_info=None + ) + + result = formatter.format(record) + assert "Critical message" in result + assert "CRITICAL" in result + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_4_custom_handler_filter.py b/tests/unit/logging_handling/log_testing/test_log_4_custom_handler_filter.py new file mode 100644 index 0000000..49b9506 --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_4_custom_handler_filter.py @@ -0,0 +1,122 @@ +""" +Unit tests for CustomHandlerFilter in logging handling +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +import logging +from pathlib import Path +import pytest +from corelibs.logging_handling.log import ( + Log, + LogSettings, + CustomHandlerFilter, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test CustomHandlerFilter +class TestCustomHandlerFilter: + """Test cases for CustomHandlerFilter""" + + def test_filter_exceptions_for_console(self): + """Test filtering exception records for console handler""" + handler_filter = CustomHandlerFilter('console', filter_exceptions=True) + record = logging.LogRecord( + name="test", + level=70, # EXCEPTION level + pathname="test.py", + lineno=1, + msg="Exception message", + args=(), + exc_info=None + ) + record.levelname = "EXCEPTION" + + result = handler_filter.filter(record) + assert result is False + + def test_filter_non_exceptions_for_console(self): + """Test non-exception records pass through console filter""" + handler_filter = CustomHandlerFilter('console', filter_exceptions=True) + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname="test.py", + lineno=1, + msg="Error message", + args=(), + exc_info=None + ) + + result = handler_filter.filter(record) + assert result is True + + def test_filter_console_flag_for_file(self): + """Test filtering console-flagged records for file handler""" + handler_filter = CustomHandlerFilter('file', filter_exceptions=False) + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname="test.py", + lineno=1, + msg="Error message", + args=(), + exc_info=None + ) + record.console = True + + result = handler_filter.filter(record) + assert result is False + + def test_filter_normal_record_for_file(self): + """Test normal records pass through file filter""" + handler_filter = CustomHandlerFilter('file', filter_exceptions=False) + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="test.py", + lineno=1, + msg="Info message", + args=(), + exc_info=None + ) + + result = handler_filter.filter(record) + assert result is True + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_5_handler_management.py b/tests/unit/logging_handling/log_testing/test_log_5_handler_management.py new file mode 100644 index 0000000..cc0491f --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_5_handler_management.py @@ -0,0 +1,108 @@ +""" +Unit tests for Log handler management +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +import logging +from pathlib import Path +import pytest +from corelibs.logging_handling.log import ( + Log, + LogParent, + LogSettings, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test Handler Management +class TestHandlerManagement: + """Test cases for handler management""" + + def test_add_handler_before_init(self, tmp_log_path: Path): + """Test adding handler before logger initialization""" + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": False, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + custom_handler = logging.StreamHandler() + custom_handler.set_name("custom") + + log = Log( + log_path=tmp_log_path, + log_name="test", + log_settings=settings, + other_handlers={"custom": custom_handler} + ) + + assert "custom" in log.handlers + + def test_add_handler_after_init_raises(self, log_instance: Log): + """Test adding handler after initialization raises error""" + custom_handler = logging.StreamHandler() + custom_handler.set_name("custom2") + + with pytest.raises(ValueError, match="Cannot add handler"): + log_instance.add_handler("custom2", custom_handler) + + def test_add_duplicate_handler_returns_false(self): + """Test adding duplicate handler returns False""" + # Create a Log instance in a way we can test before initialization + log = object.__new__(Log) + LogParent.__init__(log) + log.handlers = {} + log.listener = None + + handler1 = logging.StreamHandler() + handler1.set_name("test") + handler2 = logging.StreamHandler() + handler2.set_name("test") + + result1 = log.add_handler("test", handler1) + assert result1 is True + + result2 = log.add_handler("test", handler2) + assert result2 is False + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_6_logger.py b/tests/unit/logging_handling/log_testing/test_log_6_logger.py new file mode 100644 index 0000000..9a47d5b --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_6_logger.py @@ -0,0 +1,92 @@ +""" +Unit tests for Log, Logger, and LogParent classes +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +from pathlib import Path +import pytest +from corelibs.logging_handling.log import ( + Log, + Logger, + LogSettings, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test Logger Class +class TestLogger: + """Test cases for Logger class""" + + def test_logger_init(self, log_instance: Log): + """Test Logger initialization""" + logger_settings = log_instance.get_logger_settings() + logger = Logger(logger_settings) + + assert logger.logger is not None + assert logger.lg == logger.logger + assert logger.l == logger.logger + assert isinstance(logger.handlers, dict) + assert len(logger.handlers) > 0 + + def test_logger_logging_methods(self, log_instance: Log, tmp_log_path: Path): + """Test Logger logging methods""" + logger_settings = log_instance.get_logger_settings() + logger = Logger(logger_settings) + + logger.debug("Debug from Logger") + logger.info("Info from Logger") + logger.warning("Warning from Logger") + logger.error("Error from Logger") + logger.critical("Critical from Logger") + + log_file = tmp_log_path / "testlog.log" + content = log_file.read_text() + assert "Debug from Logger" in content + assert "Info from Logger" in content + assert "Warning from Logger" in content + assert "Error from Logger" in content + assert "Critical from Logger" in content + + def test_logger_shared_queue(self, log_instance: Log): + """Test Logger shares the same log queue""" + logger_settings = log_instance.get_logger_settings() + logger = Logger(logger_settings) + + assert logger.log_queue == log_instance.log_queue + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_7_edge_cases.py b/tests/unit/logging_handling/log_testing/test_log_7_edge_cases.py new file mode 100644 index 0000000..64eefe4 --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_7_edge_cases.py @@ -0,0 +1,113 @@ +""" +Unit tests for Log, Logger, and LogParent classes +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +import logging +from pathlib import Path +import pytest +from corelibs.logging_handling.log import ( + Log, + LogSettings, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test Edge Cases +class TestEdgeCases: + """Test edge cases and special scenarios""" + + def test_log_name_sanitization(self, tmp_log_path: Path, basic_log_settings: LogSettings): + """Test log name with special characters gets sanitized""" + _ = Log( + log_path=tmp_log_path, + log_name="test@#$%log", + log_settings=basic_log_settings + ) + + # Special characters should be removed from filename + log_file = tmp_log_path / "testlog.log" + assert log_file.exists() or any(tmp_log_path.glob("test*.log")) + + def test_multiple_log_instances(self, tmp_log_path: Path, basic_log_settings: LogSettings): + """Test creating multiple Log instances""" + log1 = Log(tmp_log_path, "log1", basic_log_settings) + log2 = Log(tmp_log_path, "log2", basic_log_settings) + + log1.info("From log1") + log2.info("From log2") + + log_file1 = tmp_log_path / "log1.log" + log_file2 = tmp_log_path / "log2.log" + + assert log_file1.exists() + assert log_file2.exists() + assert "From log1" in log_file1.read_text() + assert "From log2" in log_file2.read_text() + + def test_destructor_calls_stop_listener(self, tmp_log_path: Path): + """Test destructor calls stop_listener""" + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": False, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": True, # Enable end info + "log_queue": None, + } + + log = Log(tmp_log_path, "test", settings) + del log + + # Check that the log file was finalized + log_file = tmp_log_path / "test.log" + if log_file.exists(): + content = log_file.read_text() + assert "[END]" in content + + def test_get_logger_settings(self, log_instance: Log): + """Test get_logger_settings returns correct structure""" + settings = log_instance.get_logger_settings() + + assert "logger" in settings + assert "log_queue" in settings + assert isinstance(settings["logger"], logging.Logger) + +# __END__ diff --git a/tests/unit/logging_handling/log_testing/test_log_99_queue_listener.py b/tests/unit/logging_handling/log_testing/test_log_99_queue_listener.py new file mode 100644 index 0000000..95c75e7 --- /dev/null +++ b/tests/unit/logging_handling/log_testing/test_log_99_queue_listener.py @@ -0,0 +1,140 @@ +""" +Unit tests for Log, Logger, and LogParent classes +""" + +# pylint: disable=protected-access,redefined-outer-name,use-implicit-booleaness-not-comparison + +import logging +from pathlib import Path +from unittest.mock import Mock, MagicMock, patch +from multiprocessing import Queue +import pytest +from corelibs.logging_handling.log import ( + Log, + LogSettings, +) +from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel + + +# MARK: Fixtures +@pytest.fixture +def tmp_log_path(tmp_path: Path) -> Path: + """Create a temporary directory for log files""" + log_dir = tmp_path / "logs" + log_dir.mkdir(exist_ok=True) + return log_dir + + +@pytest.fixture +def basic_log_settings() -> LogSettings: + """Basic log settings for testing""" + return { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": True, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": None, + } + + +@pytest.fixture +def log_instance(tmp_log_path: Path, basic_log_settings: LogSettings) -> Log: + """Create a basic Log instance""" + return Log( + log_path=tmp_log_path, + log_name="test_log", + log_settings=basic_log_settings + ) + + +# MARK: Test Queue Listener +class TestQueueListener: + """Test cases for queue listener functionality""" + + @patch('logging.handlers.QueueListener') + def test_init_listener(self, mock_listener_class: MagicMock, tmp_log_path: Path): + """Test listener initialization with queue""" + # Create a mock queue without spec to allow attribute setting + mock_queue = MagicMock() + mock_queue.empty.return_value = True + # Configure queue attributes to prevent TypeError in comparisons + mock_queue._maxsize = -1 # Standard Queue default + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": False, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": mock_queue, # type: ignore + } + + log = Log( + log_path=tmp_log_path, + log_name="test", + log_settings=settings + ) + + assert log.log_queue == mock_queue + mock_listener_class.assert_called_once() + + def test_stop_listener_no_listener(self, log_instance: Log): + """Test stop_listener when no listener exists""" + log_instance.stop_listener() # Should not raise + + @patch('logging.handlers.QueueListener') + def test_stop_listener_with_listener(self, mock_listener_class: MagicMock, tmp_log_path: Path): + """Test stop_listener with active listener""" + # Create a mock queue without spec to allow attribute setting + mock_queue = MagicMock() + mock_queue.empty.return_value = True + # Configure queue attributes to prevent TypeError in comparisons + mock_queue._maxsize = -1 # Standard Queue default + mock_listener = MagicMock() + mock_listener_class.return_value = mock_listener + + settings: LogSettings = { + "log_level_console": LoggingLevel.WARNING, + "log_level_file": LoggingLevel.DEBUG, + "per_run_log": False, + "console_enabled": False, + "console_color_output_enabled": False, + "add_start_info": False, + "add_end_info": False, + "log_queue": mock_queue, # type: ignore + } + + log = Log( + log_path=tmp_log_path, + log_name="test", + log_settings=settings + ) + + log.stop_listener() + mock_listener.stop.assert_called_once() + + +# MARK: Test Static Methods +class TestStaticMethods: + """Test cases for static methods""" + + @patch('logging.getLogger') + def test_init_worker_logging(self, mock_get_logger: MagicMock): + """Test init_worker_logging static method""" + mock_queue = Mock(spec=Queue) + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + + result = Log.init_worker_logging(mock_queue) + + assert result == mock_logger + mock_get_logger.assert_called_once_with() + mock_logger.setLevel.assert_called_once_with(logging.DEBUG) + mock_logger.handlers.clear.assert_called_once() + assert mock_logger.addHandler.called + +# __END__ diff --git a/tests/unit/logging_handling/test_error_handling.py b/tests/unit/logging_handling/test_error_handling.py new file mode 100644 index 0000000..e433a65 --- /dev/null +++ b/tests/unit/logging_handling/test_error_handling.py @@ -0,0 +1,503 @@ +""" +Test cases for ErrorMessage class +""" + +# pylint: disable=use-implicit-booleaness-not-comparison + +from typing import Any +import pytest +from corelibs.logging_handling.error_handling import ErrorMessage + + +class TestErrorMessageWarnings: + """Test cases for warning-related methods""" + + def test_add_warning_basic(self): + """Test adding a basic warning message""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + message = {"code": "W001", "description": "Test warning"} + error_msg.add_warning(message) + + warnings = error_msg.get_warnings() + assert len(warnings) == 1 + assert warnings[0]["code"] == "W001" + assert warnings[0]["description"] == "Test warning" + assert warnings[0]["level"] == "Warning" + + def test_add_warning_with_base_message(self): + """Test adding a warning with base message""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + base_message = {"timestamp": "2025-10-24", "module": "test"} + message = {"code": "W002", "description": "Another warning"} + error_msg.add_warning(message, base_message) + + warnings = error_msg.get_warnings() + assert len(warnings) == 1 + assert warnings[0]["timestamp"] == "2025-10-24" + assert warnings[0]["module"] == "test" + assert warnings[0]["code"] == "W002" + assert warnings[0]["description"] == "Another warning" + assert warnings[0]["level"] == "Warning" + + def test_add_warning_with_none_base_message(self): + """Test adding a warning with None as base message""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + message = {"code": "W003", "description": "Warning with None base"} + error_msg.add_warning(message, None) + + warnings = error_msg.get_warnings() + assert len(warnings) == 1 + assert warnings[0]["code"] == "W003" + assert warnings[0]["level"] == "Warning" + + def test_add_warning_with_invalid_base_message(self): + """Test adding a warning with invalid base message (not a dict)""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + message = {"code": "W004", "description": "Warning with invalid base"} + error_msg.add_warning(message, "invalid_base") # type: ignore + + warnings = error_msg.get_warnings() + assert len(warnings) == 1 + assert warnings[0]["code"] == "W004" + assert warnings[0]["level"] == "Warning" + + def test_add_multiple_warnings(self): + """Test adding multiple warnings""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + error_msg.add_warning({"code": "W001", "description": "First warning"}) + error_msg.add_warning({"code": "W002", "description": "Second warning"}) + error_msg.add_warning({"code": "W003", "description": "Third warning"}) + + warnings = error_msg.get_warnings() + assert len(warnings) == 3 + assert warnings[0]["code"] == "W001" + assert warnings[1]["code"] == "W002" + assert warnings[2]["code"] == "W003" + + def test_get_warnings_empty(self): + """Test getting warnings when list is empty""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + warnings = error_msg.get_warnings() + assert warnings == [] + assert len(warnings) == 0 + + def test_has_warnings_true(self): + """Test has_warnings returns True when warnings exist""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + error_msg.add_warning({"code": "W001", "description": "Test warning"}) + assert error_msg.has_warnings() is True + + def test_has_warnings_false(self): + """Test has_warnings returns False when no warnings exist""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + assert error_msg.has_warnings() is False + + def test_reset_warnings(self): + """Test resetting warnings list""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + error_msg.add_warning({"code": "W001", "description": "Test warning"}) + assert error_msg.has_warnings() is True + + error_msg.reset_warnings() + assert error_msg.has_warnings() is False + assert len(error_msg.get_warnings()) == 0 + + def test_warning_level_override(self): + """Test that level is always set to Warning even if base contains different level""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + base_message = {"level": "Error"} # Should be overridden + message = {"code": "W001", "description": "Test warning"} + error_msg.add_warning(message, base_message) + + warnings = error_msg.get_warnings() + assert warnings[0]["level"] == "Warning" + + +class TestErrorMessageErrors: + """Test cases for error-related methods""" + + def test_add_error_basic(self): + """Test adding a basic error message""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + message = {"code": "E001", "description": "Test error"} + error_msg.add_error(message) + + errors = error_msg.get_errors() + assert len(errors) == 1 + assert errors[0]["code"] == "E001" + assert errors[0]["description"] == "Test error" + assert errors[0]["level"] == "Error" + + def test_add_error_with_base_message(self): + """Test adding an error with base message""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + base_message = {"timestamp": "2025-10-24", "module": "test"} + message = {"code": "E002", "description": "Another error"} + error_msg.add_error(message, base_message) + + errors = error_msg.get_errors() + assert len(errors) == 1 + assert errors[0]["timestamp"] == "2025-10-24" + assert errors[0]["module"] == "test" + assert errors[0]["code"] == "E002" + assert errors[0]["description"] == "Another error" + assert errors[0]["level"] == "Error" + + def test_add_error_with_none_base_message(self): + """Test adding an error with None as base message""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + message = {"code": "E003", "description": "Error with None base"} + error_msg.add_error(message, None) + + errors = error_msg.get_errors() + assert len(errors) == 1 + assert errors[0]["code"] == "E003" + assert errors[0]["level"] == "Error" + + def test_add_error_with_invalid_base_message(self): + """Test adding an error with invalid base message (not a dict)""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + message = {"code": "E004", "description": "Error with invalid base"} + error_msg.add_error(message, "invalid_base") # type: ignore + + errors = error_msg.get_errors() + assert len(errors) == 1 + assert errors[0]["code"] == "E004" + assert errors[0]["level"] == "Error" + + def test_add_multiple_errors(self): + """Test adding multiple errors""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + error_msg.add_error({"code": "E001", "description": "First error"}) + error_msg.add_error({"code": "E002", "description": "Second error"}) + error_msg.add_error({"code": "E003", "description": "Third error"}) + + errors = error_msg.get_errors() + assert len(errors) == 3 + assert errors[0]["code"] == "E001" + assert errors[1]["code"] == "E002" + assert errors[2]["code"] == "E003" + + def test_get_errors_empty(self): + """Test getting errors when list is empty""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + errors = error_msg.get_errors() + assert errors == [] + assert len(errors) == 0 + + def test_has_errors_true(self): + """Test has_errors returns True when errors exist""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + error_msg.add_error({"code": "E001", "description": "Test error"}) + assert error_msg.has_errors() is True + + def test_has_errors_false(self): + """Test has_errors returns False when no errors exist""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + assert error_msg.has_errors() is False + + def test_reset_errors(self): + """Test resetting errors list""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + error_msg.add_error({"code": "E001", "description": "Test error"}) + assert error_msg.has_errors() is True + + error_msg.reset_errors() + assert error_msg.has_errors() is False + assert len(error_msg.get_errors()) == 0 + + def test_error_level_override(self): + """Test that level is always set to Error even if base contains different level""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + base_message = {"level": "Warning"} # Should be overridden + message = {"code": "E001", "description": "Test error"} + error_msg.add_error(message, base_message) + + errors = error_msg.get_errors() + assert errors[0]["level"] == "Error" + + +class TestErrorMessageMixed: + """Test cases for mixed warning and error operations""" + + def test_errors_and_warnings_independent(self): + """Test that errors and warnings are stored independently""" + error_msg = ErrorMessage() + error_msg.reset_errors() + error_msg.reset_warnings() + + error_msg.add_error({"code": "E001", "description": "Test error"}) + error_msg.add_warning({"code": "W001", "description": "Test warning"}) + + assert len(error_msg.get_errors()) == 1 + assert len(error_msg.get_warnings()) == 1 + assert error_msg.has_errors() is True + assert error_msg.has_warnings() is True + + def test_reset_errors_does_not_affect_warnings(self): + """Test that resetting errors does not affect warnings""" + error_msg = ErrorMessage() + error_msg.reset_errors() + error_msg.reset_warnings() + + error_msg.add_error({"code": "E001", "description": "Test error"}) + error_msg.add_warning({"code": "W001", "description": "Test warning"}) + + error_msg.reset_errors() + + assert error_msg.has_errors() is False + assert error_msg.has_warnings() is True + assert len(error_msg.get_warnings()) == 1 + + def test_reset_warnings_does_not_affect_errors(self): + """Test that resetting warnings does not affect errors""" + error_msg = ErrorMessage() + error_msg.reset_errors() + error_msg.reset_warnings() + + error_msg.add_error({"code": "E001", "description": "Test error"}) + error_msg.add_warning({"code": "W001", "description": "Test warning"}) + + error_msg.reset_warnings() + + assert error_msg.has_errors() is True + assert error_msg.has_warnings() is False + assert len(error_msg.get_errors()) == 1 + + +class TestErrorMessageClassVariables: + """Test cases to verify class-level variable behavior""" + + def test_class_variable_shared_across_instances(self): + """Test that error and warning lists are shared across instances""" + error_msg1 = ErrorMessage() + error_msg2 = ErrorMessage() + + error_msg1.reset_errors() + error_msg1.reset_warnings() + + error_msg1.add_error({"code": "E001", "description": "Error from instance 1"}) + error_msg1.add_warning({"code": "W001", "description": "Warning from instance 1"}) + + # Both instances should see the same data + assert len(error_msg2.get_errors()) == 1 + assert len(error_msg2.get_warnings()) == 1 + assert error_msg2.has_errors() is True + assert error_msg2.has_warnings() is True + + def test_reset_affects_all_instances(self): + """Test that reset operations affect all instances""" + error_msg1 = ErrorMessage() + error_msg2 = ErrorMessage() + + error_msg1.reset_errors() + error_msg1.reset_warnings() + + error_msg1.add_error({"code": "E001", "description": "Test error"}) + error_msg1.add_warning({"code": "W001", "description": "Test warning"}) + + error_msg2.reset_errors() + + # Both instances should reflect the reset + assert error_msg1.has_errors() is False + assert error_msg2.has_errors() is False + + error_msg2.reset_warnings() + + assert error_msg1.has_warnings() is False + assert error_msg2.has_warnings() is False + + +class TestErrorMessageEdgeCases: + """Test edge cases and special scenarios""" + + def test_empty_message_dict(self): + """Test adding empty message dictionaries""" + error_msg = ErrorMessage() + error_msg.reset_errors() + error_msg.reset_warnings() + + error_msg.add_error({}) + error_msg.add_warning({}) + + errors = error_msg.get_errors() + warnings = error_msg.get_warnings() + + assert len(errors) == 1 + assert len(warnings) == 1 + assert errors[0] == {"level": "Error"} + assert warnings[0] == {"level": "Warning"} + + def test_message_with_complex_data(self): + """Test adding messages with complex data structures""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + complex_message = { + "code": "E001", + "description": "Complex error", + "details": { + "nested": "data", + "list": [1, 2, 3], + }, + "count": 42, + } + error_msg.add_error(complex_message) + + errors = error_msg.get_errors() + assert errors[0]["code"] == "E001" + assert errors[0]["details"]["nested"] == "data" + assert errors[0]["details"]["list"] == [1, 2, 3] + assert errors[0]["count"] == 42 + assert errors[0]["level"] == "Error" + + def test_base_message_merge_override(self): + """Test that message values override base_message values""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + base_message = {"code": "BASE", "description": "Base description", "timestamp": "2025-10-24"} + message = {"code": "E001", "description": "Override description"} + error_msg.add_error(message, base_message) + + errors = error_msg.get_errors() + assert errors[0]["code"] == "E001" # Overridden + assert errors[0]["description"] == "Override description" # Overridden + assert errors[0]["timestamp"] == "2025-10-24" # From base + assert errors[0]["level"] == "Error" # Set by add_error + + def test_sequential_operations(self): + """Test sequential add and reset operations""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + error_msg.add_error({"code": "E001"}) + assert len(error_msg.get_errors()) == 1 + + error_msg.add_error({"code": "E002"}) + assert len(error_msg.get_errors()) == 2 + + error_msg.reset_errors() + assert len(error_msg.get_errors()) == 0 + + error_msg.add_error({"code": "E003"}) + assert len(error_msg.get_errors()) == 1 + assert error_msg.get_errors()[0]["code"] == "E003" + + +class TestParametrized: + """Parametrized tests for comprehensive coverage""" + + @pytest.mark.parametrize("base_message,message,expected_keys", [ + (None, {"code": "E001"}, {"code", "level"}), + ({}, {"code": "E001"}, {"code", "level"}), + ({"timestamp": "2025-10-24"}, {"code": "E001"}, {"code", "level", "timestamp"}), + ({"a": 1, "b": 2}, {"c": 3}, {"a", "b", "c", "level"}), + ]) + def test_error_message_merge_parametrized( + self, + base_message: dict[str, Any] | None, + message: dict[str, Any], + expected_keys: set[str] + ): + """Test error message merging with various combinations""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + error_msg.add_error(message, base_message) + errors = error_msg.get_errors() + + assert len(errors) == 1 + assert set(errors[0].keys()) == expected_keys + assert errors[0]["level"] == "Error" + + @pytest.mark.parametrize("base_message,message,expected_keys", [ + (None, {"code": "W001"}, {"code", "level"}), + ({}, {"code": "W001"}, {"code", "level"}), + ({"timestamp": "2025-10-24"}, {"code": "W001"}, {"code", "level", "timestamp"}), + ({"a": 1, "b": 2}, {"c": 3}, {"a", "b", "c", "level"}), + ]) + def test_warning_message_merge_parametrized( + self, + base_message: dict[str, Any] | None, + message: dict[str, Any], + expected_keys: set[str] + ): + """Test warning message merging with various combinations""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + error_msg.add_warning(message, base_message) + warnings = error_msg.get_warnings() + + assert len(warnings) == 1 + assert set(warnings[0].keys()) == expected_keys + assert warnings[0]["level"] == "Warning" + + @pytest.mark.parametrize("count", [0, 1, 5, 10, 100]) + def test_multiple_errors_parametrized(self, count: int): + """Test adding multiple errors""" + error_msg = ErrorMessage() + error_msg.reset_errors() + + for i in range(count): + error_msg.add_error({"code": f"E{i:03d}"}) + + errors = error_msg.get_errors() + assert len(errors) == count + assert error_msg.has_errors() == (count > 0) + + @pytest.mark.parametrize("count", [0, 1, 5, 10, 100]) + def test_multiple_warnings_parametrized(self, count: int): + """Test adding multiple warnings""" + error_msg = ErrorMessage() + error_msg.reset_warnings() + + for i in range(count): + error_msg.add_warning({"code": f"W{i:03d}"}) + + warnings = error_msg.get_warnings() + assert len(warnings) == count + assert error_msg.has_warnings() == (count > 0) + +# __END__