jpmschweitzerandClaude Fable 5
aace128ef9
docker logs tatlock has been empty for months and the /app/logs
mount untouched since 2025-12. Not a broken handler: the stdout
StreamHandler was correct all along, but the production default
level was WARNING ("minimal noise") and a healthy service warns
roughly never — startup, backend flavor detection and the wrapper's
balancing/compaction signals were all suppressed, which is exactly
what made the serving cutover unverifiable from logs on 2026-09-12.
Production now defaults to INFO; LOG_LEVEL stays the override for
when noise is ever the real problem; development keeps DEBUG. The
mount was always decorative — no file handler exists, docker logs is
the log. Three tests pin the level table, the WARNING-revert
mutation shown to fail its test.
Also filed T-7: the full suite flaked once and passed on rerun for
the second time — next occurrence keeps the failing run's output.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
289 lines
9.3 KiB
Python
289 lines
9.3 KiB
Python
"""
|
|
Tests for structured logging configuration.
|
|
|
|
Tests logging setup, context management, and FastAPI integration.
|
|
"""
|
|
|
|
import logging
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from src.core.logging_config import (
|
|
add_log_level,
|
|
add_timestamp,
|
|
get_logger,
|
|
get_uvicorn_log_config,
|
|
log_operation,
|
|
)
|
|
|
|
|
|
class TestLoggingProcessors:
|
|
"""Test logging processor functions."""
|
|
|
|
def test_add_timestamp(self):
|
|
"""Test timestamp processor adds ISO timestamp."""
|
|
event_dict = {}
|
|
result = add_timestamp(None, "info", event_dict)
|
|
|
|
assert "timestamp" in result
|
|
assert isinstance(result["timestamp"], str)
|
|
# Should be ISO 8601 format
|
|
assert "T" in result["timestamp"] or "-" in result["timestamp"]
|
|
|
|
def test_add_log_level(self):
|
|
"""Test log level processor."""
|
|
event_dict = {}
|
|
result = add_log_level(None, "info", event_dict)
|
|
|
|
assert result["level"] == "INFO"
|
|
|
|
result = add_log_level(None, "error", {})
|
|
assert result["level"] == "ERROR"
|
|
|
|
|
|
class TestGetLogger:
|
|
"""Test logger retrieval."""
|
|
|
|
def test_get_logger_returns_bound_logger(self):
|
|
"""Test get_logger returns structlog BoundLogger."""
|
|
logger = get_logger("test")
|
|
# Logger should have standard logging methods
|
|
assert hasattr(logger, "info")
|
|
assert hasattr(logger, "debug")
|
|
assert hasattr(logger, "warning")
|
|
assert hasattr(logger, "error")
|
|
|
|
def test_get_logger_with_module_name(self):
|
|
"""Test logger with module name."""
|
|
logger = get_logger(__name__)
|
|
assert logger is not None
|
|
|
|
def test_logger_has_standard_methods(self):
|
|
"""Test logger has standard logging methods."""
|
|
logger = get_logger("test")
|
|
assert hasattr(logger, "debug")
|
|
assert hasattr(logger, "info")
|
|
assert hasattr(logger, "warning")
|
|
assert hasattr(logger, "error")
|
|
assert hasattr(logger, "exception")
|
|
|
|
|
|
class TestLogOperation:
|
|
"""Test log_operation context manager."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_operation_success(self):
|
|
"""Test log_operation for successful operation."""
|
|
get_logger("test")
|
|
|
|
async with log_operation("test_operation", {"user_id": "123"}) as ctx:
|
|
# Can update context during operation
|
|
ctx["result_count"] = 5
|
|
|
|
# Context should have been updated with success info
|
|
assert ctx["success"] is True
|
|
assert ctx["result_count"] == 5
|
|
assert "duration_seconds" in ctx
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_operation_failure(self):
|
|
"""Test log_operation for failed operation."""
|
|
get_logger("test")
|
|
|
|
with pytest.raises(ValueError):
|
|
async with log_operation("test_operation") as ctx:
|
|
raise ValueError("Test error")
|
|
|
|
# Context should have failure info
|
|
assert ctx["success"] is False
|
|
assert ctx["error"] == "Test error"
|
|
assert ctx["error_type"] == "ValueError"
|
|
assert "duration_seconds" in ctx
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_operation_timing(self):
|
|
"""Test log_operation records duration."""
|
|
import asyncio
|
|
|
|
async with log_operation("test_operation") as ctx:
|
|
await asyncio.sleep(0.01) # Small delay
|
|
|
|
# Should have measurable duration
|
|
assert ctx["duration_seconds"] > 0
|
|
assert ctx["duration_seconds"] < 1.0 # Should be quick
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_operation_initial_context(self):
|
|
"""Test log_operation with initial context."""
|
|
initial = {"request_id": "abc123", "user": "test_user"}
|
|
|
|
async with log_operation("test_operation", initial) as ctx:
|
|
pass
|
|
|
|
# Initial context should be preserved
|
|
assert ctx["request_id"] == "abc123"
|
|
assert ctx["user"] == "test_user"
|
|
assert ctx["operation"] == "test_operation"
|
|
|
|
|
|
class TestUvicornLogConfig:
|
|
"""Test uvicorn logging configuration."""
|
|
|
|
def test_get_uvicorn_log_config_returns_dict(self):
|
|
"""Test uvicorn config returns valid dict."""
|
|
config = get_uvicorn_log_config()
|
|
|
|
assert isinstance(config, dict)
|
|
assert "version" in config
|
|
assert "formatters" in config
|
|
assert "handlers" in config
|
|
assert "loggers" in config
|
|
|
|
def test_uvicorn_log_config_has_required_loggers(self):
|
|
"""Test config includes uvicorn loggers."""
|
|
config = get_uvicorn_log_config()
|
|
|
|
loggers = config["loggers"]
|
|
assert "uvicorn" in loggers
|
|
assert "uvicorn.error" in loggers
|
|
assert "uvicorn.access" in loggers
|
|
|
|
def test_uvicorn_log_config_format_selection(self):
|
|
"""Test config format changes based on environment."""
|
|
# Just test that the config is valid, format is determined by environment
|
|
config = get_uvicorn_log_config()
|
|
# Should have required structure
|
|
assert "version" in config
|
|
assert "formatters" in config
|
|
assert "handlers" in config
|
|
assert "loggers" in config
|
|
|
|
|
|
class TestLoggingIntegration:
|
|
"""Test logging integration with standard library."""
|
|
|
|
def test_standard_logging_works(self):
|
|
"""Test standard logging.getLogger works."""
|
|
logger = logging.getLogger("test.standard")
|
|
# Should not raise
|
|
logger.info("Test message")
|
|
|
|
def test_structlog_and_stdlib_coexist(self):
|
|
"""Test structlog and stdlib can coexist."""
|
|
struct_logger = get_logger("test.struct")
|
|
std_logger = logging.getLogger("test.std")
|
|
|
|
# Both should work
|
|
struct_logger.info("Structured log")
|
|
std_logger.info("Standard log")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logging_in_async_context(self):
|
|
"""Test logging works in async context."""
|
|
logger = get_logger("test.async")
|
|
|
|
async def async_function():
|
|
logger.info("Async log message", task="async_task")
|
|
|
|
await async_function()
|
|
|
|
|
|
class TestLoggingOutput:
|
|
"""Test actual logging output."""
|
|
|
|
def test_logger_outputs_structured_data(self):
|
|
"""Test logger can output structured data."""
|
|
logger = get_logger("test.output")
|
|
|
|
# Log with structured data
|
|
logger.info(
|
|
"user_action",
|
|
user_id="123",
|
|
action="login",
|
|
success=True,
|
|
)
|
|
# Should not raise, output tested in integration tests
|
|
|
|
def test_logger_handles_exceptions(self):
|
|
"""Test logger handles exception logging."""
|
|
logger = get_logger("test.exceptions")
|
|
|
|
try:
|
|
raise ValueError("Test error")
|
|
except ValueError:
|
|
logger.exception("Error occurred", extra_field="value")
|
|
# Should not raise
|
|
|
|
def test_different_log_levels(self):
|
|
"""Test different log levels."""
|
|
logger = get_logger("test.levels")
|
|
|
|
logger.debug("Debug message", level="debug")
|
|
logger.info("Info message", level="info")
|
|
logger.warning("Warning message", level="warning")
|
|
logger.error("Error message", level="error")
|
|
# Should not raise
|
|
|
|
|
|
class TestLoggingConfiguration:
|
|
"""Test logging configuration behavior."""
|
|
|
|
def test_logging_respects_environment(self):
|
|
"""Test logging format changes with environment."""
|
|
from src.core.config import Environment, config
|
|
|
|
# In development, should use console format
|
|
if config.ENVIRONMENT == Environment.DEVELOPMENT:
|
|
assert config.log_format == "console"
|
|
|
|
# Mock production environment
|
|
with patch.object(config, "ENVIRONMENT", Environment.PRODUCTION):
|
|
assert config.log_format == "json"
|
|
|
|
def test_production_default_level_is_info(self):
|
|
"""Production logs INFO to stdout (T-5).
|
|
|
|
The old WARNING default made `docker logs` empty for months — a
|
|
healthy service warns roughly never, so startup, flavor
|
|
detection and the wrapper signals were all suppressed and the
|
|
serving cutover could not be verified from logs.
|
|
"""
|
|
from src.core.config import Environment, config
|
|
|
|
with (
|
|
patch.object(config, "ENVIRONMENT", Environment.PRODUCTION),
|
|
patch.object(config, "LOG_LEVEL", None),
|
|
):
|
|
assert config.effective_log_level == "INFO"
|
|
|
|
def test_log_level_env_still_quiets_production(self):
|
|
"""LOG_LEVEL stays the override for when noise is the problem."""
|
|
from src.core.config import Environment, config
|
|
|
|
with (
|
|
patch.object(config, "ENVIRONMENT", Environment.PRODUCTION),
|
|
patch.object(config, "LOG_LEVEL", "WARNING"),
|
|
):
|
|
assert config.effective_log_level == "WARNING"
|
|
|
|
def test_development_default_stays_debug(self):
|
|
from src.core.config import Environment, config
|
|
|
|
with (
|
|
patch.object(config, "ENVIRONMENT", Environment.DEVELOPMENT),
|
|
patch.object(config, "LOG_LEVEL", None),
|
|
):
|
|
assert config.effective_log_level == "DEBUG"
|
|
|
|
def test_multiple_loggers_independent(self):
|
|
"""Test multiple loggers are independent."""
|
|
logger1 = get_logger("test.logger1")
|
|
logger2 = get_logger("test.logger2")
|
|
|
|
assert logger1 is not logger2
|
|
|
|
# Both should work independently
|
|
logger1.info("Logger 1 message")
|
|
logger2.info("Logger 2 message")
|