Repository: pytest-dev/pluggy (python)

Issue:
`PluginManager.subset_hook_caller` can cause high memory usage
`subset_hook_caller` creates a `HookManager` which excludes calls to a set of a plugins.

A major user of `subset_hook_caller` is pytest, for its conftest mechanism - conftest plugins not relevant for a given path are excluded. pytest calls `subset_hook_caller` quite a lot, usually using the `HookCaller` and immediately throwing it away. In pytest, the lifetime of the subset `HookCaller` is much shorter than the `PluginManager`.

pluggy however needs to handle the following scenario:

1. `subset_hook_caller` is called: `shc = pm.subset_hook_caller(..)`
2. A certain (non-excluded) plugin is unregistered: `pm.unregister(plugin)`.
3. The subset `HookCaller` is called: `shc.mymethod(...)`.

In step 3, `plugin` must not be called since it was unregistered in step 2. This means that the subset `HookCaller` must still be attached to the `PluginManager` somehow. This is done using a `_plugin2hookcallers` dict. The subset `HookCaller` adds itself to this dict:

https://github.com/pytest-dev/pluggy/blob/2b9998a277714b8ac99329e7ea5ce5908a1a89d7/src/pluggy/_manager.py#L453-L455

Alas, once the hookcaller is added to this dict, it is never removed; hence the memory is only released when the entire `PluginManager` is gone, not when the lifetime of just the subset `HookCaller` is done.

Pull request:
Improve subset hook caller & related changes
This PR contains 3 changes, which are (mostly) conceptually independent but better understood together.

The second commit fixes #346 and #347. The idea is, instead of creating a full-fledged HookCaller, create one which just proxies to the real HookCaller + the plugin filtering. The implementation is a bit hacky but the best I could come up after trying several approaches.

I ran the pytest testsuite and everything passes (after https://github.com/pytest-dev/pytest/pull/9512).

pluggy's benchmark suite looks good (not much difference, maybe a slight improvement). I still need to check pytest though, as pluggy's benchmark doesn't cover `subset_hook_caller` at all.

Please see the commits for details.

Patch:
diff --git a/src/pluggy/_hooks.py b/src/pluggy/_hooks.py
index 79c5054..403c39d 100644
--- a/src/pluggy/_hooks.py
+++ b/src/pluggy/_hooks.py
@@ -6,6 +6,7 @@ import sys
 import warnings
 from types import ModuleType
 from typing import (
+    AbstractSet,
     Any,
     Callable,
     Generator,
@@ -286,8 +287,7 @@ class _HookCaller:
         "name",
         "spec",
         "_hookexec",
-        "_wrappers",
-        "_nonwrappers",
+        "_hookimpls",
         "_call_history",
     )
 
@@ -300,8 +300,7 @@ class _HookCaller:
     ) -> None:
         self.name: "Final" = name
         self._hookexec: "Final" = hook_execute
-        self._wrappers: "Final[List[HookImpl]]" = []
-        self._nonwrappers: "Final[List[HookImpl]]" = []
+        self._hookimpls: "Final[List[HookImpl]]" = []
         self._call_history: Optional[_CallHistory] = None
         self.spec: Optional[HookSpec] = None
         if specmodule_or_class is not None:
@@ -325,38 +324,38 @@ class _HookCaller:
         return self._call_history is not None
 
     def _remove_plugin(self, plugin: _Plugin) -> None:
-        def remove(wrappers: List[HookImpl]) -> Optional[bool]:
-            for i, method in enumerate(wrappers):
-                if method.plugin == plugin:
-                    del wrappers[i]
-                    return True
-            return None
-
-        if remove(self._wrappers) is None:
-            if remove(self._nonwrappers) is None:
-                raise ValueError(f"plugin {plugin!r} not found")
+        for i, method in enumerate(self._hookimpls):
+            if method.plugin == plugin:
+                del self._hookimpls[i]
+                return
+        raise ValueError(f"plugin {plugin!r} not found")
 
     def get_hookimpls(self) -> List["HookImpl"]:
-        # Order is important for _hookexec
-        return self._nonwrappers + self._wrappers
+        return self._hookimpls.copy()
 
     def _add_hookimpl(self, hookimpl: "HookImpl") -> None:
         """Add an implementation to the callback chain."""
+        for i, method in enumerate(self._hookimpls):
+            if method.hookwrapper:
+                splitpoint = i
+                break
+        else:
+            splitpoint = len(self._hookimpls)
         if hookimpl.hookwrapper:
-            methods = self._wrappers
+            start, end = splitpoint, len(self._hookimpls)
         else:
-            methods = self._nonwrappers
+            start, end = 0, splitpoint
 
         if hookimpl.trylast:
-            methods.insert(0, hookimpl)
+            self._hookimpls.insert(start, hookimpl)
         elif hookimpl.tryfirst:
-            methods.append(hookimpl)
+            self._hookimpls.insert(end, hookimpl)
         else:
             # find last non-tryfirst method
-            i = len(methods) - 1
-            while i >= 0 and methods[i].tryfirst:
+            i = end - 1
+            while i >= start and self._hookimpls[i].tryfirst:
                 i -= 1
-            methods.insert(i + 1, hookimpl)
+            self._hookimpls.insert(i + 1, hookimpl)
 
     def __repr__(self) -> str:
         return f"<_HookCaller {self.name!r}>"
@@ -387,7 +386,7 @@ class _HookCaller:
         ), "Cannot directly call a historic hook - use call_historic instead."
         self._verify_all_args_are_provided(kwargs)
         firstresult = self.spec.opts.get("firstresult", False) if self.spec else False
-        return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
+        return self._hookexec(self.name, self._hookimpls, kwargs, firstresult)
 
     def call_historic(
         self,
@@ -406,7 +405,7 @@ class _HookCaller:
         self._call_history.append((kwargs, result_callback))
         # Historizing hooks don't return results.
         # Remember firstresult isn't compatible with historic.
-        res = self._hookexec(self.name, self.get_hookimpls(), kwargs, False)
+        res = self._hookexec(self.name, self._hookimpls, kwargs, False)
         if result_callback is None:
             return
         if isinstance(res, list):
@@ -429,13 +428,12 @@ class _HookCaller:
             "tryfirst": False,
             "specname": None,
         }
-        hookimpls = self.get_hookimpls()
+        hookimpls = self._hookimpls.copy()
         for method in methods:
             hookimpl = HookImpl(None, "<temp>", method, opts)
             # Find last non-tryfirst nonwrapper method.
             i = len(hookimpls) - 1
-            until = len(self._nonwrappers)
-            while i >= until and hookimpls[i].tryfirst:
+            while i >= 0 and hookimpls[i].tryfirst and not hookimpls[i].hookwrapper:
                 i -= 1
             hookimpls.insert(i + 1, hookimpl)
         firstresult = self.spec.opts.get("firstresult", False) if self.spec else False
@@ -453,6 +451,53 @@ class _HookCaller:
                     result_callback(res[0])
 
 
+class _SubsetHookCaller(_HookCaller):
+    """A proxy to another HookCaller which manages calls to all registered
+    plugins except the ones from remove_plugins."""
+
+    # This class is unusual: in inhertits from `_HookCaller` so all of
+    # the *code* runs in the class, but it delegates all underlying *data*
+    # to the original HookCaller.
+    # `subset_hook_caller` used to be implemented by creating a full-fledged
+    # HookCaller, copying all hookimpls from the original. This had problems
+    # with memory leaks (#346) and historic calls (#347), which make a proxy
+    # approach better.
+    # An alternative implementation is to use a `_getattr__`/`__getattribute__`
+    # proxy, however that adds more overhead and is more tricky to implement.
+
+    __slots__ = (
+        "_orig",
+        "_remove_plugins",
+        "name",
+        "_hookexec",
+    )
+
+    def __init__(self, orig: _HookCaller, remove_plugins: AbstractSet[_Plugin]) -> None:
+        self._orig = orig
+        self._remove_plugins = remove_plugins
+        self.name = orig.name  # type: ignore[misc]
+        self._hookexec = orig._hookexec  # type: ignore[misc]
+
+    @property  # type: ignore[misc]
+    def _hookimpls(self) -> List["HookImpl"]:  # type: ignore[override]
+        return [
+            impl
+            for impl in self._orig._hookimpls
+            if impl.plugin not in self._remove_plugins
+        ]
+
+    @property
+    def spec(self) -> Optional["HookSpec"]:  # type: ignore[override]
+        return self._orig.spec
+
+    @property
+    def _call_history(self) -> Optional[_CallHistory]:  # type: ignore[override]
+        return self._orig._call_history
+
+    def __repr__(self) -> str:
+        return f"<_SubsetHookCaller {self.name!r}>"
+
+
 class HookImpl:
     __slots__ = (
         "function",
diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py
index 906936c..3e16011 100644
--- a/src/pluggy/_manager.py
+++ b/src/pluggy/_manager.py
@@ -25,6 +25,7 @@ from ._hooks import (
     HookImpl,
     HookSpec,
     _HookCaller,
+    _SubsetHookCaller,
     _HookImplFunction,
     _HookRelay,
     _Namespace,
@@ -114,7 +115,6 @@ class PluginManager:
     def __init__(self, project_name: str) -> None:
         self.project_name: "Final" = project_name
         self._name2plugin: "Final[Dict[str, _Plugin]]" = {}
-        self._plugin2hookcallers: "Final[Dict[_Plugin, List[_HookCaller]]]" = {}
         self._plugin_distinfo: "Final[List[Tuple[_Plugin, DistFacade]]]" = []
         self.trace: "Final" = _tracing.TagTracer().get("pluginmanage")
         self.hook: "Final" = _HookRelay()
@@ -137,11 +137,17 @@ class PluginManager:
         is already registered."""
         plugin_name = name or self.get_canonical_name(plugin)
 
-        if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
+        if plugin_name in self._name2plugin:
             if self._name2plugin.get(plugin_name, -1) is None:
                 return None  # blocked plugin, return None to indicate no registration
             raise ValueError(
-                "Plugin already registered: %s=%s\n%s"
+                "Plugin name already registered: %s=%s\n%s"
+                % (plugin_name, plugin, self._name2plugin)
+            )
+
+        if plugin in self._name2plugin.values():
+            raise ValueError(
+                "Plugin already registered under a different name: %s=%s\n%s"
                 % (plugin_name, plugin, self._name2plugin)
             )
 
@@ -150,8 +156,6 @@ class PluginManager:
         self._name2plugin[plugin_name] = plugin
 
         # register matching hook implementations of the plugin
-        hookcallers: List[_HookCaller] = []
-        self._plugin2hookcallers[plugin] = hookcallers
         for name in dir(plugin):
             hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
             if hookimpl_opts is not None:
@@ -167,7 +171,6 @@ class PluginManager:
                     self._verify_hook(hook, hookimpl)
                     hook._maybe_apply_history(hookimpl)
                 hook._add_hookimpl(hookimpl)
-                hookcallers.append(hook)
         return plugin_name
 
     def parse_hookimpl_opts(
@@ -200,14 +203,16 @@ class PluginManager:
         if plugin is None:
             plugin = self.get_plugin(name)
 
+        hookcallers = self.get_hookcallers(plugin)
+        if hookcallers:
+            for hookcaller in hookcallers:
+                hookcaller._remove_plugin(plugin)
+
         # if self._name2plugin[name] == None registration was blocked: ignore
         if self._name2plugin.get(name):
             assert name is not None
             del self._name2plugin[name]
 
-        for hookcaller in self._plugin2hookcallers.pop(plugin, []):
-            hookcaller._remove_plugin(plugin)
-
         return plugin
 
     def set_blocked(self, name: str) -> None:
@@ -253,11 +258,11 @@ class PluginManager:
 
     def get_plugins(self) -> Set[Any]:
         """return the set of registered plugins."""
-        return set(self._plugin2hookcallers)
+        return set(self._name2plugin.values())
 
     def is_registered(self, plugin: _Plugin) -> bool:
         """Return ``True`` if the plugin is already registered."""
-        return plugin in self._plugin2hookcallers
+        return any(plugin == val for val in self._name2plugin.values())
 
     def get_canonical_name(self, plugin: _Plugin) -> str:
         """Return canonical name for a plugin object. Note that a plugin
@@ -372,7 +377,14 @@ class PluginManager:
 
     def get_hookcallers(self, plugin: _Plugin) -> Optional[List[_HookCaller]]:
         """get all hook callers for the specified plugin."""
-        return self._plugin2hookcallers.get(plugin)
+        if self.get_name(plugin) is None:
+            return None
+        hookcallers = []
+        for hookcaller in self.hook.__dict__.values():
+            for hookimpl in hookcaller.get_hookimpls():
+                if hookimpl.plugin is plugin:
+                    hookcallers.append(hookcaller)
+        return hookcallers
 
     def add_hookcall_monitoring(
         self, before: _BeforeTrace, after: _AfterTrace
@@ -436,24 +448,13 @@ class PluginManager:
     def subset_hook_caller(
         self, name: str, remove_plugins: Iterable[_Plugin]
     ) -> _HookCaller:
-        """Return a new :py:class:`._hooks._HookCaller` instance for the named method
-        which manages calls to all registered plugins except the
-        ones from remove_plugins."""
+        """Return a proxy :py:class:`._hooks._HookCaller` instance for the named
+        method which manages calls to all registered plugins except the ones
+        from remove_plugins."""
         orig: _HookCaller = getattr(self.hook, name)
-        plugins_to_remove = [plug for plug in remove_plugins if hasattr(plug, name)]
+        plugins_to_remove = {plug for plug in remove_plugins if hasattr(plug, name)}
         if plugins_to_remove:
-            assert orig.spec is not None
-            hc = _HookCaller(
-                orig.name, orig._hookexec, orig.spec.namespace, orig.spec.opts
-            )
-            for hookimpl in orig.get_hookimpls():
-                plugin = hookimpl.plugin
-                if plugin not in plugins_to_remove:
-                    hc._add_hookimpl(hookimpl)
-                    # we also keep track of this hook caller so it
-                    # gets properly removed on plugin unregistration
-                    self._plugin2hookcallers.setdefault(plugin, []).append(hc)
-            return hc
+            return _SubsetHookCaller(orig, plugins_to_remove)
         return orig
 
 


Test patch:
diff --git a/testing/test_details.py b/testing/test_details.py
index b4e826b..a153de0 100644
--- a/testing/test_details.py
+++ b/testing/test_details.py
@@ -34,13 +34,18 @@ def test_parse_hookimpl_override() -> None:
     pm = MyPluginManager(hookspec.project_name)
     pm.register(Plugin())
     pm.add_hookspecs(Spec)
-    assert not pm.hook.x1meth._nonwrappers[0].hookwrapper
-    assert not pm.hook.x1meth._nonwrappers[0].tryfirst
-    assert not pm.hook.x1meth._nonwrappers[0].trylast
-    assert not pm.hook.x1meth._nonwrappers[0].optionalhook
 
-    assert pm.hook.x1meth2._wrappers[0].tryfirst
-    assert pm.hook.x1meth2._wrappers[0].hookwrapper
+    hookimpls = pm.hook.x1meth.get_hookimpls()
+    assert len(hookimpls) == 1
+    assert not hookimpls[0].hookwrapper
+    assert not hookimpls[0].tryfirst
+    assert not hookimpls[0].trylast
+    assert not hookimpls[0].optionalhook
+
+    hookimpls = pm.hook.x1meth2.get_hookimpls()
+    assert len(hookimpls) == 1
+    assert hookimpls[0].hookwrapper
+    assert hookimpls[0].tryfirst
 
 
 def test_warn_when_deprecated_specified(recwarn) -> None:
@@ -127,6 +132,6 @@ def test_repr() -> None:
 
     plugin = Plugin()
     pname = pm.register(plugin)
-    assert repr(pm.hook.myhook._nonwrappers[0]) == (
+    assert repr(pm.hook.myhook.get_hookimpls()[0]) == (
         f"<HookImpl plugin_name={pname!r}, plugin={plugin!r}>"
     )
diff --git a/testing/test_hookcaller.py b/testing/test_hookcaller.py
index 196a081..4c123db 100644
--- a/testing/test_hookcaller.py
+++ b/testing/test_hookcaller.py
@@ -61,7 +61,7 @@ def test_adding_nonwrappers(hc: _HookCaller, addmeth: AddMeth) -> None:
     def he_method3() -> None:
         pass
 
-    assert funcs(hc._nonwrappers) == [he_method1, he_method2, he_method3]
+    assert funcs(hc.get_hookimpls()) == [he_method1, he_method2, he_method3]
 
 
 def test_adding_nonwrappers_trylast(hc: _HookCaller, addmeth: AddMeth) -> None:
@@ -77,7 +77,7 @@ def test_adding_nonwrappers_trylast(hc: _HookCaller, addmeth: AddMeth) -> None:
     def he_method1_b() -> None:
         pass
 
-    assert funcs(hc._nonwrappers) == [he_method1, he_method1_middle, he_method1_b]
+    assert funcs(hc.get_hookimpls()) == [he_method1, he_method1_middle, he_method1_b]
 
 
 def test_adding_nonwrappers_trylast3(hc: _HookCaller, addmeth: AddMeth) -> None:
@@ -97,7 +97,7 @@ def test_adding_nonwrappers_trylast3(hc: _HookCaller, addmeth: AddMeth) -> None:
     def he_method1_d() -> None:
         pass
 
-    assert funcs(hc._nonwrappers) == [
+    assert funcs(hc.get_hookimpls()) == [
         he_method1_d,
         he_method1_b,
         he_method1_a,
@@ -118,7 +118,7 @@ def test_adding_nonwrappers_trylast2(hc: _HookCaller, addmeth: AddMeth) -> None:
     def he_method1() -> None:
         pass
 
-    assert funcs(hc._nonwrappers) == [he_method1, he_method1_middle, he_method1_b]
+    assert funcs(hc.get_hookimpls()) == [he_method1, he_method1_middle, he_method1_b]
 
 
 def test_adding_nonwrappers_tryfirst(hc: _HookCaller, addmeth: AddMeth) -> None:
@@ -134,7 +134,7 @@ def test_adding_nonwrappers_tryfirst(hc: _HookCaller, addmeth: AddMeth) -> None:
     def he_method1_b() -> None:
         pass
 
-    assert funcs(hc._nonwrappers) == [he_method1_middle, he_method1_b, he_method1]
+    assert funcs(hc.get_hookimpls()) == [he_method1_middle, he_method1_b, he_method1]
 
 
 def test_adding_wrappers_ordering(hc: _HookCaller, addmeth: AddMeth) -> None:
@@ -150,8 +150,11 @@ def test_adding_wrappers_ordering(hc: _HookCaller, addmeth: AddMeth) -> None:
     def he_method3() -> None:
         pass
 
-    assert funcs(hc._nonwrappers) == [he_method1_middle]
-    assert funcs(hc._wrappers) == [he_method1, he_method3]
+    assert funcs(hc.get_hookimpls()) == [
+        he_method1_middle,
+        he_method1,
+        he_method3,
+    ]
 
 
 def test_adding_wrappers_ordering_tryfirst(hc: _HookCaller, addmeth: AddMeth) -> None:
@@ -163,8 +166,116 @@ def test_adding_wrappers_ordering_tryfirst(hc: _HookCaller, addmeth: AddMeth) ->
     def he_method2() -> None:
         pass
 
-    assert hc._nonwrappers == []
-    assert funcs(hc._wrappers) == [he_method2, he_method1]
+    assert funcs(hc.get_hookimpls()) == [he_method2, he_method1]
+
+
+def test_adding_wrappers_complex(hc: _HookCaller, addmeth: AddMeth) -> None:
+    assert funcs(hc.get_hookimpls()) == []
+
+    @addmeth(hookwrapper=True, trylast=True)
+    def m1() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m1]
+
+    @addmeth()
+    def m2() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m2, m1]
+
+    @addmeth(trylast=True)
+    def m3() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m3, m2, m1]
+
+    @addmeth(hookwrapper=True)
+    def m4() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m3, m2, m1, m4]
+
+    @addmeth(hookwrapper=True, tryfirst=True)
+    def m5() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m3, m2, m1, m4, m5]
+
+    @addmeth(tryfirst=True)
+    def m6() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m3, m2, m6, m1, m4, m5]
+
+    @addmeth()
+    def m7() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m3, m2, m7, m6, m1, m4, m5]
+
+    @addmeth(hookwrapper=True)
+    def m8() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m3, m2, m7, m6, m1, m4, m8, m5]
+
+    @addmeth(trylast=True)
+    def m9() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m9, m3, m2, m7, m6, m1, m4, m8, m5]
+
+    @addmeth(tryfirst=True)
+    def m10() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m9, m3, m2, m7, m6, m10, m1, m4, m8, m5]
+
+    @addmeth(hookwrapper=True, trylast=True)
+    def m11() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [m9, m3, m2, m7, m6, m10, m11, m1, m4, m8, m5]
+
+    @addmeth(hookwrapper=True)
+    def m12() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [
+        m9,
+        m3,
+        m2,
+        m7,
+        m6,
+        m10,
+        m11,
+        m1,
+        m4,
+        m8,
+        m12,
+        m5,
+    ]
+
+    @addmeth()
+    def m13() -> None:
+        ...
+
+    assert funcs(hc.get_hookimpls()) == [
+        m9,
+        m3,
+        m2,
+        m7,
+        m13,
+        m6,
+        m10,
+        m11,
+        m1,
+        m4,
+        m8,
+        m12,
+        m5,
+    ]
 
 
 def test_hookspec(pm: PluginManager) -> None:
diff --git a/testing/test_pluginmanager.py b/testing/test_pluginmanager.py
index 794b80e..b222192 100644
--- a/testing/test_pluginmanager.py
+++ b/testing/test_pluginmanager.py
@@ -153,20 +153,25 @@ def test_register_hookwrapper_not_a_generator_function(he_pm: PluginManager) ->
 
 def test_register(pm: PluginManager) -> None:
     class MyPlugin:
-        pass
+        @hookimpl
+        def he_method1(self):
+            ...
 
     my = MyPlugin()
     pm.register(my)
-    assert my in pm.get_plugins()
+    assert pm.get_plugins() == {my}
     my2 = MyPlugin()
     pm.register(my2)
-    assert {my, my2}.issubset(pm.get_plugins())
+    assert pm.get_plugins() == {my, my2}
 
     assert pm.is_registered(my)
     assert pm.is_registered(my2)
     pm.unregister(my)
     assert not pm.is_registered(my)
-    assert my not in pm.get_plugins()
+    assert pm.get_plugins() == {my2}
+
+    with pytest.raises(AssertionError, match=r"not registered"):
+        pm.unregister(my)
 
 
 def test_register_unknown_hooks(pm: PluginManager) -> None:
@@ -221,6 +226,39 @@ def test_register_historic(pm: PluginManager) -> None:
     assert out == [1, 10, 120, 12]
 
 
+def test_historic_with_subset_hook_caller(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def he_method1(self, arg):
+            ...
+
+    pm.add_hookspecs(Hooks)
+
+    out = []
+
+    class Plugin:
+        @hookimpl
+        def he_method1(self, arg):
+            out.append(arg)
+
+    plugin = Plugin()
+    pm.register(plugin)
+
+    class Plugin2:
+        @hookimpl
+        def he_method1(self, arg):
+            out.append(arg * 10)
+
+    shc = pm.subset_hook_caller("he_method1", remove_plugins=[plugin])
+    shc.call_historic(kwargs=dict(arg=1))
+
+    pm.register(Plugin2())
+    assert out == [10]
+
+    pm.register(Plugin())
+    assert out == [10, 1]
+
+
 @pytest.mark.parametrize("result_callback", [True, False])
 def test_with_result_memorized(pm: PluginManager, result_callback: bool) -> None:
     """Verify that ``_HookCaller._maybe_apply_history()`
@@ -400,6 +438,8 @@ def test_subset_hook_caller(pm: PluginManager) -> None:
     pm.hook.he_method1(arg=1)
     assert out == [10]
 
+    assert repr(hc) == "<_SubsetHookCaller 'he_method1'>"
+
 
 def test_get_hookimpls(pm: PluginManager) -> None:
     class Hooks:
@@ -433,6 +473,58 @@ def test_get_hookimpls(pm: PluginManager) -> None:
     assert hook_plugins == [plugin1, plugin2]
 
 
+def test_get_hookcallers(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec
+        def he_method1(self):
+            ...
+
+        @hookspec
+        def he_method2(self):
+            ...
+
+    pm.add_hookspecs(Hooks)
+
+    class Plugin1:
+        @hookimpl
+        def he_method1(self):
+            ...
+
+        @hookimpl
+        def he_method2(self):
+            ...
+
+    class Plugin2:
+        @hookimpl
+        def he_method1(self):
+            ...
+
+    class Plugin3:
+        @hookimpl
+        def he_method2(self):
+            ...
+
+    plugin1 = Plugin1()
+    pm.register(plugin1)
+    plugin2 = Plugin2()
+    pm.register(plugin2)
+    plugin3 = Plugin3()
+    pm.register(plugin3)
+
+    hookcallers1 = pm.get_hookcallers(plugin1)
+    assert hookcallers1 is not None
+    assert len(hookcallers1) == 2
+    hookcallers2 = pm.get_hookcallers(plugin2)
+    assert hookcallers2 is not None
+    assert len(hookcallers2) == 1
+    hookcallers3 = pm.get_hookcallers(plugin3)
+    assert hookcallers3 is not None
+    assert len(hookcallers3) == 1
+    assert hookcallers1 == hookcallers2 + hookcallers3
+
+    assert pm.get_hookcallers(object()) is None
+
+
 def test_add_hookspecs_nohooks(pm: PluginManager) -> None:
     class NoHooks:
         pass