From c843e96029a5706b9acb3d4cee6b49369fa69593 Mon Sep 17 00:00:00 2001 From: mikegit123git Date: Sun, 1 Nov 2026 15:21:58 +0800 Subject: [PATCH] fix: undo hookcall monitors independently --- changelog/815.bugfix.rst | 1 + src/pluggy/_manager.py | 54 +++++++--- testing/test_monitor_undo.py | 189 +++++++++++++++++++++++++++++++++++ 3 files changed, 229 insertions(+), 15 deletions(-) create mode 100644 changelog/815.bugfix.rst create mode 100644 testing/test_monitor_undo.py diff --git a/changelog/815.bugfix.rst b/changelog/815.bugfix.rst new file mode 100644 index 00000000..fdda0ef9 --- /dev/null +++ b/changelog/815.bugfix.rst @@ -0,0 +1 @@ +Fix hook-call monitoring so undo functions work regardless of registration order. Removing a monitor no longer removes other active monitors or restores previously removed monitors. Repeated undo calls are harmless. diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py index f9cb47dc..b1e823ba 100644 --- a/src/pluggy/_manager.py +++ b/src/pluggy/_manager.py @@ -168,6 +168,8 @@ def __init__(self, project_name: str) -> None: "pluginmanage" ) self._inner_hookexec = _multicall + self._hookcall_monitor_base = _multicall + self._hookcall_monitors: list[tuple[_BeforeTrace, _AfterTrace]] = [] def _hookexec( self, @@ -587,25 +589,47 @@ def add_hookcall_monitoring( same arguments as ``before`` but also a :class:`~pluggy.Result` object which represents the result of the overall hook call. """ - oldcall = self._inner_hookexec + if not self._hookcall_monitors: + self._hookcall_monitor_base = self._inner_hookexec + + monitor = (before, after) + self._hookcall_monitors.append(monitor) + + def wrap( + oldcall: Callable[..., object | list[object]], + callbacks: tuple[_BeforeTrace, _AfterTrace], + ) -> Callable[..., object | list[object]]: + before_cb, after_cb = callbacks + + def traced_hookexec( + hook_name: str, + hook_impls: Sequence[HookImpl], + caller_kwargs: Mapping[str, object], + firstresult: bool, + ) -> object | list[object]: + before_cb(hook_name, hook_impls, caller_kwargs) + outcome = Result.from_call( + lambda: oldcall(hook_name, hook_impls, caller_kwargs, firstresult) + ) + after_cb(outcome, hook_name, hook_impls, caller_kwargs) + return outcome.get_result() - def traced_hookexec( - hook_name: str, - hook_impls: Sequence[HookImpl], - caller_kwargs: Mapping[str, object], - firstresult: bool, - ) -> object | list[object]: - before(hook_name, hook_impls, caller_kwargs) - outcome = Result.from_call( - lambda: oldcall(hook_name, hook_impls, caller_kwargs, firstresult) - ) - after(outcome, hook_name, hook_impls, caller_kwargs) - return outcome.get_result() + return traced_hookexec + + def rebuild() -> None: + hookexec = self._hookcall_monitor_base + for active in self._hookcall_monitors: + hookexec = wrap(hookexec, active) + self._inner_hookexec = hookexec - self._inner_hookexec = traced_hookexec + rebuild() def undo() -> None: - self._inner_hookexec = oldcall + for index, active in enumerate(self._hookcall_monitors): + if active is monitor: + del self._hookcall_monitors[index] + rebuild() + break return undo diff --git a/testing/test_monitor_undo.py b/testing/test_monitor_undo.py new file mode 100644 index 00000000..02c57272 --- /dev/null +++ b/testing/test_monitor_undo.py @@ -0,0 +1,189 @@ +"""Regression tests for issue #815.""" + +from __future__ import annotations + +from collections.abc import Callable + +from pluggy import HookimplMarker +from pluggy import HookspecMarker +from pluggy import PluginManager + + +def make_manager() -> PluginManager: + hookspec = HookspecMarker("monitor-undo") + hookimpl = HookimplMarker("monitor-undo") + + class Spec: + @hookspec + def ping(self): + """Test hook.""" + + class Plugin: + @hookimpl + def ping(self): + return "ok" + + manager = PluginManager("monitor-undo") + manager.add_hookspecs(Spec) + manager.register(Plugin()) + return manager + + +def add_monitor( + manager: PluginManager, events: list[str], label: str +) -> Callable[[], None]: + def before(*args): + events.append(f"{label}:before") + + def after(*args): + events.append(f"{label}:after") + + return manager.add_hookcall_monitoring(before, after) + + +def test_non_lifo_removal(): + manager = make_manager() + events: list[str] = [] + + undo_a = add_monitor(manager, events, "A") + undo_b = add_monitor(manager, events, "B") + + undo_a() + + assert manager.hook.ping() == ["ok"] + assert events == ["B:before", "B:after"] + + events.clear() + undo_b() + + assert manager.hook.ping() == ["ok"] + assert events == [] + + +def test_repeated_undo(): + manager = make_manager() + events: list[str] = [] + + undo_a = add_monitor(manager, events, "A") + undo_a() + + undo_b = add_monitor(manager, events, "B") + undo_a() + + assert manager.hook.ping() == ["ok"] + assert events == ["B:before", "B:after"] + + undo_b() + + +def test_lifo_order(): + manager = make_manager() + events: list[str] = [] + + undo_a = add_monitor(manager, events, "A") + undo_b = add_monitor(manager, events, "B") + + assert manager.hook.ping() == ["ok"] + assert events == [ + "B:before", + "A:before", + "A:after", + "B:after", + ] + + events.clear() + undo_b() + + assert manager.hook.ping() == ["ok"] + assert events == ["A:before", "A:after"] + + undo_a() + + +def test_undo_during_hook(): + manager = make_manager() + events: list[str] = [] + undo_callbacks: list[Callable[[], None]] = [] + + def before(*args): + events.append("before") + undo_callbacks[0]() + + def after(*args): + events.append("after") + + undo_callbacks.append(manager.add_hookcall_monitoring(before, after)) + + assert manager.hook.ping() == ["ok"] + assert events == ["before", "after"] + + events.clear() + assert manager.hook.ping() == ["ok"] + assert events == [] + + +def test_restore_custom_executor(): + manager = make_manager() + original = manager._inner_hookexec + + def custom_exec(name, impls, kwargs, firstresult): + return original(name, impls, kwargs, firstresult) + + manager._inner_hookexec = custom_exec + events: list[str] = [] + + undo = add_monitor(manager, events, "A") + undo() + + assert manager._inner_hookexec is custom_exec + assert manager.hook.ping() == ["ok"] + assert events == [] + + +def test_enable_tracing_removed_before_monitor() -> None: + manager = make_manager() + messages: list[str] = [] + events: list[str] = [] + + manager.trace.root.setwriter(messages.append) + + undo_trace = manager.enable_tracing() + undo_monitor = add_monitor(manager, events, "B") + + undo_trace() + + assert manager.hook.ping() == ["ok"] + assert events == ["B:before", "B:after"] + assert messages == [] + assert manager.trace.root.indent == 0 + + undo_monitor() + undo_trace() + + events.clear() + assert manager.hook.ping() == ["ok"] + assert events == [] + + +def test_remove_monitor_before_tracing() -> None: + manager = make_manager() + messages: list[str] = [] + events: list[str] = [] + + manager.trace.root.setwriter(messages.append) + + undo_monitor = add_monitor(manager, events, "A") + undo_trace = manager.enable_tracing() + + undo_monitor() + + assert manager.hook.ping() == ["ok"] + assert events == [] + assert messages + assert manager.trace.root.indent == 0 + + messages.clear() + undo_trace() + + assert manager.hook.ping() == ["ok"] + assert messages == []