jpmschweitzerandClaude
5b67f5b66c
The 21 the automatic pass could not make on its own. `ruff check` and `ruff format --check` are both clean now; typecheck is still red and is next. `in_reasoning` in chat/service.py was a complete state machine that nothing read: initialised False, set True when a reasoning delta arrived, set False when the summary ended — three assignments, zero reads. Ruff reported one at a time, and removing each revealed the next, so what looked like a single stray variable took three passes to bottom out. The branches themselves do real work and are untouched; only the flag is gone. Four `raise HTTPException` inside `except` blocks now chain with `from e`. Until now a failure while handling an error was indistinguishable from the error, which matters most in exactly the situation where the traceback is all you have. In biographer/tools.py the binding was unused but the call is not: MemoryType() is called for the ValueError it raises on an invalid name. The binding is gone and the call and its comment stay, because dropping the line would have removed the validation. The rest are unused bindings in tests where the assertions are on something else (call_args, mostly), plus three unused loop variables and an isinstance tuple. One correction to my own work: removing a dead comprehension in test_error_handling.py left an `if` block with nothing but comments in it, which is a SyntaxError. Ruff caught it immediately. The block now says what the test actually pins — that the stream parses without crashing, which reaching that line demonstrates — rather than computing a list nobody asserts on. `make test` is intermittent here, and it is not this change. test_tatlock_tool_call_logging_calculator failed in two of five full runs across both HEAD and this branch, and passes in the other three; it also fails in isolation at HEAD while passing in isolation here. Order- or timing-dependent. Recorded rather than chased, since tests are not gated in this repo yet. Co-Authored-By: Claude <noreply@anthropic.com>
254 lines
7.9 KiB
Python
254 lines
7.9 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_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")
|