378 lines
14 KiB
Python
378 lines
14 KiB
Python
# [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]
|