"""Tests for logging configuration and PII/secrets sanitization."""
from mcp_task_aggregator.logging import sanitize_message
class TestSanitizeMessage:
"""Tests for the sanitize_message function."""
def test_sanitizes_home_directory_path(self) -> None:
"""Test that home directory paths are replaced with ~."""
from pathlib import Path
home = str(Path.home())
message = f"Database at {home}/.mcp-task-aggregator/tasks.db"
result = sanitize_message(message)
assert home not in result
assert "~/.mcp-task-aggregator/tasks.db" in result
def test_sanitizes_api_key(self) -> None:
"""Test that API keys are redacted."""
message = "Using api_key=FAKE_TEST_KEY_123"
result = sanitize_message(message)
assert "FAKE_TEST_KEY_123" not in result
assert "<REDACTED>" in result
def test_sanitizes_token(self) -> None:
"""Test that tokens are redacted."""
message = "Auth token: FAKE_TEST_TOKEN"
result = sanitize_message(message)
assert "FAKE_TEST_TOKEN" not in result
assert "<REDACTED>" in result
def test_sanitizes_bearer_token(self) -> None:
"""Test that Bearer tokens are redacted."""
message = "Authorization: Bearer FAKE_JWT_TEST_VALUE"
result = sanitize_message(message)
assert "FAKE_JWT_TEST_VALUE" not in result
assert "Bearer <REDACTED>" in result
def test_sanitizes_email(self) -> None:
"""Test that email addresses are redacted."""
message = "Assigned to testuser@example.test"
result = sanitize_message(message)
assert "testuser@example.test" not in result
assert "<EMAIL>" in result
def test_sanitizes_url_with_credentials(self) -> None:
"""Test that URLs with embedded credentials are redacted."""
message = "Connecting to https://fakeuser:fakevalue@jira.example.test/api"
result = sanitize_message(message)
assert "fakeuser" not in result
assert "fakevalue" not in result
assert "https://<USER>:<PASS>@jira.example.test/api" in result
def test_sanitizes_uuid(self) -> None:
"""Test that UUIDs (cloud IDs) are redacted."""
message = "Cloud ID: a1b2c3d4-e5f6-7890-abcd-ef1234567890"
result = sanitize_message(message)
assert "a1b2c3d4-e5f6-7890-abcd-ef1234567890" not in result
assert "<UUID>" in result
def test_preserves_normal_text(self) -> None:
"""Test that normal text is not modified."""
message = "Fetched 10 tasks from Jira"
result = sanitize_message(message)
assert result == message
def test_sanitizes_secret_in_config(self) -> None:
"""Test that secret values are redacted."""
message = "Config: secret='FAKE_SECRET_VALUE'"
result = sanitize_message(message)
assert "FAKE_SECRET_VALUE" not in result
assert "<REDACTED>" in result
def test_multiple_patterns_in_one_message(self) -> None:
"""Test that multiple sensitive items are all sanitized."""
message = "User test@example.test connected with api_key=FAKEKEY"
result = sanitize_message(message)
assert "test@example.test" not in result
assert "FAKEKEY" not in result
assert "<EMAIL>" in result
assert "<REDACTED>" in result
def test_sanitizes_users_path_macos(self) -> None:
"""Test that /Users/username paths are sanitized."""
import os
username = os.environ.get("USER", "testuser")
message = f"Found file at /Users/{username}/Documents/data.txt"
result = sanitize_message(message)
assert f"/Users/{username}" not in result
assert "~/Documents/data.txt" in result
def test_sanitizes_home_path_linux(self) -> None:
"""Test that /home/username paths are sanitized."""
import os
username = os.environ.get("USER", "testuser")
message = f"Found file at /home/{username}/Documents/data.txt"
result = sanitize_message(message)
assert f"/home/{username}" not in result
assert "~/Documents/data.txt" in result
class TestSanitizingFormat:
"""Tests for the log format functions."""
def test_sanitizing_format_sanitizes_message(self) -> None:
"""Test that _sanitizing_format sanitizes sensitive data."""
from mcp_task_aggregator.logging import _sanitizing_format
record = {
"message": "api_key=FAKE_SECRET_123",
"extra": {},
}
result = _sanitizing_format(record)
assert "sanitized_message" in record["extra"]
assert "FAKE_SECRET_123" not in record["extra"]["sanitized_message"]
assert "<REDACTED>" in record["extra"]["sanitized_message"]
assert isinstance(result, str)
def test_sanitizing_file_format_sanitizes_message(self) -> None:
"""Test that _sanitizing_file_format sanitizes sensitive data."""
from mcp_task_aggregator.logging import _sanitizing_file_format
record = {
"message": "token=FAKE_FILE_TOKEN",
"extra": {},
}
result = _sanitizing_file_format(record)
assert "sanitized_message" in record["extra"]
assert "FAKE_FILE_TOKEN" not in record["extra"]["sanitized_message"]
assert "<REDACTED>" in record["extra"]["sanitized_message"]
assert isinstance(result, str)
class TestSetupLogging:
"""Tests for setup_logging function."""
def test_setup_logging_configures_console_handler(self) -> None:
"""Test that setup_logging configures console output."""
from unittest.mock import patch
from mcp_task_aggregator.logging import setup_logging
with patch("mcp_task_aggregator.logging.get_settings") as mock_settings:
mock_settings.return_value.log_level = "INFO"
mock_settings.return_value.log_file = None
with patch("mcp_task_aggregator.logging.logger") as mock_logger:
setup_logging()
mock_logger.remove.assert_called_once()
mock_logger.add.assert_called_once()
mock_logger.info.assert_called_once()
def test_setup_logging_with_file_handler(self) -> None:
"""Test that setup_logging configures file output when log_file is set."""
import tempfile
from pathlib import Path
from unittest.mock import patch
from mcp_task_aggregator.logging import setup_logging
with tempfile.TemporaryDirectory() as tmpdir:
log_file = Path(tmpdir) / "test.log"
with patch("mcp_task_aggregator.logging.get_settings") as mock_settings:
mock_settings.return_value.log_level = "DEBUG"
mock_settings.return_value.log_file = log_file
with patch("mcp_task_aggregator.logging.logger") as mock_logger:
setup_logging()
mock_logger.remove.assert_called_once()
# Should be called twice: once for stderr, once for file
assert mock_logger.add.call_count == 2
class TestGetLogger:
"""Tests for get_logger function."""
def test_get_logger_returns_bound_logger(self) -> None:
"""Test that get_logger returns a logger bound to the given name."""
from mcp_task_aggregator.logging import get_logger
logger = get_logger("test_module")
assert logger is not None
def test_get_logger_binds_name(self) -> None:
"""Test that get_logger binds the name to the logger."""
from unittest.mock import patch
from mcp_task_aggregator.logging import get_logger
with patch("mcp_task_aggregator.logging.logger") as mock_logger:
get_logger("my_module")
mock_logger.bind.assert_called_once_with(name="my_module")
class TestConfigureLoggingForTests:
"""Tests for configure_logging_for_tests function."""
def test_configure_logging_for_tests_default(self) -> None:
"""Test configure_logging_for_tests with default settings."""
from unittest.mock import patch
from mcp_task_aggregator.logging import configure_logging_for_tests
with patch("mcp_task_aggregator.logging.logger") as mock_logger:
configure_logging_for_tests()
mock_logger.remove.assert_called_once()
mock_logger.add.assert_called_once()
def test_configure_logging_for_tests_with_file(self) -> None:
"""Test configure_logging_for_tests with file output."""
import tempfile
from pathlib import Path
from unittest.mock import patch
from mcp_task_aggregator.logging import configure_logging_for_tests
with tempfile.TemporaryDirectory() as tmpdir:
log_file = Path(tmpdir) / "test.log"
with patch("mcp_task_aggregator.logging.logger") as mock_logger:
configure_logging_for_tests(log_level="WARNING", log_file=log_file)
mock_logger.remove.assert_called_once()
# Should be called twice: once for stderr, once for file
assert mock_logger.add.call_count == 2
def test_configure_logging_for_tests_custom_level(self) -> None:
"""Test configure_logging_for_tests with custom log level."""
from unittest.mock import patch
from mcp_task_aggregator.logging import configure_logging_for_tests
with patch("mcp_task_aggregator.logging.logger") as mock_logger:
configure_logging_for_tests(log_level="ERROR")
mock_logger.remove.assert_called_once()
# Check that the level was passed correctly
add_call = mock_logger.add.call_args
assert add_call.kwargs.get("level") == "ERROR"
