From e751d6a1d351efa1fee5e8c96b6f4621962d012f Mon Sep 17 00:00:00 2001 From: JP-Ellis Date: Thu, 17 Sep 2026 20:03:05 +1000 Subject: [PATCH] feat: add plugin observability module Pact FFI 0.5.5 lets a host correlate plugin activity with a test and receive plugin log entries. `pact.plugins` exposes this in the main library: - `set_test_run_id` tags plugin requests from the current thread. - `register_log_callback` delivers each entry as a `PluginLogEntry`. - `forward_to_logging` emits entries through the standard `logging` module, with the instance ID, test run ID and plugin target attached to each record. - `get_logs` returns the entries buffered for a plugin instance, with the timestamp converted to a UTC `datetime`. A single CFFI trampoline is registered with the library and dispatches to whichever Python callable is current. The library ignores a NULL registration, so this is what makes `register_log_callback(None)` stop delivery. The FFI only installs its plugin log sink from `pactffi_init` and `pactffi_init_with_log_level`, so the module and the logging docs direct users to `pact_ffi.init_with_log_level` in place of `log_to_stderr`. Signed-off-by: JP-Ellis Assisted-by: Claude Code:claude-opus-5 --- docs/logging.md | 65 ++++++++++ src/pact/__init__.py | 2 + src/pact/plugins.py | 289 ++++++++++++++++++++++++++++++++++++++++++ tests/test_plugins.py | 204 +++++++++++++++++++++++++++++ 4 files changed, 560 insertions(+) create mode 100644 src/pact/plugins.py create mode 100644 tests/test_plugins.py diff --git a/docs/logging.md b/docs/logging.md index 6fb46f28b..ae714c988 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -96,6 +96,70 @@ The functions `logger_init`, `logger_attach_sink`, and `logger_apply` are curren For the most advanced scenarios, the FFI supports configuring multiple log sinks simultaneously (e.g., logging to both stderr and a file). This requires using the lower-level `logger_init`, `logger_attach_sink`, and `logger_apply` functions, which are planned for future implementation. +## Plugin Logs + +Pact plugins (such as the protobuf and gRPC plugins) run as separate processes, so their log output is not part of the FFI logging configured above. The [`pact.plugins`][pact.plugins] module exposes the plugin observability features of the Pact FFI. + +/// warning | Initialisation +The FFI only captures plugin log entries once it has been initialised with [`init_with_log_level`][pact_ffi.init_with_log_level] (or [`init`][pact_ffi.init]). These initialise the FFI logger to stderr in the same way as `log_to_stderr`, so use one in its place; calling both results in the "Logger already initialized" error. +/// + +### Forwarding to `logging` + +[`forward_to_logging`][pact.plugins.forward_to_logging] delivers every plugin log entry through the standard library `logging` module, at the equivalent level, via the `pact.plugins` logger by default: + +```python +import pact_ffi +from pact import plugins + +pact_ffi.init_with_log_level("INFO") +plugins.forward_to_logging() +``` + +The plugin instance ID, test run ID and plugin-side logger target are attached to each log record as `plugin_instance_id`, `test_run_id` and `plugin_target`, so they can be included in a formatter: + +```python +logging.basicConfig( + format="%(levelname)s %(name)s [%(plugin_instance_id)s] %(message)s", +) +``` + +Unlike the FFI logger, the plugin log callback can be replaced at any time. [`register_log_callback`][pact.plugins.register_log_callback] accepts any callable taking a [`PluginLogEntry`][pact.plugins.PluginLogEntry] for custom handling. + +### Correlating Logs with Tests + +Plugins serve every test in the process, so their log entries do not identify the test which triggered them. [`set_test_run_id`][pact.plugins.set_test_run_id] tags requests made from the current thread with an identifier which the plugin reports back on each log entry. Using the pytest node ID is the natural choice: + +```python +import pytest + +import pact_ffi +from pact import plugins + + +@pytest.fixture(autouse=True, scope="session") +def plugin_logging(): + pact_ffi.init_with_log_level("INFO") + plugins.forward_to_logging() + + +@pytest.fixture(autouse=True) +def plugin_test_run_id(request): + plugins.set_test_run_id(request.node.nodeid) + yield request.node.nodeid + plugins.set_test_run_id(None) +``` + +### Retrieving Buffered Logs + +Every plugin log entry is also buffered by the Pact library for the lifetime of the process. [`get_logs`][pact.plugins.get_logs] returns the entries for a plugin instance, whose ID is reported on each [`PluginLogEntry`][pact.plugins.PluginLogEntry] received through the callback: + +```python +entries = plugins.get_logs(plugin_instance_id) +for entry in entries: + print(entry.timestamp, entry.level, entry.message) +``` + ## Troubleshooting ### "Logger already initialized" Error @@ -122,3 +186,4 @@ For complete API documentation, see: - [`pact_ffi.log_to_file`][pact_ffi.log_to_file] - [`pact_ffi.log_to_buffer`][pact_ffi.log_to_buffer] - [`pact_ffi.LevelFilter`][pact_ffi.LevelFilter] +- [`pact.plugins`][pact.plugins] diff --git a/src/pact/__init__.py b/src/pact/__init__.py index a84f9ab38..ab4ace038 100644 --- a/src/pact/__init__.py +++ b/src/pact/__init__.py @@ -111,6 +111,7 @@ from __future__ import annotations +from pact import plugins as plugins from pact import xml as xml from pact.__version__ import __version__, __version_tuple__ from pact.pact import Pact @@ -125,5 +126,6 @@ "Verifier", "__version__", "__version_tuple__", + "plugins", "xml", ] diff --git a/src/pact/plugins.py b/src/pact/plugins.py new file mode 100644 index 000000000..0328e709d --- /dev/null +++ b/src/pact/plugins.py @@ -0,0 +1,289 @@ +""" +Plugin observability for Pact. + +Pact plugins run as separate processes, so their log output is not visible to +the Python test process by default. This module exposes the mechanisms the Pact +FFI provides to correlate plugin activity with a test and to retrieve plugin +logs: + +- [`set_test_run_id`][plugins.set_test_run_id] tags requests sent to + plugins from the current thread with an identifier, so that plugin log + entries can be traced back to the test which caused them. +- [`register_log_callback`][plugins.register_log_callback] invokes a + Python callable for every log entry emitted by any running plugin. +- [`forward_to_logging`][plugins.forward_to_logging] is a ready-made + callback which emits plugin log entries through the standard library + `logging` module. +- [`get_logs`][plugins.get_logs] returns the log entries buffered for a + plugin instance, whether or not a callback was registered. + +The Pact FFI only captures plugin log entries once it has been initialised +with [`init_with_log_level`][pact_ffi.init_with_log_level] or +[`init`][pact_ffi.init]. These also configure the FFI's own logging, in place of +[`log_to_stderr`][pact_ffi.log_to_stderr]. + +A typical pytest setup initialises the FFI and forwards plugin logs to +`logging` once per session, and tags each test with its node ID: + +```python +import pytest + +import pact_ffi +from pact import plugins + + +@pytest.fixture(autouse=True, scope="session") +def plugin_logging(): + pact_ffi.init_with_log_level("INFO") + plugins.forward_to_logging() + + +@pytest.fixture(autouse=True) +def plugin_test_run_id(request): + plugins.set_test_run_id(request.node.nodeid) + yield request.node.nodeid + plugins.set_test_run_id(None) +``` +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any, Literal + +import pact_ffi + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + +logger = logging.getLogger(__name__) + +# The FFI's callback is registered once and dispatches to whichever Python +# callable is current, so that deregistration works even though the underlying +# library ignores a NULL registration. +_callback: Callable[[PluginLogEntry], None] | None = None +_trampoline_registered = False + +_LEVELS: Mapping[str, int] = { + "TRACE": logging.DEBUG, + "DEBUG": logging.DEBUG, + "INFO": logging.INFO, + "WARN": logging.WARNING, + "WARNING": logging.WARNING, + "ERROR": logging.ERROR, +} + + +@dataclass(frozen=True) +class PluginLogEntry: + """ + A log entry emitted by a running plugin. + + Entries delivered through + [`register_log_callback`][plugins.register_log_callback] carry only the + fields the FFI passes to the callback; `plugin_name`, `timestamp` and + `source` are populated only for entries retrieved with + [`get_logs`][plugins.get_logs]. + """ + + 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`][plugins.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=timezone.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 constant. Unknown levels map to + `logging.INFO`. + """ + return _LEVELS.get(self.level.upper(), logging.INFO) + + +def set_test_run_id(test_run_id: str | None) -> None: + """ + Set the test run ID for the current thread. + + The ID is attached to every request sent to a plugin from the current + thread, and is reported back in the `test_run_id` of the plugin's log + entries. Using the test's identifier (such as pytest's `request.node.nodeid`) + makes it possible to attribute plugin logs to the test which triggered them. + + The ID is thread-local, so it must be set on the thread which drives the + interaction or verification. + + Args: + test_run_id: + The identifier to attach, or `None` to clear a previously set ID. + """ + pact_ffi.set_test_run_id(test_run_id) + + +def _dispatch( + plugin_instance_id: str, + test_run_id: str, + level: str, + target: str, + message: str, +) -> None: + if _callback is None: + return + _callback( + PluginLogEntry( + plugin_instance_id=plugin_instance_id, + level=level, + message=message, + test_run_id=test_run_id or None, + target=target or None, + ) + ) + + +def register_log_callback( + callback: Callable[[PluginLogEntry], None] | None, +) -> None: + """ + Register a callback invoked for every plugin log entry. + + The callback receives a [`PluginLogEntry`][plugins.PluginLogEntry] for + each entry emitted by any running plugin. Only one callback is active at a + time; registering a new one replaces the previous one. + + The callback runs on a thread owned by the Pact library, so it must be + thread-safe and must return promptly. It must not call into Pact itself. + Exceptions raised by the callback are reported by CFFI and otherwise + ignored. + + Args: + callback: + The callable to invoke, or `None` to stop receiving entries. + """ + global _callback, _trampoline_registered # noqa: PLW0603 + + _callback = callback + if callback is not None and not _trampoline_registered: + pact_ffi.register_plugin_log_callback(_dispatch) + _trampoline_registered = True + + +def forward_to_logging(target: logging.Logger | str | None = None) -> None: + """ + Forward plugin log entries to the standard library `logging` module. + + Each entry is emitted at the equivalent `logging` level, with the plugin + instance ID, test run ID and target available on the log record as + `plugin_instance_id`, `test_run_id` and `plugin_target` for use in + formatters and filters. + + This registers a callback with + [`register_log_callback`][plugins.register_log_callback], and so + replaces any callback registered previously. + + Args: + target: + The logger, or logger name, to emit entries through. Defaults to + the `pact.plugins` logger. + """ + if target is None: + destination = logger + elif isinstance(target, str): + destination = logging.getLogger(target) + else: + destination = target + + def _emit(entry: PluginLogEntry) -> None: + destination.log( + entry.logging_level, + "%s", + entry.message, + extra={ + "plugin_instance_id": entry.plugin_instance_id, + "test_run_id": entry.test_run_id, + "plugin_target": entry.target, + }, + ) + + register_log_callback(_emit) + + +def get_logs(plugin_instance_id: str) -> list[PluginLogEntry]: + """ + Return the log entries buffered for a plugin instance. + + The Pact library buffers every plugin log entry for the lifetime of the + process, independently of any registered callback. The plugin instance ID + is reported on each [`PluginLogEntry`][plugins.PluginLogEntry], and is + otherwise not exposed by the library. + + Args: + plugin_instance_id: + The plugin instance whose logs to retrieve. + + Returns: + The buffered entries in the order they were recorded. An unknown + instance ID yields an empty list. + """ + return [ + PluginLogEntry.from_dict(entry) + for entry in pact_ffi.get_plugin_logs(plugin_instance_id) + ] diff --git a/tests/test_plugins.py b/tests/test_plugins.py new file mode 100644 index 000000000..3e8dfe9bd --- /dev/null +++ b/tests/test_plugins.py @@ -0,0 +1,204 @@ +""" +Unit tests for the [`pact.plugins`][pact.plugins] observability helpers. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +import pytest + +import pact_ffi +from pact import plugins +from pact.plugins import PluginLogEntry + +if TYPE_CHECKING: + from collections.abc import Generator + + +def emit( + plugin_instance_id: str = "instance", + test_run_id: str = "", + level: str = "INFO", + target: str = "", + message: str = "hello", +) -> None: + """ + Invoke the callback registered with the FFI as the library would. + """ + callback = pact_ffi._plugin_log_callback # noqa: SLF001 + assert callback is not None + callback( + plugin_instance_id.encode("utf-8"), + test_run_id.encode("utf-8"), + level.encode("utf-8"), + target.encode("utf-8"), + message.encode("utf-8"), + ) + + +@pytest.fixture(autouse=True) +def _reset_callback() -> Generator[None, None, None]: + yield + plugins.register_log_callback(None) + + +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=timezone.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", logging.DEBUG, id="trace"), + pytest.param("DEBUG", logging.DEBUG, id="debug"), + pytest.param("INFO", logging.INFO, id="info"), + pytest.param("WARN", logging.WARNING, id="warn"), + pytest.param("ERROR", logging.ERROR, id="error"), + pytest.param("warn", logging.WARNING, id="lowercase"), + 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 TestSetTestRunId: + """Tests for [`set_test_run_id`][pact.plugins.set_test_run_id].""" + + def test_set_and_clear(self) -> None: + plugins.set_test_run_id("tests/test_x.py::test_y") + plugins.set_test_run_id(None) + + +class TestRegisterLogCallback: + """Tests for [`register_log_callback`][pact.plugins.register_log_callback].""" + + def test_callback_receives_entry(self) -> None: + received: list[PluginLogEntry] = [] + plugins.register_log_callback(received.append) + + emit("inst", "run-1", "DEBUG", "mod", "hello") + + assert received == [ + PluginLogEntry( + plugin_instance_id="inst", + level="DEBUG", + message="hello", + test_run_id="run-1", + target="mod", + ) + ] + + def test_empty_strings_become_none(self) -> None: + received: list[PluginLogEntry] = [] + plugins.register_log_callback(received.append) + + emit(test_run_id="", target="") + + assert received[0].test_run_id is None + assert received[0].target is None + + def test_replaces_previous_callback(self) -> None: + first: list[PluginLogEntry] = [] + second: list[PluginLogEntry] = [] + plugins.register_log_callback(first.append) + plugins.register_log_callback(second.append) + + emit() + + assert first == [] + assert len(second) == 1 + + def test_none_deregisters(self) -> None: + received: list[PluginLogEntry] = [] + plugins.register_log_callback(received.append) + plugins.register_log_callback(None) + + emit() + + assert received == [] + + +class TestForwardToLogging: + """Tests for [`forward_to_logging`][pact.plugins.forward_to_logging].""" + + def test_default_logger(self, caplog: pytest.LogCaptureFixture) -> None: + plugins.forward_to_logging() + + with caplog.at_level(logging.DEBUG, logger="pact.plugins"): + emit("inst", "run-1", "WARN", "mod", "careful") + + assert len(caplog.records) == 1 + record = caplog.records[0] + assert record.name == "pact.plugins" + assert record.levelno == logging.WARNING + assert record.getMessage() == "careful" + assert record.__dict__["plugin_instance_id"] == "inst" + assert record.__dict__["test_run_id"] == "run-1" + assert record.__dict__["plugin_target"] == "mod" + + def test_named_logger(self, caplog: pytest.LogCaptureFixture) -> None: + plugins.forward_to_logging("my.plugins") + + with caplog.at_level(logging.DEBUG, logger="my.plugins"): + emit(level="ERROR", message="boom") + + assert [(r.name, r.levelno) for r in caplog.records] == [ + ("my.plugins", logging.ERROR) + ] + + def test_logger_instance(self, caplog: pytest.LogCaptureFixture) -> None: + plugins.forward_to_logging(logging.getLogger("other")) + + with caplog.at_level(logging.DEBUG, logger="other"): + emit(level="TRACE", message="detail") + + assert [(r.name, r.levelno) for r in caplog.records] == [ + ("other", logging.DEBUG) + ] + + +class TestGetLogs: + """Tests for [`get_logs`][pact.plugins.get_logs].""" + + def test_unknown_instance(self) -> None: + assert plugins.get_logs("does-not-exist") == []