Add pytest for logging
This commit is contained in:
Clemens Schwaighofer
@@ -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()
|
||||
|
||||
|
||||
@@ -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__
|
||||
@@ -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__
|
||||
@@ -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__
|
||||
@@ -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__
|
||||
@@ -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__
|
||||
92
tests/unit/logging_handling/log_testing/test_log_6_logger.py
Normal file
92
tests/unit/logging_handling/log_testing/test_log_6_logger.py
Normal file
@@ -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__
|
||||
113
tests/unit/logging_handling/log_testing/test_log_7_edge_cases.py
Normal file
113
tests/unit/logging_handling/log_testing/test_log_7_edge_cases.py
Normal file
@@ -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__
|
||||
@@ -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__
|
||||
503
tests/unit/logging_handling/test_error_handling.py
Normal file
503
tests/unit/logging_handling/test_error_handling.py
Normal file
@@ -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__
|
||||
Reference in New Issue
Block a user