Make subset_hook_caller() return a proxy HookCaller instead of a standalone HookCaller

Fix #346.
Fix #347.

Author: bluetech

src/pluggy/_hooks.py

@@ -6,6 +6,7 @@
 import warnings
 from types import ModuleType
 from typing import (
+    AbstractSet,
     Any,
     Callable,
     Generator,
@@ -450,6 +451,53 @@ def _maybe_apply_history(self, method: "HookImpl") -> None:
                     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",

src/pluggy/_manager.py

@@ -25,6 +25,7 @@
     HookImpl,
     HookSpec,
     _HookCaller,
+    _SubsetHookCaller,
     _HookImplFunction,
     _HookRelay,
     _Namespace,
@@ -436,24 +437,13 @@ def after(
     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
 
 

testing/test_pluginmanager.py

@@ -221,6 +221,39 @@ def he_method1(self, arg):
     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 +433,8 @@ class PluginNo:
     pm.hook.he_method1(arg=1)
     assert out == [10]
 
+    assert repr(hc) == "<_SubsetHookCaller 'he_method1'>"
+
 
 def test_get_hookimpls(pm: PluginManager) -> None:
     class Hooks: