# [DEF:test_task_logger:Module] # @SEMANTICS: test, task_logger, task_context, unit_test # @PURPOSE: Unit tests for TaskLogger and TaskContext. # @LAYER: Test # @RELATION: TESTS -> TaskLogger, TaskContext # @TIER: STANDARD # [SECTION: IMPORTS] import pytest from unittest.mock import Mock, MagicMock from datetime import datetime from src.core.task_manager.task_logger import TaskLogger from src.core.task_manager.context import TaskContext from src.core.task_manager.models import LogEntry # [/SECTION] # [DEF:TestTaskLogger:Class] # @PURPOSE: Test suite for TaskLogger. # @TIER: STANDARD class TestTaskLogger: # [DEF:setup_method:Function] # @PURPOSE: Setup for each test method. # @PRE: None. # @POST: Mock add_log_fn created. def setup_method(self): """Create a mock add_log function for testing.""" self.mock_add_log = Mock() self.logger = TaskLogger( task_id="test-task-1", add_log_fn=self.mock_add_log, source="test_source" ) # [/DEF:setup_method:Function] # [DEF:test_init:Function] # @PURPOSE: Test TaskLogger initialization. # @PRE: None. # @POST: Logger instance created with correct attributes. def test_init(self): """Test TaskLogger initialization.""" assert self.logger._task_id == "test-task-1" assert self.logger._default_source == "test_source" assert self.logger._add_log == self.mock_add_log # [/DEF:test_init:Function] # [DEF:test_with_source:Function] # @PURPOSE: Test creating a sub-logger with different source. # @PRE: Logger initialized. # @POST: New logger created with different source but same task_id. def test_with_source(self): """Test creating a sub-logger with different source.""" sub_logger = self.logger.with_source("new_source") assert sub_logger._task_id == "test-task-1" assert sub_logger._default_source == "new_source" assert sub_logger._add_log == self.mock_add_log # [/DEF:test_with_source:Function] # [DEF:test_debug:Function] # @PURPOSE: Test debug log level. # @PRE: Logger initialized. # @POST: add_log_fn called with DEBUG level. def test_debug(self): """Test debug logging.""" self.logger.debug("Debug message") self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="DEBUG", message="Debug message", source="test_source", metadata=None ) # [/DEF:test_debug:Function] # [DEF:test_info:Function] # @PURPOSE: Test info log level. # @PRE: Logger initialized. # @POST: add_log_fn called with INFO level. def test_info(self): """Test info logging.""" self.logger.info("Info message") self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="INFO", message="Info message", source="test_source", metadata=None ) # [/DEF:test_info:Function] # [DEF:test_warning:Function] # @PURPOSE: Test warning log level. # @PRE: Logger initialized. # @POST: add_log_fn called with WARNING level. def test_warning(self): """Test warning logging.""" self.logger.warning("Warning message") self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="WARNING", message="Warning message", source="test_source", metadata=None ) # [/DEF:test_warning:Function] # [DEF:test_error:Function] # @PURPOSE: Test error log level. # @PRE: Logger initialized. # @POST: add_log_fn called with ERROR level. def test_error(self): """Test error logging.""" self.logger.error("Error message") self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="ERROR", message="Error message", source="test_source", metadata=None ) # [/DEF:test_error:Function] # [DEF:test_error_with_metadata:Function] # @PURPOSE: Test error logging with metadata. # @PRE: Logger initialized. # @POST: add_log_fn called with ERROR level and metadata. def test_error_with_metadata(self): """Test error logging with metadata.""" metadata = {"error_code": 500, "details": "Connection failed"} self.logger.error("Error message", metadata=metadata) self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="ERROR", message="Error message", source="test_source", metadata=metadata ) # [/DEF:test_error_with_metadata:Function] # [DEF:test_progress:Function] # @PURPOSE: Test progress logging. # @PRE: Logger initialized. # @POST: add_log_fn called with INFO level and progress metadata. def test_progress(self): """Test progress logging.""" self.logger.progress("Processing items", percent=50) expected_metadata = {"progress": 50} self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="INFO", message="Processing items", source="test_source", metadata=expected_metadata ) # [/DEF:test_progress:Function] # [DEF:test_progress_clamping:Function] # @PURPOSE: Test progress value clamping (0-100). # @PRE: Logger initialized. # @POST: Progress values clamped to 0-100 range. def test_progress_clamping(self): """Test progress value clamping.""" # Test below 0 self.logger.progress("Below 0", percent=-10) call1 = self.mock_add_log.call_args_list[0] assert call1.kwargs["metadata"]["progress"] == 0 self.mock_add_log.reset_mock() # Test above 100 self.logger.progress("Above 100", percent=150) call2 = self.mock_add_log.call_args_list[0] assert call2.kwargs["metadata"]["progress"] == 100 # [/DEF:test_progress_clamping:Function] # [DEF:test_source_override:Function] # @PURPOSE: Test overriding the default source. # @PRE: Logger initialized. # @POST: add_log_fn called with overridden source. def test_source_override(self): """Test overriding the default source.""" self.logger.info("Message", source="override_source") self.mock_add_log.assert_called_once_with( task_id="test-task-1", level="INFO", message="Message", source="override_source", metadata=None ) # [/DEF:test_source_override:Function] # [DEF:test_sub_logger_source_independence:Function] # @PURPOSE: Test sub-logger independence from parent. # @PRE: Logger and sub-logger initialized. # @POST: Sub-logger has different source, parent unchanged. def test_sub_logger_source_independence(self): """Test sub-logger source independence from parent.""" sub_logger = self.logger.with_source("sub_source") # Log with parent self.logger.info("Parent message") # Log with sub-logger sub_logger.info("Sub message") # Verify both calls were made with correct sources calls = self.mock_add_log.call_args_list assert len(calls) == 2 assert calls[0].kwargs["source"] == "test_source" assert calls[1].kwargs["source"] == "sub_source" # [/DEF:test_sub_logger_source_independence:Function] # [/DEF:TestTaskLogger:Class] # [DEF:TestTaskContext:Class] # @PURPOSE: Test suite for TaskContext. # @TIER: STANDARD class TestTaskContext: # [DEF:setup_method:Function] # @PURPOSE: Setup for each test method. # @PRE: None. # @POST: Mock add_log_fn created. def setup_method(self): """Create a mock add_log function for testing.""" self.mock_add_log = Mock() self.params = {"param1": "value1", "param2": "value2"} self.context = TaskContext( task_id="test-task-2", add_log_fn=self.mock_add_log, params=self.params, default_source="plugin" ) # [/DEF:setup_method:Function] # [DEF:test_init:Function] # @PURPOSE: Test TaskContext initialization. # @PRE: None. # @POST: Context instance created with correct attributes. def test_init(self): """Test TaskContext initialization.""" assert self.context._task_id == "test-task-2" assert self.context._params == self.params assert isinstance(self.context._logger, TaskLogger) assert self.context._logger._default_source == "plugin" # [/DEF:test_init:Function] # [DEF:test_task_id_property:Function] # @PURPOSE: Test task_id property. # @PRE: Context initialized. # @POST: Returns correct task_id. def test_task_id_property(self): """Test task_id property.""" assert self.context.task_id == "test-task-2" # [/DEF:test_task_id_property:Function] # [DEF:test_logger_property:Function] # @PURPOSE: Test logger property. # @PRE: Context initialized. # @POST: Returns TaskLogger instance. def test_logger_property(self): """Test logger property.""" logger = self.context.logger assert isinstance(logger, TaskLogger) assert logger._task_id == "test-task-2" assert logger._default_source == "plugin" # [/DEF:test_logger_property:Function] # [DEF:test_params_property:Function] # @PURPOSE: Test params property. # @PRE: Context initialized. # @POST: Returns correct params dict. def test_params_property(self): """Test params property.""" assert self.context.params == self.params # [/DEF:test_params_property:Function] # [DEF:test_get_param:Function] # @PURPOSE: Test getting a specific parameter. # @PRE: Context initialized with params. # @POST: Returns parameter value or default. def test_get_param(self): """Test getting a specific parameter.""" assert self.context.get_param("param1") == "value1" assert self.context.get_param("param2") == "value2" assert self.context.get_param("nonexistent") is None assert self.context.get_param("nonexistent", "default") == "default" # [/DEF:test_get_param:Function] # [DEF:test_create_sub_context:Function] # @PURPOSE: Test creating a sub-context with different source. # @PRE: Context initialized. # @POST: New context created with different logger source. def test_create_sub_context(self): """Test creating a sub-context with different source.""" sub_context = self.context.create_sub_context("new_source") assert sub_context._task_id == "test-task-2" assert sub_context._params == self.params assert sub_context._logger._default_source == "new_source" assert sub_context._logger._task_id == "test-task-2" # [/DEF:test_create_sub_context:Function] # [DEF:test_context_logger_delegates_to_task_logger:Function] # @PURPOSE: Test context logger delegates to TaskLogger. # @PRE: Context initialized. # @POST: Logger calls are delegated to TaskLogger. def test_context_logger_delegates_to_task_logger(self): """Test context logger delegates to TaskLogger.""" # Call through context self.context.logger.info("Test message") # Verify the mock was called self.mock_add_log.assert_called_once_with( task_id="test-task-2", level="INFO", message="Test message", source="plugin", metadata=None ) # [/DEF:test_context_logger_delegates_to_task_logger:Function] # [DEF:test_sub_context_with_source:Function] # @PURPOSE: Test sub-context logger uses new source. # @PRE: Context initialized. # @POST: Sub-context logger uses new source. def test_sub_context_with_source(self): """Test sub-context logger uses new source.""" sub_context = self.context.create_sub_context("api_source") # Log through sub-context sub_context.logger.info("API message") # Verify the mock was called with new source self.mock_add_log.assert_called_once_with( task_id="test-task-2", level="INFO", message="API message", source="api_source", metadata=None ) # [/DEF:test_sub_context_with_source:Function] # [DEF:test_multiple_sub_contexts:Function] # @PURPOSE: Test creating multiple sub-contexts. # @PRE: Context initialized. # @POST: Each sub-context has independent logger source. def test_multiple_sub_contexts(self): """Test creating multiple sub-contexts.""" sub1 = self.context.create_sub_context("source1") sub2 = self.context.create_sub_context("source2") sub3 = self.context.create_sub_context("source3") assert sub1._logger._default_source == "source1" assert sub2._logger._default_source == "source2" assert sub3._logger._default_source == "source3" # All should have same task_id and params assert sub1._task_id == "test-task-2" assert sub2._task_id == "test-task-2" assert sub3._task_id == "test-task-2" assert sub1._params == self.params assert sub2._params == self.params assert sub3._params == self.params # [/DEF:test_multiple_sub_contexts:Function] # [/DEF:TestTaskContext:Class] # [/DEF:test_task_logger:Module]