feat: expose buffered plugin log entries

docs/logging.md

@@ -148,6 +148,42 @@ class ConsumerTest(PactTestMixin, unittest.TestCase):
     def test_something(self) -> None: ...
 ```
 
+## Buffered Plugin Logs
+
+Every plugin log entry is also buffered by the library for the lifetime of the
+process. [`get_logs`][pact.plugins.get_logs] returns the entries for a plugin
+instance, whose ID is the `plugin_instance_id` attribute of the forwarded
+records. The buffered entries include the plugin name, a timestamp and the
+entry's source.
+
+A filter on the `pact.plugins` logger can collect the instance IDs as records
+arrive:
+
+```python
+import logging
+
+from pact import plugins
+
+instance_ids: set[str] = set()
+
+
+class CollectInstanceIds(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        instance_ids.add(record.plugin_instance_id)
+        return True
+
+
+logging.getLogger("pact.plugins").addFilter(CollectInstanceIds())
+
+# Later, once the plugin has been used:
+for instance_id in instance_ids:
+    for entry in plugins.get_logs(instance_id):
+        print(entry.timestamp, entry.level, entry.message)
+```
+
+Call `get_logs` outside the filter. Filters and handlers on `pact.plugins` run
+on the library's own thread, which must not call back into the library.
+
 ## Migrating from `pact_ffi.log_to_stderr`
 
 Earlier releases left the library's logging unconfigured, and test suites
@@ -179,6 +215,7 @@ Under pytest, `--log-cli-level=INFO` shows the records as each test runs.
 ## Further Information
 
 -   [`pact.logging`][pact.logging]
+-   [`pact.plugins`][pact.plugins]
 -   [`pact.testing.pytest`][pact.testing.pytest]
 -   [`pact.testing.unittest`][pact.testing.unittest]
 -   [`pact_ffi.LevelFilter`][pact_ffi.LevelFilter]

src/pact/__init__.py

@@ -125,6 +125,7 @@
 
 if TYPE_CHECKING:
     from pact import logging as logging
+    from pact import plugins as plugins
     from pact import testing as testing
     from pact.pact import Pact
     from pact.verifier import Verifier
@@ -151,6 +152,7 @@
     "Pact": ("pact.pact", "Pact"),
     "Verifier": ("pact.verifier", "Verifier"),
     "logging": ("pact.logging", None),
+    "plugins": ("pact.plugins", None),
     "testing": ("pact.testing", None),
 }
 

src/pact/plugins.py

@@ -0,0 +1,134 @@
+"""
+Plugin log entries buffered by the Pact library.
+
+Pact plugins run as separate processes. Their log entries are forwarded to the
+`pact.plugins` logger by [`pact.logging`][logging], and are also buffered
+by the Pact library for the lifetime of the process, keyed by plugin instance.
+[`get_logs`][plugins.get_logs] returns the buffered entries as
+[`PluginLogEntry`][plugins.PluginLogEntry] objects, which carry the plugin
+name, timestamp and source in addition to the fields available on the
+forwarded log records.
+
+The plugin instance ID is reported on each forwarded record as
+`plugin_instance_id`, and is otherwise not exposed by the library.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any, Literal
+
+import pact_ffi
+from pact.logging import ensure_initialised, from_ffi_level
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+__all__ = ["PluginLogEntry", "get_logs"]
+
+
+@dataclass(frozen=True)
+class PluginLogEntry:
+    """
+    A log entry emitted by a running plugin.
+
+    Entries are returned by [`get_logs`][plugins.get_logs]. Each one is either
+    a line the plugin wrote to its stderr or a structured record it sent over
+    the plugin protocol, as given by `source`.
+    """
+
+    plugin_instance_id: str
+    """UUID assigned by the plugin driver when the plugin instance started."""
+
+    level: str
+    """Log level: one of `TRACE`, `DEBUG`, `INFO`, `WARN` or `ERROR`."""
+
+    message: str
+    """Human-readable log message."""
+
+    test_run_id: str | None = None
+    """
+    Test run ID set with [`set_test_run_id`][logging.set_test_run_id]
+    when the plugin request was made, if any.
+    """
+
+    target: str | None = None
+    """Logger name or module path within the plugin, if known."""
+
+    plugin_name: str | None = None
+    """Plugin name from its manifest."""
+
+    timestamp: datetime | None = None
+    """Time at which the entry was recorded, in UTC."""
+
+    source: Literal["Stderr", "LogRpc"] | None = None
+    """
+    Where the entry originated: a raw line on the plugin's stderr, or a
+    structured record sent over the plugin protocol.
+    """
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any]) -> PluginLogEntry:
+        """
+        Create an entry from the JSON object emitted by the Pact FFI.
+
+        Args:
+            data:
+                A single decoded log entry as returned by
+                [`get_plugin_logs`][pact_ffi.get_plugin_logs].
+
+        Returns:
+            The entry, with `timestamp_ms` converted to a UTC `datetime` and
+            empty optional fields normalised to `None`.
+        """
+        timestamp_ms = data.get("timestamp_ms")
+        return cls(
+            plugin_instance_id=data["plugin_instance_id"],
+            level=data["level"],
+            message=data["message"],
+            test_run_id=data.get("test_run_id") or None,
+            target=data.get("target") or None,
+            plugin_name=data.get("plugin_name") or None,
+            timestamp=(
+                datetime.fromtimestamp(timestamp_ms / 1000, tz=UTC)
+                if timestamp_ms is not None
+                else None
+            ),
+            source=data.get("source"),
+        )
+
+    @property
+    def logging_level(self) -> int:
+        """
+        The entry's level as a standard library `logging` level.
+
+        Returns:
+            The matching `logging` level, as given by
+            [`from_ffi_level`][logging.from_ffi_level].
+        """
+        return from_ffi_level(self.level)
+
+
+def get_logs(plugin_instance_id: str) -> list[PluginLogEntry]:
+    """
+    Return the log entries buffered for a plugin instance.
+
+    Args:
+        plugin_instance_id:
+            The plugin instance whose logs to retrieve, as reported on the
+            `plugin_instance_id` attribute of forwarded log records.
+
+    Returns:
+        The buffered entries in the order they were recorded. An unknown
+        instance ID yields an empty list.
+
+    Raises:
+        RuntimeError:
+            If the Pact library cannot return the buffered entries.
+    """
+    ensure_initialised()
+    return [
+        PluginLogEntry.from_dict(entry)
+        for entry in pact_ffi.get_plugin_logs(plugin_instance_id)
+    ]

tests/test_import.py

@@ -45,6 +45,7 @@ def test_import_does_not_load(module: str) -> None:
         pytest.param("Pact", "pact.pact", "Pact", id="Pact"),
         pytest.param("Verifier", "pact.verifier", "Verifier", id="Verifier"),
         pytest.param("logging", "pact.logging", None, id="logging"),
+        pytest.param("plugins", "pact.plugins", None, id="plugins"),
         pytest.param("testing", "pact.testing", None, id="testing"),
     ],
 )
@@ -69,7 +70,8 @@ def test_dir_lists_lazy_attributes() -> None:
         text=True,
         check=True,
     )
-    assert {"Pact", "Verifier", "logging", "testing"} <= set(result.stdout.split())
+    names = set(result.stdout.split())
+    assert {"Pact", "Verifier", "logging", "plugins", "testing"} <= names
 
 
 def test_state_handler_url_alias() -> None:

tests/test_plugins.py

@@ -0,0 +1,139 @@
+"""
+Unit tests for [`pact.plugins`][pact.plugins].
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import sys
+import time
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+import pytest
+
+from pact import Pact, plugins
+from pact.plugins import PluginLogEntry
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class TestPluginLogEntry:
+    """Tests for [`PluginLogEntry`][pact.plugins.PluginLogEntry]."""
+
+    def test_from_dict(self) -> None:
+        entry = PluginLogEntry.from_dict({
+            "plugin_name": "protobuf",
+            "plugin_instance_id": "abc",
+            "test_run_id": "tests/test_x.py::test_y",
+            "level": "WARN",
+            "message": "something",
+            "target": "plugin::module",
+            "timestamp_ms": 1_700_000_000_000,
+            "source": "LogRpc",
+        })
+        assert entry == PluginLogEntry(
+            plugin_instance_id="abc",
+            level="WARN",
+            message="something",
+            test_run_id="tests/test_x.py::test_y",
+            target="plugin::module",
+            plugin_name="protobuf",
+            timestamp=datetime(2023, 11, 14, 22, 13, 20, tzinfo=UTC),
+            source="LogRpc",
+        )
+
+    def test_from_dict_optional_fields(self) -> None:
+        entry = PluginLogEntry.from_dict({
+            "plugin_instance_id": "abc",
+            "test_run_id": None,
+            "level": "INFO",
+            "message": "something",
+            "target": None,
+        })
+        assert entry.test_run_id is None
+        assert entry.target is None
+        assert entry.plugin_name is None
+        assert entry.timestamp is None
+        assert entry.source is None
+
+    @pytest.mark.parametrize(
+        ("level", "expected"),
+        [
+            pytest.param("TRACE", 5, id="trace"),
+            pytest.param("WARN", logging.WARNING, id="warn"),
+            pytest.param("BOGUS", logging.INFO, id="unknown"),
+        ],
+    )
+    def test_logging_level(self, level: str, expected: int) -> None:
+        entry = PluginLogEntry(plugin_instance_id="abc", level=level, message="")
+        assert entry.logging_level == expected
+
+
+class TestGetLogs:
+    """Tests for [`get_logs`][pact.plugins.get_logs]."""
+
+    def test_unknown_instance(self) -> None:
+        assert plugins.get_logs("does-not-exist") == []
+
+    def test_initialises_logging(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        calls: list[None] = []
+        monkeypatch.setattr(plugins, "ensure_initialised", lambda: calls.append(None))
+        plugins.get_logs("does-not-exist")
+        assert calls == [None]
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="fake plugin is a shell script")
+    def test_buffered_stderr(
+        self,
+        caplog: pytest.LogCaptureFixture,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+    ) -> None:
+        # A plugin which writes to stderr and exits without completing the
+        # handshake. Its instance ID is only reported on the forwarded records.
+        plugin_dir = tmp_path / "fake-0.0.1"
+        plugin_dir.mkdir()
+        (plugin_dir / "pact-plugin.json").write_text(
+            json.dumps({
+                "pluginInterfaceVersion": 1,
+                "name": "fake",
+                "version": "0.0.1",
+                "executableType": "exec",
+                "entryPoint": "fake-plugin",
+            })
+        )
+        entry_point = plugin_dir / "fake-plugin"
+        entry_point.write_text("#!/bin/sh\necho 'fake plugin starting' >&2\nexit 1\n")
+        entry_point.chmod(0o755)
+        monkeypatch.setenv("PACT_PLUGIN_DIR", str(tmp_path))
+        monkeypatch.setenv("PACT_DO_NOT_TRACK", "true")
+
+        pact = Pact("consumer", "provider").with_specification("V4")
+        with pytest.raises(RuntimeError, match="Failed to load the plugin fake"):
+            pact.using_plugin("fake", "0.0.1")
+
+        # The plugin's output is read on a separate thread, which may lag
+        # behind the failed load.
+        forwarded: list[logging.LogRecord] = []
+        deadline = time.monotonic() + 5
+        while not forwarded and time.monotonic() < deadline:
+            forwarded = [
+                r
+                for r in caplog.records
+                if r.name == "pact.plugins" and r.getMessage() == "fake plugin starting"
+            ]
+            time.sleep(0.05)
+        assert len(forwarded) == 1, "the plugin's stderr line was not forwarded"
+        record = forwarded[0]
+        instance_id = record.__dict__["plugin_instance_id"]
+
+        entries = plugins.get_logs(instance_id)
+
+        entry = next(e for e in entries if e.message == "fake plugin starting")
+        assert entry.plugin_instance_id == instance_id
+        assert entry.logging_level == record.levelno
+        assert entry.plugin_name == "fake"
+        assert entry.source == "Stderr"
+        assert entry.timestamp is not None