Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog/811.bugfix.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix hook implementations remaining active when a plugin is unregistered or blocked during historic replay.
22 changes: 20 additions & 2 deletions src/pluggy/_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,7 @@ def __init__(self, project_name: str) -> None:
self.project_name: Final = project_name
self._name2plugin: Final[dict[str, _Plugin]] = {}
self._name2dist: Final[dict[str, importlib.metadata.Distribution]] = {}
self._registration_tokens: Final[dict[str, object]] = {}
#: The "hook relay", used to call a hook on all registered plugins.
#: See :ref:`calling`.
self.hook: Final = HookRelay()
Expand Down Expand Up @@ -239,8 +240,23 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
hookimpls.append((hook_name, hookimpl, verified))

self._name2plugin[plugin_name] = plugin
token = object()
self._registration_tokens[plugin_name] = token
try:
self._register_hookimpls(plugin_name, hookimpls, token)
finally:
if self._registration_tokens.get(plugin_name) is token:
del self._registration_tokens[plugin_name]
return plugin_name

def _register_hookimpls(
self,
plugin_name: str,
hookimpls: Sequence[tuple[str, HookImpl, bool]],
token: object,
) -> None:
for hook_name, hookimpl, verified in hookimpls:
hook = getattr(self.hook, hook_name, None)
hook: HookCaller | None = getattr(self.hook, hook_name, None)
if hook is None:
hook = HookCaller(hook_name, self._hookexec)
setattr(self.hook, hook_name, hook)
Expand All @@ -250,8 +266,9 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
# registered, but only user code during replay can get here.
self._verify_hook(hook, hookimpl)
hook._maybe_apply_history(hookimpl)
if self._registration_tokens.get(plugin_name) is not token:
return
hook._add_hookimpl(hookimpl)
return plugin_name

def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None:
"""Try to obtain a hook implementation from an item with the given name
Expand Down Expand Up @@ -309,6 +326,7 @@ def unregister(
f" plugin is registered as {registered_name!r})"
)

self._registration_tokens.pop(name, None)
hookcallers = self.get_hookcallers(plugin)
if hookcallers:
for hookcaller in hookcallers:
Expand Down
Loading
Loading