Added logged module (#17)

The `logged` module exposes a public `Logged` `ABC`, along with `log()`,
`info()`, `warning()`, `error()` and `critical()` top-level functions
for creating concrete `Logged` instances.

The `>>`, `&` and `|` operators are overloaded to allow composing
`Logged` into arbitrary chords.

Also removed the `LevelNo` `NewType`, since it wasn't as useful as I
hoped!

logot/_match.py

@@ -39,8 +39,8 @@
 }
 
 
-def _match_regex(pattern: re.Pattern[str], value: str) -> bool:
-    return pattern.fullmatch(value) is not None
+def _match_regex(pattern: re.Pattern[str], msg: str) -> bool:
+    return pattern.fullmatch(msg) is not None
 
 
 def compile_matcher(pattern: str) -> Matcher:

logot/_util.py

@@ -1,23 +1,19 @@
 from __future__ import annotations
 
 import logging
-from typing import NewType
 
-# An integer log level corresponding to a registered log level.
-LevelNo = NewType("LevelNo", int)
 
-
-def to_levelno(level: int | str) -> LevelNo:
+def to_levelno(level: int | str) -> int:
     # Handle `int` level.
     if isinstance(level, int):
         if logging.getLevelName(level).startswith("Level "):
             raise ValueError(f"Unknown level: {level!r}")
-        return LevelNo(level)
+        return level
     # Handle `str` level.
     if isinstance(level, str):
         levelno = logging.getLevelName(level)
         if not isinstance(levelno, int):
             raise ValueError(f"Unknown level: {level!r}")
-        return LevelNo(levelno)
+        return levelno
     # Fail on other types.
     raise TypeError(f"Invalid level: {level!r}")

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
+
+
+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}"
+
+
+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 _ComposedLogged(Logged):
+    __slots__ = ("_logs",)
+
+    def __init__(self, logs: tuple[Logged, ...]) -> None:
+        assert len(logs) > 1, "Unreachable"
+        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, "Unreachable"
+        # 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}"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "logot"
-version = "0.0.1a3"
+version = "0.0.1a4"
 description = "Log-based testing"
 authors = ["Dave Hall <[email protected]>"]
 license = "MIT"