Testing Utilities

class ReturnLogger:
    """
    Logger that returns the arguments it's called with instead of logging.
    
    Useful for testing to capture what would have been logged without
    actually producing any output.
    """
    
    def msg(self, *args, **kw):
        """
        Return arguments instead of logging.
        
        Args:
            *args: Positional arguments
            **kw: Keyword arguments
            
        Returns:
            - Single argument if only one arg and no kwargs
            - Tuple of (args, kwargs) otherwise
        """
    
    # Alias methods for different log levels
    def debug(self, *args, **kw): ...
    def info(self, *args, **kw): ...
    def warning(self, *args, **kw): ...
    def warn(self, *args, **kw): ...
    def error(self, *args, **kw): ...
    def critical(self, *args, **kw): ...
    def fatal(self, *args, **kw): ...
    def exception(self, *args, **kw): ...

class ReturnLoggerFactory:
    """Factory for creating ReturnLogger instances."""
    
    def __call__(self, *args) -> ReturnLogger:
        """
        Create a ReturnLogger instance.
        
        Args:
            *args: Arguments (ignored)
            
        Returns:
            ReturnLogger: New ReturnLogger instance
        """

class CapturingLogger:
    """
    Logger that stores all method calls in a list for inspection.
    
    Captures all logging calls with their arguments for later assertion
    in test cases.
    """
    
    calls: list[CapturedCall]
    """List of captured logging calls."""
    
    def __getattr__(self, name):
        """
        Handle any method call by capturing it.
        
        Args:
            name (str): Method name
            
        Returns:
            callable: Function that captures the call
        """

class CapturingLoggerFactory:
    """Factory for creating CapturingLogger instances."""
    
    logger: CapturingLogger
    """The CapturingLogger instance created by this factory."""
    
    def __call__(self, *args) -> CapturingLogger:
        """
        Create or return the CapturingLogger instance.
        
        Args:
            *args: Arguments (ignored)
            
        Returns:
            CapturingLogger: The logger instance
        """

class CapturedCall(NamedTuple):
    """
    Represents a captured logging method call.
    
    Contains the method name and arguments for a single logging call.
    """
    
    method_name: str
    """Name of the logging method that was called."""
    
    args: tuple[Any, ...]
    """Positional arguments passed to the method."""
    
    kwargs: dict[str, Any]
    """Keyword arguments passed to the method."""

class LogCapture:
    """
    Processor that captures log messages in a list.
    
    Stores all processed log entries in the entries list and raises
    DropEvent to prevent further processing.
    """
    
    entries: list[EventDict]
    """List of captured log entries (event dictionaries)."""
    
    def __call__(self, logger, method_name, event_dict) -> NoReturn:
        """
        Capture log entry and drop event.
        
        Args:
            logger: Logger instance
            method_name (str): Logger method name
            event_dict (dict): Event dictionary
            
        Raises:
            DropEvent: Always raised to stop further processing
        """

def capture_logs() -> Generator[list[EventDict], None, None]:
    """
    Context manager for capturing log messages.
    
    Temporarily configures structlog to capture all log entries
    in a list, which is yielded to the with block.
    
    Yields:
        list: List that will contain captured EventDict objects
        
    Example:
        with capture_logs() as captured:
            logger.info("test message", value=42)
            assert len(captured) == 1
            assert captured[0]["event"] == "test message"
    """

import structlog
from structlog.testing import ReturnLoggerFactory

def test_logging_behavior():
    # Configure structlog with ReturnLogger
    structlog.configure(
        processors=[],  # No processors needed for testing
        wrapper_class=structlog.BoundLogger,
        logger_factory=ReturnLoggerFactory(),
        cache_logger_on_first_use=False,  # Don't cache for testing
    )
    
    logger = structlog.get_logger()
    
    # Test single argument
    result = logger.info("Simple message")
    assert result == "Simple message"
    
    # Test multiple arguments
    result = logger.info("Message with data", user_id=123, action="login")
    expected_args = ("Message with data",)
    expected_kwargs = {"user_id": 123, "action": "login"}
    assert result == (expected_args, expected_kwargs)

import structlog
from structlog.testing import CapturingLoggerFactory, CapturedCall

def test_multiple_log_calls():
    # Set up capturing
    cap_factory = CapturingLoggerFactory()
    
    structlog.configure(
        processors=[],
        wrapper_class=structlog.BoundLogger,
        logger_factory=cap_factory,
        cache_logger_on_first_use=False,
    )
    
    logger = structlog.get_logger()
    
    # Make several log calls
    logger.info("First message", step=1)
    logger.warning("Warning message", code="W001")
    logger.error("Error occurred", error="timeout")
    
    # Inspect captured calls
    calls = cap_factory.logger.calls
    
    assert len(calls) == 3
    
    # Check first call
    assert calls[0].method_name == "info"
    assert calls[0].args == ("First message",)
    assert calls[0].kwargs == {"step": 1}
    
    # Check second call
    assert calls[1].method_name == "warning"
    assert calls[1].args == ("Warning message",)
    assert calls[1].kwargs == {"code": "W001"}
    
    # Check third call
    assert calls[2].method_name == "error"
    assert calls[2].args == ("Error occurred",)
    assert calls[2].kwargs == {"error": "timeout"}

import structlog
from structlog.testing import capture_logs

def test_with_capture_logs():
    # Configure structlog normally
    structlog.configure(
        processors=[
            structlog.processors.TimeStamper(),
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
    )
    
    logger = structlog.get_logger()
    
    # Capture logs during test
    with capture_logs() as captured:
        logger.info("Test message", value=42)
        logger.error("Error message", code="E001")
        
        # Assert on captured entries
        assert len(captured) == 2
        
        # Check first entry
        assert captured[0]["event"] == "Test message"
        assert captured[0]["value"] == 42
        assert "timestamp" in captured[0]
        
        # Check second entry
        assert captured[1]["event"] == "Error message"
        assert captured[1]["code"] == "E001"

import structlog
from structlog.testing import capture_logs

def test_bound_logger_context():
    structlog.configure(
        processors=[structlog.processors.JSONRenderer()],
        wrapper_class=structlog.BoundLogger,
    )
    
    base_logger = structlog.get_logger()
    bound_logger = base_logger.bind(user_id=123, session="abc")
    
    with capture_logs() as captured:
        bound_logger.info("User action", action="login")
        
        # Verify context is included
        entry = captured[0]
        assert entry["event"] == "User action"
        assert entry["user_id"] == 123
        assert entry["session"] == "abc"
        assert entry["action"] == "login"

import structlog
from structlog.testing import capture_logs
from structlog import processors

def test_processor_chain():
    structlog.configure(
        processors=[
            processors.TimeStamper(fmt="iso"),
            processors.add_log_level,
            # capture_logs() will intercept before JSONRenderer
            processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
    )
    
    logger = structlog.get_logger()
    
    with capture_logs() as captured:
        logger.warning("Test warning", component="auth")
        
        entry = captured[0]
        
        # Verify processors ran
        assert "timestamp" in entry
        assert entry["level"] == "warning"
        assert entry["event"] == "Test warning"
        assert entry["component"] == "auth"

import structlog
from structlog.testing import capture_logs

def test_exception_logging():
    structlog.configure(
        processors=[
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
    )
    
    logger = structlog.get_logger()
    
    with capture_logs() as captured:
        try:
            raise ValueError("Test exception")
        except ValueError:
            logger.exception("An error occurred", context="testing")
        
        entry = captured[0]
        
        assert entry["event"] == "An error occurred"
        assert entry["context"] == "testing"
        assert "exception" in entry
        assert "ValueError: Test exception" in entry["exception"]

import structlog
from structlog.testing import capture_logs

def add_hostname_processor(logger, method_name, event_dict):
    """Custom processor for testing."""
    event_dict["hostname"] = "test-host"
    return event_dict

def test_custom_processor():
    structlog.configure(
        processors=[
            add_hostname_processor,
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
    )
    
    logger = structlog.get_logger()
    
    with capture_logs() as captured:
        logger.info("Test message")
        
        entry = captured[0]
        
        # Verify custom processor ran
        assert entry["hostname"] == "test-host"
        assert entry["event"] == "Test message"

import pytest
import structlog
from structlog.testing import capture_logs

@pytest.fixture
def logger():
    """Pytest fixture for logger with capture."""
    structlog.configure(
        processors=[
            structlog.processors.TimeStamper(),
            structlog.processors.add_log_level,
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
        cache_logger_on_first_use=False,
    )
    return structlog.get_logger()

def test_user_service_logging(logger):
    """Test logging in user service."""
    with capture_logs() as captured:
        # Simulate user service operations
        logger.info("User service started")
        logger.info("User created", user_id=123, username="alice")
        logger.warning("Duplicate email detected", email="alice@example.com")
        
        # Assertions
        assert len(captured) == 3
        
        start_log = captured[0]
        assert start_log["event"] == "User service started"
        assert start_log["level"] == "info"
        
        create_log = captured[1]
        assert create_log["event"] == "User created"
        assert create_log["user_id"] == 123
        assert create_log["username"] == "alice"
        
        warning_log = captured[2]
        assert warning_log["level"] == "warning"
        assert warning_log["email"] == "alice@example.com"

import structlog
from structlog.testing import ReturnLoggerFactory, CapturingLoggerFactory

def test_configuration_switching():
    """Test switching between different test configurations."""
    
    # Test with ReturnLogger
    structlog.configure(
        processors=[],
        wrapper_class=structlog.BoundLogger,
        logger_factory=ReturnLoggerFactory(),
        cache_logger_on_first_use=False,
    )
    
    logger = structlog.get_logger()
    result = logger.info("test")
    assert result == "test"
    
    # Switch to CapturingLogger
    cap_factory = CapturingLoggerFactory()
    structlog.configure(
        processors=[],
        wrapper_class=structlog.BoundLogger,
        logger_factory=cap_factory,
        cache_logger_on_first_use=False,
    )
    
    logger = structlog.get_logger()
    logger.info("captured")
    
    assert len(cap_factory.logger.calls) == 1
    assert cap_factory.logger.calls[0].method_name == "info"