Temporary registration via context manager

changelog/52.feature.rst

@@ -0,0 +1,4 @@
+Add ``PluginManager.register_all``, a context manager for registering multiple
+plugins temporarily. Plugins registered by the context are unregistered in
+reverse order on exit, including when the block or registration raises an
+exception.

src/pluggy/_manager.py

@@ -2,8 +2,10 @@
 
 from collections.abc import Callable
 from collections.abc import Iterable
+from collections.abc import Iterator
 from collections.abc import Mapping
 from collections.abc import Sequence
+from contextlib import contextmanager
 import inspect
 import types
 from typing import Any
@@ -252,6 +254,36 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
             hook._add_hookimpl(hookimpl)
         return plugin_name
 
+    @contextmanager
+    def register_all(self, plugins: Iterable[_Plugin]) -> Iterator[PluginManager]:
+        """Temporarily register plugins for the duration of a ``with`` block.
+
+        :param plugins:
+            Plugins to register, in order, using :meth:`register`.
+        :returns:
+            This plugin manager while the context is active.
+
+        For example::
+
+            with pm.register_all([plugin1, plugin2]) as active_pm:
+                assert active_pm is pm
+
+        Only plugins successfully registered by this context are unregistered,
+        in reverse order, even if registration or the block raises an exception.
+        Blocked plugin names remain blocked. Historic hook calls are replayed
+        upon registration; unregistering cannot undo those effects.
+        """
+        registered: list[tuple[_Plugin, str]] = []
+        try:
+            for plugin in plugins:
+                name = self.register(plugin)
+                if name is not None:
+                    registered.append((plugin, name))
+            yield self
+        finally:
+            for plugin, name in reversed(registered):
+                self.unregister(plugin=plugin, name=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
         in the given plugin which is being searched for hook impls.

testing/test_pluginmanager.py

@@ -103,6 +103,172 @@ class A:
     assert not pm.get_plugins()
 
 
+def test_register_all_temporarily_adds_hook_implementations(
+    he_pm: PluginManager,
+) -> None:
+    class Plugin:
+        def __init__(self, increment: int) -> None:
+            self.increment = increment
+
+        @hookimpl
+        def he_method1(self, arg: int) -> int:
+            return arg + self.increment
+
+    existing, first, second = Plugin(0), Plugin(1), Plugin(2)
+    assert he_pm.register(existing, name="existing") == "existing"
+
+    with he_pm.register_all([first, second]) as active_pm:
+        assert active_pm is he_pm
+        assert he_pm.get_plugins() == {existing, first, second}
+        assert he_pm.get_name(first) is not None
+        assert he_pm.get_name(second) is not None
+        assert he_pm.hook.he_method1(arg=10) == [12, 11, 10]
+
+    assert he_pm.get_plugins() == {existing}
+    assert he_pm.get_name(first) is None
+    assert he_pm.get_name(second) is None
+    assert he_pm.get_plugin("existing") is existing
+    assert he_pm.hook.he_method1(arg=10) == [10]
+
+
+def test_register_all_single_and_empty_iterable(pm: PluginManager) -> None:
+    plugin = object()
+    with pm.register_all(iter(())) as active_pm:
+        assert active_pm is pm
+        assert pm.get_plugins() == set()
+
+    with pm.register_all(iter((plugin,))) as active_pm:
+        assert active_pm is pm
+        assert pm.get_plugins() == {plugin}
+    assert pm.get_plugins() == set()
+
+
+def test_register_all_unregisters_in_reverse_order(
+    pm: PluginManager, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    plugins = [object(), object(), object()]
+    removed: list[object | None] = []
+    original_unregister = pm.unregister
+
+    def recording_unregister(
+        plugin: object | None = None, name: str | None = None
+    ) -> object | None:
+        removed.append(plugin)
+        return original_unregister(plugin=plugin, name=name)
+
+    monkeypatch.setattr(pm, "unregister", recording_unregister)
+    with pm.register_all(plugins):
+        assert pm.get_plugins() == set(plugins)
+    assert removed == list(reversed(plugins))
+
+
+def test_register_all_cleans_up_after_body_exception(pm: PluginManager) -> None:
+    plugin = object()
+    with (
+        pytest.raises(RuntimeError, match="body failed"),
+        pm.register_all([plugin]),
+    ):
+        assert pm.is_registered(plugin)
+        raise RuntimeError("body failed")
+    assert not pm.is_registered(plugin)
+    assert pm.get_plugins() == set()
+
+
+def test_register_all_rolls_back_duplicate_plugin(pm: PluginManager) -> None:
+    existing, temporary = object(), object()
+    assert pm.register(existing) is not None
+
+    with (
+        pytest.raises(ValueError, match="already registered"),
+        pm.register_all([temporary, existing]),
+    ):
+        pass
+
+    assert pm.get_plugins() == {existing}
+    assert pm.get_name(temporary) is None
+
+
+def test_register_all_rolls_back_validation_failure(he_pm: PluginManager) -> None:
+    class Valid:
+        @hookimpl
+        def he_method1(self, arg: int) -> None:
+            pass  # pragma: no cover
+
+    class Invalid:
+        @hookimpl
+        def he_method1(self, unexpected: int) -> None:
+            pass  # pragma: no cover
+
+    temporary = Valid()
+    with (
+        pytest.raises(PluginValidationError),
+        he_pm.register_all([temporary, Invalid()]),
+    ):
+        pass
+
+    assert he_pm.get_plugins() == set()
+    assert he_pm.get_name(temporary) is None
+    assert he_pm.hook.he_method1(arg=1) == []
+
+
+def test_register_all_preserves_blocked_names(pm: PluginManager) -> None:
+    blocked, temporary = object(), object()
+    blocked_name = pm.get_canonical_name(blocked)
+    pm.set_blocked(blocked_name)
+
+    with pm.register_all([blocked, temporary]):
+        assert pm.is_blocked(blocked_name)
+        assert pm.get_plugins() == {temporary}
+
+    assert pm.is_blocked(blocked_name)
+    assert pm.get_plugins() == set()
+    assert pm.register(blocked) is None
+
+
+def test_register_all_nested_contexts(he_pm: PluginManager) -> None:
+    class Plugin:
+        def __init__(self, result: int) -> None:
+            self.result = result
+
+        @hookimpl
+        def he_method1(self, arg: int) -> int:
+            return self.result
+
+    outer, inner = Plugin(1), Plugin(2)
+    with he_pm.register_all([outer]):
+        assert he_pm.hook.he_method1(arg=0) == [1]
+        with he_pm.register_all([inner]):
+            assert he_pm.hook.he_method1(arg=0) == [2, 1]
+        assert he_pm.get_plugins() == {outer}
+        assert he_pm.hook.he_method1(arg=0) == [1]
+    assert he_pm.get_plugins() == set()
+    assert he_pm.hook.he_method1(arg=0) == []
+
+
+def test_register_all_replays_historic_hooks_on_entry(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def he_method1(self, arg: int) -> None: ...
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.he_method1.call_historic(kwargs={"arg": 4})
+    seen: list[int] = []
+
+    class Plugin:
+        @hookimpl
+        def he_method1(self, arg: int) -> None:
+            seen.append(arg)
+
+    plugin = Plugin()
+    with pm.register_all([plugin]):
+        assert seen == [4]
+        assert pm.is_registered(plugin)
+
+    pm.hook.he_method1.call_historic(kwargs={"arg": 9})
+    assert seen == [4]
+    assert pm.get_plugins() == set()
+
+
 def test_set_blocked(pm: PluginManager) -> None:
     class A:
         pass