Added Logot class

_logot_pytest.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from typing import TYPE_CHECKING
+
+import pytest
+
+if TYPE_CHECKING:
+    # Don't import `logot` until the fixture runs.
+    # This avoids problem with pytest coverage reporting.
+    from logot import Logot
+
+
+@pytest.fixture()
+def logot() -> Generator[Logot, None, None]:
+    from logot import Logot
+
+    with Logot().capturing() as logot:
+        yield logot

logot/__init__.py

@@ -1 +1,4 @@
 from __future__ import annotations
+
+from logot._logged import Logged as Logged
+from logot._logot import Logot as Logot

logot/_logged.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+import logging
+from abc import ABC, abstractmethod
+
+from logot._match import compile_matcher
+from logot._util import to_levelno
+
+
+class Logged(ABC):
+    __slots__ = ()
+
+    def __rshift__(self, log: Logged) -> Logged:
+        return _OrderedAllLogged.from_compose(self, log)
+
+    def __and__(self, log: Logged) -> Logged:
+        return _UnorderedAllLogged.from_compose(self, log)
+
+    def __or__(self, log: Logged) -> Logged:
+        return _AnyLogged.from_compose(self, log)
+
+    def __str__(self) -> str:
+        return self._str(indent="")
+
+    @abstractmethod
+    def __eq__(self, other: object) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    def __repr__(self) -> str:
+        raise NotImplementedError
+
+    @abstractmethod
+    def _reduce(self, record: logging.LogRecord) -> Logged | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def _str(self, *, indent: str) -> str:
+        raise NotImplementedError
+
+
+def log(level: int | str, msg: str) -> Logged:
+    return _LogRecordLogged(to_levelno(level), msg)
+
+
+def debug(msg: str) -> Logged:
+    return _LogRecordLogged(logging.DEBUG, msg)
+
+
+def info(msg: str) -> Logged:
+    return _LogRecordLogged(logging.INFO, msg)
+
+
+def warning(msg: str) -> Logged:
+    return _LogRecordLogged(logging.WARNING, msg)
+
+
+def error(msg: str) -> Logged:
+    return _LogRecordLogged(logging.ERROR, msg)
+
+
+def critical(msg: str) -> Logged:
+    return _LogRecordLogged(logging.CRITICAL, msg)
+
+
+class _LogRecordLogged(Logged):
+    __slots__ = ("_levelno", "_msg", "_matcher")
+
+    def __init__(self, levelno: int, msg: str) -> None:
+        self._levelno = levelno
+        self._msg = msg
+        self._matcher = compile_matcher(msg)
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, _LogRecordLogged) and other._levelno == self._levelno and other._msg == self._msg
+
+    def __repr__(self) -> str:
+        return f"log({logging.getLevelName(self._levelno)!r}, {self._msg!r})"
+
+    def _reduce(self, record: logging.LogRecord) -> Logged | None:
+        if self._levelno == record.levelno and self._matcher(record.getMessage()):
+            return None
+        return self
+
+    def _str(self, *, indent: str) -> str:
+        return f"[{logging.getLevelName(self._levelno)}] {self._msg}"
+
+
+class _ComposedLogged(Logged):
+    __slots__ = ("_logs",)
+
+    def __init__(self, logs: tuple[Logged, ...]) -> None:
+        assert len(logs) > 1
+        self._logs = logs
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, self.__class__) and other._logs == self._logs
+
+    @classmethod
+    def from_compose(cls, log_a: Logged, log_b: Logged) -> Logged:
+        # If possible, flatten nested logs of the same type.
+        if isinstance(log_a, cls):
+            if isinstance(log_b, cls):
+                return cls((*log_a._logs, *log_b._logs))
+            return cls((*log_a._logs, log_b))
+        if isinstance(log_b, cls):
+            return cls((log_a, *log_b._logs))
+        # Wrap the logs without flattening.
+        return cls((log_a, log_b))
+
+    @classmethod
+    def from_reduce(cls, logs: tuple[Logged, ...]) -> Logged | None:
+        assert logs
+        # If there is a single log, do not wrap it.
+        if len(logs) == 1:
+            return logs[0]
+        # Wrap the logs.
+        return cls(logs)
+
+
+class _OrderedAllLogged(_ComposedLogged):
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return f"({' >> '.join(map(repr, self._logs))})"
+
+    def _reduce(self, record: logging.LogRecord) -> Logged | None:
+        log = self._logs[0]
+        reduced_log = log._reduce(record)
+        # Handle full reduction.
+        if reduced_log is None:
+            return _OrderedAllLogged.from_reduce(self._logs[1:])
+        # Handle partial reduction.
+        if reduced_log is not log:
+            return _OrderedAllLogged((reduced_log, *self._logs[1:]))
+        # Handle no reduction.
+        return self
+
+    def _str(self, *, indent: str) -> str:
+        return f"\n{indent}".join(log._str(indent=indent) for log in self._logs)
+
+
+class _UnorderedAllLogged(_ComposedLogged):
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return f"({' & '.join(map(repr, self._logs))})"
+
+    def _reduce(self, record: logging.LogRecord) -> Logged | None:
+        for n, log in enumerate(self._logs):
+            reduced_log = log._reduce(record)
+            # Handle full reduction.
+            if reduced_log is None:
+                return _UnorderedAllLogged.from_reduce((*self._logs[:n], *self._logs[n + 1 :]))
+            # Handle partial reduction.
+            if reduced_log is not log:
+                return _UnorderedAllLogged((*self._logs[:n], reduced_log, *self._logs[n + 1 :]))
+        # Handle no reduction.
+        return self
+
+    def _str(self, *, indent: str) -> str:
+        nested_indent = indent + "  "
+        logs_str = "".join(f"\n{indent}- {log._str(indent=nested_indent)}" for log in self._logs)
+        return f"Unordered:{logs_str}"
+
+
+class _AnyLogged(_ComposedLogged):
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return f"({' | '.join(map(repr, self._logs))})"
+
+    def _reduce(self, record: logging.LogRecord) -> Logged | None:
+        for n, log in enumerate(self._logs):
+            reduced_log = log._reduce(record)
+            # Handle full reduction.
+            if reduced_log is None:
+                return None
+            # Handle partial reduction.
+            if reduced_log is not log:
+                return _AnyLogged((*self._logs[:n], reduced_log, *self._logs[n + 1 :]))
+        # Handle no reduction.
+        return self
+
+    def _str(self, *, indent: str) -> str:
+        nested_indent = indent + "  "
+        logs_str = "".join(f"\n{indent}- {log._str(indent=nested_indent)}" for log in self._logs)
+        return f"Any:{logs_str}"

logot/_logot.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import logging
+from collections import deque
+from contextlib import AbstractContextManager
+from threading import Lock
+from types import TracebackType
+from typing import ClassVar, TypeVar
+from weakref import WeakValueDictionary
+
+from logot._logged import Logged
+from logot._util import to_levelno, to_logger, to_timeout
+from logot._waiter import AsyncWaiter, SyncWaiter, Waiter
+
+W = TypeVar("W", bound=Waiter)
+
+
+class Logot:
+    __slots__ = ("_timeout", "_lock", "_seen_records", "_queue", "_waiter")
+
+    DEFAULT_LEVEL: ClassVar[int | str] = logging.NOTSET
+    DEFAULT_LOGGER: ClassVar[logging.Logger | str | None] = None
+    DEFAULT_TIMEOUT: ClassVar[float] = 3.0
+
+    def __init__(
+        self,
+        *,
+        timeout: float = DEFAULT_TIMEOUT,
+    ) -> None:
+        self._timeout = to_timeout(timeout)
+        self._lock = Lock()
+        self._seen_records: WeakValueDictionary[int, logging.LogRecord] = WeakValueDictionary()
+        self._queue: deque[logging.LogRecord] = deque()
+        self._waiter: Waiter | None = None
+
+    def capturing(
+        self,
+        *,
+        level: int | str = DEFAULT_LEVEL,
+        logger: logging.Logger | str | None = DEFAULT_LOGGER,
+    ) -> AbstractContextManager[Logot]:
+        levelno = to_levelno(level)
+        logger = to_logger(logger)
+        return _Capturing(self, _Handler(self, levelno=levelno), logger=logger)
+
+    def assert_logged(self, log: Logged) -> None:
+        reduced_log = self._reduce(log)
+        if reduced_log is not None:
+            raise AssertionError(f"Not logged:\n\n{reduced_log}")
+
+    def assert_not_logged(self, log: Logged) -> None:
+        reduced_log = self._reduce(log)
+        if reduced_log is None:
+            raise AssertionError(f"Logged:\n\n{log}")
+
+    def wait_for(self, log: Logged, *, timeout: float | None = None) -> None:
+        waiter = self._open_waiter(log, SyncWaiter, timeout=timeout)
+        try:
+            waiter.wait()
+        finally:
+            self._close_waiter(waiter)
+
+    async def await_for(self, log: Logged, *, timeout: float | None = None) -> None:
+        waiter = self._open_waiter(log, AsyncWaiter, timeout=timeout)
+        try:
+            await waiter.wait()
+        finally:
+            self._close_waiter(waiter)
+
+    def _emit(self, record: logging.LogRecord) -> None:
+        with self._lock:
+            # De-duplicate log records.
+            # Duplicate log records are possible if we have multiple active captures.
+            record_id = id(record)
+            if record_id in self._seen_records:  # pragma: no cover
+                return
+            self._seen_records[record_id] = record
+            # If there is a waiter that has not been fully reduced, attempt to reduce it.
+            if self._waiter is not None and self._waiter.log is not None:
+                self._waiter.log = self._waiter.log._reduce(record)
+                # If the waiter has fully reduced, notify the blocked caller.
+                if self._waiter.log is None:
+                    self._waiter.notify()
+                return
+            # Otherwise, buffer the log record.
+            self._queue.append(record)
+
+    def _reduce(self, log: Logged | None) -> Logged | None:
+        # Drain the queue until the log is fully reduced.
+        # This does not need a lock, since `deque.popleft()` is thread-safe.
+        while self._queue and log is not None:
+            log = log._reduce(self._queue.popleft())
+        # All done!
+        return log
+
+    def _open_waiter(self, log: Logged, waiter_cls: type[W], *, timeout: float | None) -> W:
+        with self._lock:
+            # If no timeout is provided, use the default timeout.
+            # Otherwise, validate and use the provided timeout.
+            if timeout is None:
+                timeout = self._timeout
+            else:
+                timeout = to_timeout(timeout)
+            # Ensure no other waiters.
+            if self._waiter is not None:  # pragma: no cover
+                raise RuntimeError("Multiple waiters are not supported")
+            # Set a waiter.
+            waiter = self._waiter = waiter_cls(log, timeout=timeout)
+            # Apply an immediate reduction.
+            waiter.log = self._reduce(waiter.log)
+            if waiter.log is None:
+                waiter.notify()
+            # All done!
+            return waiter
+
+    def _close_waiter(self, waiter: Waiter) -> None:
+        with self._lock:
+            # Clear the waiter.
+            self._waiter = None
+            # Error if the waiter logs are not fully reduced.
+            if waiter.log is not None:
+                raise AssertionError(f"Not logged:\n\n{waiter.log}")
+
+
+class _Capturing:
+    __slots__ = ("_logot", "_handler", "_logger", "_prev_levelno")
+
+    def __init__(self, logot: Logot, handler: logging.Handler, *, logger: logging.Logger) -> None:
+        self._logot = logot
+        self._handler = handler
+        self._logger = logger
+
+    def __enter__(self) -> Logot:
+        # If the logger is less verbose than the handler, force it to the necessary verboseness.
+        self._prev_levelno = self._logger.level
+        if self._handler.level < self._logger.level:
+            self._logger.setLevel(self._handler.level)
+        # Add the handler.
+        self._logger.addHandler(self._handler)
+        return self._logot
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        # Remove the handler.
+        self._logger.removeHandler(self._handler)
+        # Restore the previous level.
+        self._logger.setLevel(self._prev_levelno)
+
+
+class _Handler(logging.Handler):
+    __slots__ = ("_logot",)
+
+    def __init__(self, logot: Logot, *, levelno: int) -> None:
+        super().__init__(levelno)
+        self._logot = logot
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self._logot._emit(record)