Type annotations (#27)

* Typing fixes for pytest_structlog

* Move to subfolder so mypy works

* Add log typing to the tests to check the types

* Add mypy checks

* 2.7 is broken

* Explicitly mark and test Python 3.6

* Downgrade Ubuntu version to make 3.6 work

* Upgrade to 3.7 so we can have structlog with .typing

* Covariance arguments fixes for pyright

* Actually install pyright

* Fix pyright comment

* Need structlog 22.2.0 for .typing support

.github/workflows/tests.yml

@@ -10,11 +10,11 @@ on:
 jobs:
   tests:
     name: "Python ${{ matrix.python-version }}"
-    runs-on: "ubuntu-latest"
+    runs-on: "ubuntu-22.04"
 
     strategy:
       matrix:
-        python-version: ["2.7", "3.11"]
+        python-version: ["3.7", "3.11"]
 
     steps:
       - uses: "actions/checkout@v3"
@@ -27,3 +27,19 @@ jobs:
           python -m pip install --editable .
       - name: "Run tests for ${{ matrix.python-version }}"
         run: python -m pytest
+      - name: "Run mypy checks for ${{ matrix.python-version }}"
+        run: |
+          python -m pip install mypy==1.4.1
+          python -m mypy pytest_structlog
+      - name: "Run mypy install checks for ${{ matrix.python-version }}"
+        run: |
+          # This checks that things like the py.typed bits work
+          cd tests
+          python -m pip install ..
+          python -m mypy .
+      - name: "Run pyright testing ${{ matrix.python-version }}"
+        run: |
+          # Pyright and mypy don't always agree, so check with both
+          cd tests
+          python -m pip install pyright==1.1.349
+          python -m pyright .          

MANIFEST.in

@@ -1,2 +1,3 @@
 include LICENSE
 recursive-include tests *.py
+include pytest_structlog/py.typed

README.rst

@@ -55,8 +55,9 @@ Then your test suite might use assertions such as shown below:
 
    # test_your_lib.py
    from your_lib import spline_reticulator
+   import pytest_structlog
 
-   def test_spline_reticulator(log):
+   def test_spline_reticulator(log: pytest_structlog.StructuredLogCapture):
        assert len(log.events) == 0
        spline_reticulator()
        assert len(log.events) == 5

mypy.ini

@@ -0,0 +1,10 @@
+[mypy]
+show_error_codes = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+warn_return_any = true
+warn_unreachable = true
+strict_equality = true

pytest_structlog.py → pytest_structlog/__init__.py

@@ -1,108 +1,110 @@
 import logging
 import os
+from typing import Any, Generator, List, Sequence, Union, cast
 
 import pytest
 import structlog
+from structlog.typing import EventDict, WrappedLogger, Processor
 
 try:
     from structlog.contextvars import merge_contextvars
     from structlog.contextvars import clear_contextvars
 except ImportError:
     # structlog < 20.1.0
     # use a "missing" sentinel to avoid a NameError later on
-    merge_contextvars = object()
+    merge_contextvars = lambda *a, **kw: {}  # noqa
     clear_contextvars = lambda *a, **kw: None  # noqa
 
 
 __version__ = "0.6"
 
 
-class EventList(list):
+class EventList(List[EventDict]):
     """A list subclass that overrides ordering operations.
     Instead of A <= B being a lexicographical comparison,
     now it means every element of A is contained within B,
     in the same order, although there may be other items
     interspersed throughout (i.e. A is a subsequence of B)
     """
 
-    def __ge__(self, other):
+    def __ge__(self, other: Sequence[EventDict]) -> bool:
         return is_subseq(other, self)
 
-    def __gt__(self, other):
+    def __gt__(self, other: Sequence[EventDict]) -> bool:
         return len(self) > len(other) and is_subseq(other, self)
 
-    def __le__(self, other):
+    def __le__(self, other: Sequence[EventDict]) -> bool:
         return is_subseq(self, other)
 
-    def __lt__(self, other):
+    def __lt__(self, other: Sequence[EventDict]) -> bool:
         return len(self) < len(other) and is_subseq(self, other)
 
 
 absent = object()
 
 
-def level_to_name(level):
+def level_to_name(level: Union[str, int]) -> str:
     """Given the name or number for a log-level, return the lower-case level name."""
     if isinstance(level, str):
         return level.lower()
-    return logging.getLevelName(level).lower()
+    return cast(str, logging.getLevelName(level)).lower()
 
 
-def is_submap(d1, d2):
+def is_submap(d1: EventDict, d2: EventDict) -> bool:
     """is every pair from d1 also in d2? (unique and order insensitive)"""
     return all(d2.get(k, absent) == v for k, v in d1.items())
 
 
-def is_subseq(l1, l2):
+def is_subseq(l1: Sequence, l2: Sequence) -> bool:
     """is every element of l1 also in l2? (non-unique and order sensitive)"""
     it = iter(l2)
     return all(d in it for d in l1)
 
 
 class StructuredLogCapture(object):
-    def __init__(self):
+    def __init__(self) -> None:
         self.events = EventList()
 
-    def process(self, logger, method_name, event_dict):
+    def process(self, logger: WrappedLogger, method_name: str, event_dict: EventDict) -> EventDict:
         event_dict["level"] = method_name
         self.events.append(event_dict)
         raise structlog.DropEvent
 
-    def has(self, message, **context):
+    def has(self, message: str, **context: Any) -> bool:
         context["event"] = message
         return any(is_submap(context, e) for e in self.events)
 
-    def log(self, level, event, **kw):
+    def log(self, level: Union[int, str], event: str, **kw: Any) -> dict:
         """Create log event to assert against"""
         return dict(level=level_to_name(level), event=event, **kw)
 
-    def debug(self, event, **kw):
+    def debug(self, event: str, **kw: Any) -> dict:
         """Create debug-level log event to assert against"""
         return self.log(logging.DEBUG, event, **kw)
 
-    def info(self, event, **kw):
+    def info(self, event: str, **kw: Any) -> dict:
         """Create info-level log event to assert against"""
         return self.log(logging.INFO, event, **kw)
 
-    def warning(self, event, **kw):
+    def warning(self, event: str, **kw: Any) -> dict:
         """Create warning-level log event to assert against"""
         return self.log(logging.WARNING, event, **kw)
 
-    def error(self, event, **kw):
+    def error(self, event: str, **kw: Any) -> dict:
         """Create error-level log event to assert against"""
         return self.log(logging.ERROR, event, **kw)
 
-    def critical(self, event, **kw):
+    def critical(self, event: str, **kw: Any) -> dict:
         """Create critical-level log event to assert against"""
         return self.log(logging.CRITICAL, event, **kw)
 
 
-def no_op(*args, **kwargs):
+def no_op(*args: Any, **kwargs: Any) -> None:
     pass
 
 
 @pytest.fixture
-def log(monkeypatch, request):
+def log(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Generator[StructuredLogCapture, None, None]:
     """Fixture providing access to captured structlog events. Interesting attributes:
 
         ``log.events`` a list of dicts, contains any events logged during the test
@@ -115,7 +117,7 @@ def log(monkeypatch, request):
 
     # redirect logging to log capture
     cap = StructuredLogCapture()
-    new_processors = []
+    new_processors: List[Processor] = []
     for processor in original_processors:
         if isinstance(processor, structlog.stdlib.PositionalArgumentsFormatter):
             # if there was a positional argument formatter in there, keep it there
@@ -127,8 +129,8 @@ def log(monkeypatch, request):
             new_processors.append(processor)
     new_processors.append(cap.process)
     structlog.configure(processors=new_processors, cache_logger_on_first_use=False)
-    cap.original_configure = configure = structlog.configure
-    cap.configure_once = structlog.configure_once
+    cap.original_configure = configure = structlog.configure # type:ignore[attr-defined]
+    cap.configure_once = structlog.configure_once # type:ignore[attr-defined]
     monkeypatch.setattr("structlog.configure", no_op)
     monkeypatch.setattr("structlog.configure_once", no_op)
     request.node.structlog_events = cap.events
@@ -141,7 +143,7 @@ def log(monkeypatch, request):
 
 
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
-def pytest_runtest_call(item):
+def pytest_runtest_call(item: pytest.Item) -> Generator[None, None, None]:
     yield
     events = getattr(item, "structlog_events", [])
     content = os.linesep.join([str(e) for e in events])

setup.py

@@ -10,14 +10,13 @@
     author="Wim Glenn",
     author_email="[email protected]",
     license="MIT",
-    install_requires=["pytest", "structlog"],
-    py_modules=["pytest_structlog"],
+    install_requires=["pytest", "structlog>=22.2.0"],
     entry_points={"pytest11": ["pytest-structlog=pytest_structlog"]},
+    python_requires=">=3.7",
     classifiers=[
         "Framework :: Pytest",
         "Programming Language :: Python",
-        "Programming Language :: Python :: 2",
         "Programming Language :: Python :: 3",
     ],
-    options={"bdist_wheel": {"universal": "1"}},
+    include_package_data=True,
 )

tests/test_issue14.py

@@ -1,4 +1,5 @@
 import structlog
+import pytest_structlog
 
 
 logger = structlog.get_logger("some logger")
@@ -18,7 +19,7 @@ def test_first():
     logger.warning("test")
 
 
-def test_second(log):
+def test_second(log: pytest_structlog.StructuredLogCapture):
     logger.warning("test")
     assert log.has("test")
 

tests/test_issue18.py

@@ -2,6 +2,8 @@
 
 import structlog
 
+import pytest_structlog
+
 
 logger = structlog.get_logger(__name__)
 
@@ -24,7 +26,7 @@ def stdlib_configure():
     )
 
 
-def test_positional_formatting(stdlib_configure, log):
+def test_positional_formatting(stdlib_configure, log: pytest_structlog.StructuredLogCapture):
     items_count = 2
     dt = 0.02
     logger.info("Processed %d CC items in total in %.2f seconds", items_count, dt)

tests/test_issue20.py

@@ -1,6 +1,8 @@
 import pytest
 import structlog
 
+import pytest_structlog
+
 
 logger = structlog.get_logger()
 
@@ -22,7 +24,7 @@ def issue20_setup():
     structlog.contextvars.clear_contextvars()
 
 
-def test_contextvar(issue20_setup, log):
+def test_contextvar(issue20_setup, log: pytest_structlog.StructuredLogCapture):
     structlog.contextvars.clear_contextvars()
     logger.info("log1", log1var="value")
     structlog.contextvars.bind_contextvars(contextvar="cv")

tests/test_issue24.py

@@ -3,6 +3,8 @@
 import pytest
 import structlog
 
+import pytest_structlog
+
 logger = structlog.get_logger()
 
 
@@ -23,7 +25,7 @@ def issue24_setup():
 
 
 @pytest.mark.parametrize("n", list(range(RUN_COUNT)))
-def test_contextvar_isolation_in_events(issue24_setup, log, n):
+def test_contextvar_isolation_in_events(issue24_setup, log: pytest_structlog.StructuredLogCapture, n):
     logger.info("without_context")
     structlog.contextvars.bind_contextvars(ctx=n)
     logger.info("with_context")