fix: add pytest plugin for caplog integration

logprise/pytest_plugin.py

@@ -0,0 +1,111 @@
+"""Pytest plugin for logprise/caplog integration.
+
+This plugin ensures that logs emitted via logprise (which uses loguru internally)
+are captured by pytest's caplog fixture.
+
+The plugin is automatically loaded by pytest when logprise is installed,
+thanks to the pytest11 entry point defined in pyproject.toml.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+import pytest
+from loguru import logger
+
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    from _pytest.logging import LogCaptureFixture
+
+
+class _CaplogState:
+    """Container for caplog fixture state to avoid module-level globals."""
+
+    fixture: LogCaptureFixture | None = None
+
+
+_state = _CaplogState()
+
+
+def _create_log_record(record: dict) -> logging.LogRecord:
+    """Create a standard logging.LogRecord from a loguru record dict."""
+    # Get the level info
+    level_no = record["level"].no
+    level_name = record["level"].name
+
+    # Get module/function info
+    module = record.get("module", "")
+    func_name = record.get("function", "")
+    line_no = record.get("line", 0)
+    file_path = record.get("file", None)
+    pathname = str(file_path) if file_path else ""
+
+    # Create a LogRecord directly
+    log_record = logging.LogRecord(
+        name=record.get("name") or module or "logprise",
+        level=level_no,
+        pathname=pathname,
+        lineno=line_no,
+        msg=record["message"],
+        args=(),
+        exc_info=record["exception"],
+        func=func_name,
+    )
+
+    # Set the level name explicitly
+    log_record.levelname = level_name
+
+    return log_record
+
+
+def _loguru_to_caplog(message: object) -> None:
+    """Sink function that forwards loguru messages directly to caplog's handler.
+
+    This function bypasses the standard logging system entirely to avoid
+    the recursion caused by logprise's InterceptHandler.
+    """
+    if _state.fixture is None:
+        return
+
+    # Get the record dict from the message
+    record = message.record  # type: ignore[union-attr]
+
+    # Create a standard LogRecord
+    log_record = _create_log_record(record)
+
+    # Check if the record's level meets the caplog's handler level threshold
+    # This respects caplog.at_level() context manager
+    if log_record.levelno >= _state.fixture.handler.level:
+        _state.fixture.handler.emit(log_record)
+
+
+@pytest.fixture
+def caplog(caplog: LogCaptureFixture) -> Generator[LogCaptureFixture, None, None]:
+    """Enhanced caplog fixture that captures logprise/loguru logs.
+
+    This fixture wraps pytest's built-in caplog fixture and adds a loguru
+    sink that forwards logs directly to caplog's handler, bypassing
+    the standard logging system to avoid recursion with logprise's
+    InterceptHandler.
+    """
+    # Store reference to caplog fixture
+    _state.fixture = caplog
+
+    # Add a loguru sink that writes directly to caplog
+    handler_id = logger.add(
+        _loguru_to_caplog,
+        format="{message}",
+        level=0,  # Capture all levels, let caplog filter
+        catch=False,
+    )
+
+    try:
+        yield caplog
+    finally:
+        # Remove the sink and clear the fixture reference
+        logger.remove(handler_id)
+        _state.fixture = None

pyproject.toml

@@ -22,6 +22,9 @@ pytest-mock = "*"
 ruff = "*"
 mypy = "*"
 
+[tool.poetry.plugins."pytest11"]
+logprise = "logprise.pytest_plugin"
+
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]
 build-backend = "poetry.core.masonry.api"

tests/test_caplog_integration.py

@@ -0,0 +1,108 @@
+"""TDD tests for caplog integration.
+
+These tests verify that logprise logs show up in pytest's caplog fixture.
+"""
+
+from __future__ import annotations
+
+import logging
+
+
+def test_loguru_logger_captured_in_caplog(caplog):
+    """Test that logs from loguru's logger show up in caplog."""
+    from logprise import logger
+
+    with caplog.at_level(logging.INFO):
+        logger.info("Test message from loguru")
+
+    assert len(caplog.records) >= 1
+    assert any("Test message from loguru" in record.message for record in caplog.records)
+
+
+def test_standard_logging_captured_in_caplog(caplog):
+    """Test that standard logging (intercepted by logprise) shows up in caplog."""
+    with caplog.at_level(logging.WARNING):
+        logging.warning("Test warning from standard logging")
+
+    assert len(caplog.records) >= 1
+    assert any("Test warning from standard logging" in record.message for record in caplog.records)
+
+
+def test_loguru_different_levels_captured(caplog):
+    """Test that different log levels are captured correctly."""
+    from logprise import logger
+
+    with caplog.at_level(logging.DEBUG):
+        logger.debug("Debug message")
+        logger.info("Info message")
+        logger.warning("Warning message")
+        logger.error("Error message")
+
+    messages = [record.message for record in caplog.records]
+    assert any("Debug message" in msg for msg in messages)
+    assert any("Info message" in msg for msg in messages)
+    assert any("Warning message" in msg for msg in messages)
+    assert any("Error message" in msg for msg in messages)
+
+
+def test_caplog_level_filtering(caplog):
+    """Test that caplog level filtering works with logprise."""
+    from logprise import logger
+
+    with caplog.at_level(logging.WARNING):
+        logger.info("Should not appear")
+        logger.warning("Should appear")
+
+    messages = [record.message for record in caplog.records]
+    assert not any("Should not appear" in msg for msg in messages)
+    assert any("Should appear" in msg for msg in messages)
+
+
+def test_caplog_records_have_correct_level(caplog):
+    """Test that captured records have the correct log level."""
+    from logprise import logger
+
+    with caplog.at_level(logging.DEBUG):
+        logger.warning("Warning test")
+        logger.error("Error test")
+
+    warning_records = [r for r in caplog.records if "Warning test" in r.message]
+    error_records = [r for r in caplog.records if "Error test" in r.message]
+
+    assert len(warning_records) >= 1
+    assert len(error_records) >= 1
+    assert warning_records[0].levelno == logging.WARNING
+    assert error_records[0].levelno == logging.ERROR
+
+
+def test_caplog_clear_works(caplog):
+    """Test that caplog.clear() works properly."""
+    from logprise import logger
+
+    with caplog.at_level(logging.INFO):
+        logger.info("First message")
+        assert len(caplog.records) >= 1
+
+        caplog.clear()
+        assert len(caplog.records) == 0
+
+        logger.info("Second message")
+        assert len(caplog.records) >= 1
+        messages = [r.message for r in caplog.records]
+        assert not any("First message" in msg for msg in messages)
+        assert any("Second message" in msg for msg in messages)
+
+
+def test_multiple_loggers_captured(caplog):
+    """Test that logs from multiple loggers are captured."""
+    from logprise import logger
+
+    named_logger = logging.getLogger("test.named.logger")
+
+    with caplog.at_level(logging.INFO):
+        logger.info("From loguru")
+        named_logger.info("From named logger")
+
+    messages = [record.message for record in caplog.records]
+    assert any("From loguru" in msg for msg in messages)
+    assert any("From named logger" in msg for msg in messages)