From fbbf14c75de7e06247b6ef5f09b1a485845d5262 Mon Sep 17 00:00:00 2001 From: xiehuanyi <74809554+xiehuanyi@users.noreply.github.com> Date: Mon, 12 Oct 2026 00:14:24 +0300 Subject: [PATCH 1/2] Roll back registration when historic replay raises --- changelog/761.bugfix.rst | 3 ++ src/pluggy/_manager.py | 30 +++++++++------- testing/test_pluginmanager.py | 67 +++++++++++++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 12 deletions(-) create mode 100644 changelog/761.bugfix.rst diff --git a/changelog/761.bugfix.rst b/changelog/761.bugfix.rst new file mode 100644 index 00000000..947cf38c --- /dev/null +++ b/changelog/761.bugfix.rst @@ -0,0 +1,3 @@ +Roll back a plugin's registration if replaying a historic hook or its result +callback raises. Previously, the plugin and some of its hook implementations +could remain registered after the failure. diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py index 4994c867..ceb7f6f6 100644 --- a/src/pluggy/_manager.py +++ b/src/pluggy/_manager.py @@ -238,18 +238,24 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None: hookimpls.append((hook_name, hookimpl, verified)) self._name2plugin[plugin_name] = plugin - for hook_name, hookimpl, verified in hookimpls: - hook = getattr(self.hook, hook_name, None) - if hook is None: - hook = HookCaller(hook_name, self._hookexec) - setattr(self.hook, hook_name, hook) - elif not verified and hook.has_spec(): - # An earlier impl's historic replay added this spec after the - # first pass. Failing here leaves the plugin partially - # registered, but only user code during replay can get here. - self._verify_hook(hook, hookimpl) - hook._maybe_apply_history(hookimpl) - hook._add_hookimpl(hookimpl) + try: + for hook_name, hookimpl, verified in hookimpls: + hook = getattr(self.hook, hook_name, None) + if hook is None: + hook = HookCaller(hook_name, self._hookexec) + setattr(self.hook, hook_name, hook) + elif not verified and hook.has_spec(): + # An earlier impl's historic replay added this spec after + # the first pass, so validate against it now. + self._verify_hook(hook, hookimpl) + hook._maybe_apply_history(hookimpl) + hook._add_hookimpl(hookimpl) + except BaseException: + # Replay runs user code, including result callbacks. Remove the + # failed plugin without disturbing other plugins registered by it. + if self._name2plugin.get(plugin_name) is plugin: + self.unregister(name=plugin_name) + raise return plugin_name def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None: diff --git a/testing/test_pluginmanager.py b/testing/test_pluginmanager.py index 06d645d7..f5d8781e 100644 --- a/testing/test_pluginmanager.py +++ b/testing/test_pluginmanager.py @@ -427,6 +427,73 @@ def late(self, qlwkje): with pytest.raises(PluginValidationError, match="qlwkje"): pm.register(hello()) + assert pm.list_name_plugin() == [] + assert pm.hook.configure.get_hookimpls() == [] + assert pm.hook.late.get_hookimpls() == [] + + +@pytest.mark.parametrize("exception_type", [RuntimeError, KeyboardInterrupt]) +@pytest.mark.parametrize("fail_in_callback", [False, True]) +def test_register_rolls_back_failed_historic_replay( + pm: PluginManager, exception_type: type[BaseException], fail_in_callback: bool +) -> None: + class Hooks: + @hookspec + def a_regular(self): ... + + @hookspec(historic=True) + def b_historic(self): ... + + @hookspec + def c_regular(self): ... + + class ExistingPlugin: + @hookimpl + def a_regular(self): + return "existing" + + error = exception_type("historic replay failed") + fail = True + + def callback(result): + if fail and fail_in_callback: + raise error + + class Plugin: + @hookimpl + def a_regular(self): + return "new" + + @hookimpl + def b_historic(self): + if fail and not fail_in_callback: + raise error + return "historic" + + @hookimpl + def c_regular(self): + return "last" + + pm.add_hookspecs(Hooks) + existing = ExistingPlugin() + pm.register(existing, name="existing") + pm.hook.b_historic.call_historic(result_callback=callback) + plugin = Plugin() + + with pytest.raises(exception_type) as excinfo: + pm.register(plugin, name="new") + + assert excinfo.value is error + assert pm.list_name_plugin() == [("existing", existing)] + assert pm.hook.a_regular() == ["existing"] + assert pm.hook.b_historic.get_hookimpls() == [] + assert pm.hook.c_regular() == [] + + fail = False + assert pm.register(plugin, name="new") == "new" + assert pm.hook.a_regular() == ["new", "existing"] + assert pm.hook.c_regular() == ["last"] + def test_register_hookwrapper_not_a_generator_function(he_pm: PluginManager) -> None: class hello: From 60cc9e4babd4ecaa4e72ac1ac6b44f63385785ad Mon Sep 17 00:00:00 2001 From: xiehuanyi <74809554+xiehuanyi@users.noreply.github.com> Date: Mon, 12 Oct 2026 00:22:02 +0300 Subject: [PATCH 2/2] Test replacement plugins after historic replay failure --- testing/test_pluginmanager.py | 41 +++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/testing/test_pluginmanager.py b/testing/test_pluginmanager.py index f5d8781e..34c7dbba 100644 --- a/testing/test_pluginmanager.py +++ b/testing/test_pluginmanager.py @@ -495,6 +495,47 @@ def c_regular(self): assert pm.hook.c_regular() == ["last"] +def test_register_keeps_replacement_after_historic_replay_failure( + pm: PluginManager, +) -> None: + class Hooks: + @hookspec + def a_regular(self): ... + + @hookspec(historic=True) + def b_historic(self): ... + + class Replacement: + @hookimpl + def a_regular(self): + return "replacement" + + replacement = Replacement() + error = RuntimeError("historic replay failed") + + class Plugin: + @hookimpl + def a_regular(self): + return "failed" # pragma: no cover + + @hookimpl + def b_historic(self): + pm.unregister(name="plugin") + pm.register(replacement, name="plugin") + raise error + + pm.add_hookspecs(Hooks) + pm.hook.b_historic.call_historic() + + with pytest.raises(RuntimeError) as excinfo: + pm.register(Plugin(), name="plugin") + + assert excinfo.value is error + assert pm.get_plugin("plugin") is replacement + assert pm.hook.a_regular() == ["replacement"] + assert pm.hook.b_historic.get_hookimpls() == [] + + def test_register_hookwrapper_not_a_generator_function(he_pm: PluginManager) -> None: class hello: @hookimpl(hookwrapper=True)