test: add tests and docs for plugin.Host

pynvim/__init__.py

@@ -44,6 +44,8 @@ def start_host(session: Optional[Session] = None) -> None:
     This function is normally called at program startup and could have been
     defined as a separate executable. It is exposed as a library function for
     testing purposes only.
+
+    See also $VIMRUNTIME/autoload/provider/pythonx.vim for python host startup.
     """
     plugins = []
     for arg in sys.argv:

pynvim/plugin/host.py

@@ -194,45 +194,73 @@ def _missing_handler_error(self, name: str, *, kind: str) -> str:
     def _load(self, plugins: Sequence[str]) -> None:
         """Load the remote plugins and register handlers defined in the plugins.
 
-        Args:
-            plugins: List of plugin paths to rplugin python modules
-                registered by remote#host#RegisterPlugin('python3', ...)
-                (see the generated ~/.local/share/nvim/rplugin.vim manifest)
+        Parameters
+        ----------
+        plugins: List of plugin paths to rplugin python modules registered by
+            `remote#host#RegisterPlugin('python3', ...)`. Each element should
+            be either:
+            (1) "script_host.py": this is a special plugin for python3
+                rplugin host. See $VIMRUNTIME/autoload/provider/python3.vim
+                ; or
+            (2) (absolute) path to the top-level plugin module directory;
+                e.g., for a top-level python module `mymodule`: it would be
+                `"/path/to/plugin/rplugin/python3/mymodule"`.
+                See the generated ~/.local/share/nvim/rplugin.vim manifest
+                for real examples.
         """
         # self.nvim.err_write("host init _load\n", async_=True)
         has_script = False
         for path in plugins:
             path = os.path.normpath(path)  # normalize path
-            err = None
-            if path in self._loaded:
-                warn('{} is already loaded'.format(path))
-                continue
             try:
-                if path == "script_host.py":
-                    module = script_host
-                    has_script = True
-                else:
-                    directory, name = os.path.split(os.path.splitext(path)[0])
-                    module = _handle_import(directory, name)
-                handlers: List[Handler] = []
-                self._discover_classes(module, handlers, path)
-                self._discover_functions(module, handlers, path, delay=False)
-                if not handlers:
-                    error('{} exports no handlers'.format(path))
+                plugin_spec = self._load_plugin(path=path)
+                if not plugin_spec:
                     continue
-                self._loaded[path] = {'handlers': handlers, 'module': module}
+                if plugin_spec["path"] == "script_host.py":
+                    has_script = True
             except Exception as e:
-                err = ('Encountered {} loading plugin at {}: {}\n{}'
-                       .format(type(e).__name__, path, e, format_exc(5)))
-                error(err)
-                self._load_errors[path] = err
+                errmsg: str = (
+                    'Encountered {} loading plugin at {}: {}\n{}'.format(
+                        type(e).__name__, path, e, format_exc(5)))
+                error(errmsg)
+                self._load_errors[path] = errmsg
 
         kind = ("script-host" if len(plugins) == 1 and has_script
                 else "rplugin-host")
         info = get_client_info(kind, 'host', host_method_spec)
         self.name = info[0]
         self.nvim.api.set_client_info(*info, async_=True)
 
+    def _load_plugin(
+        self, path: str, *,
+        module: Optional[ModuleType] = None,
+    ) -> Union[Dict[str, Any], None]:
+        # Note: path must be normalized.
+        if path in self._loaded:
+            warn('{} is already loaded'.format(path))
+            return None
+
+        if path == "script_host.py":
+            module = script_host
+        elif module is not None:
+            pass  # Note: module is provided only when testing
+        else:
+            directory, module_name = os.path.split(os.path.splitext(path)[0])
+            module = _handle_import(directory, module_name)
+        handlers: List[Handler] = []
+        self._discover_classes(module, handlers, path)
+        self._discover_functions(module, handlers, path, delay=False)
+        if not handlers:
+            error('{} exports no handlers'.format(path))
+            return None
+
+        self._loaded[path] = {
+            'handlers': handlers,
+            'module': module,
+            'path': path,
+        }
+        return self._loaded[path]
+
     def _unload(self) -> None:
         for path, plugin in self._loaded.items():
             handlers = plugin['handlers']

test/test_host.py

@@ -1,9 +1,11 @@
 # type: ignore
 # pylint: disable=protected-access
 import os
+from types import SimpleNamespace
 from typing import Sequence
 
 from pynvim.api.nvim import Nvim
+from pynvim.plugin import decorators
 from pynvim.plugin.host import Host, host_method_spec
 from pynvim.plugin.script_host import ScriptHost
 
@@ -32,14 +34,91 @@ def test_host_import_rplugin_modules(vim: Nvim):
     ]
     h._load(plugins)
     assert len(h._loaded) == 2
+    assert len(h._specs) == 2
+    assert len(h._load_errors) == 0
 
     # pylint: disable-next=unbalanced-tuple-unpacking
     simple_nvim, mymodule = list(h._loaded.values())
     assert simple_nvim['module'].__name__ == 'simple_nvim'
     assert mymodule['module'].__name__ == 'mymodule'
 
 
-def test_host_clientinfo(vim):
+# @pytest.mark.timeout(5.0)
+def test_host_register_plugin_handlers(vim: Nvim):
+    """Test whether a Host can register plugin's RPC handlers."""
+    h = Host(vim)
+
+    @decorators.plugin
+    class TestPluginModule:
+        """A plugin for testing, having all types of the decorators."""
+        def __init__(self, nvim: Nvim):
+            self._nvim = nvim
+
+        @decorators.rpc_export('python_foobar', sync=True)
+        def foobar(self):
+            pass
+
+        @decorators.command("MyCommandSync", sync=True)
+        def command(self):
+            pass
+
+        @decorators.function("MyFunction", sync=True)
+        def function(self, a, b):
+            return a + b
+
+        @decorators.autocmd("BufEnter", pattern="*.py", sync=True)
+        def buf_enter(self):
+            vim.command("echom 'BufEnter'")
+
+        @decorators.rpc_export('python_foobar_async', sync=False)
+        def foobar_async(self):
+            pass
+
+        @decorators.command("MyCommandAsync", sync=False)
+        def command_async(self):
+            pass
+
+        @decorators.function("MyFunctionAsync", sync=False)
+        def function_async(self, a, b):
+            return a + b
+
+        @decorators.autocmd("BufEnter", pattern="*.async", sync=False)
+        def buf_enter_async(self):
+            vim.command("echom 'BufEnter'")
+
+        @decorators.shutdown_hook
+        def shutdown_hook():
+            print("bye")
+
+    @decorators.function("ModuleFunction")
+    def module_function(self):
+        pass
+
+    dummy_module = SimpleNamespace(
+        TestPluginModule=TestPluginModule,
+        module_function=module_function,
+    )
+    h._load_plugin("virtual://dummy_module", module=dummy_module)
+    assert list(h._loaded.keys()) == ["virtual://dummy_module"]
+    assert h._loaded['virtual://dummy_module']['module'] is dummy_module
+
+    # _notification_handlers: async commands and functions
+    print(h._notification_handlers.keys())
+    assert 'python_foobar_async' in h._notification_handlers
+    assert 'virtual://dummy_module:autocmd:BufEnter:*.async' in h._notification_handlers
+    assert 'virtual://dummy_module:command:MyCommandAsync' in h._notification_handlers
+    assert 'virtual://dummy_module:function:MyFunctionAsync' in h._notification_handlers
+    assert 'virtual://dummy_module:function:ModuleFunction' in h._notification_handlers
+
+    # _request_handlers: sync commands and functions
+    print(h._request_handlers.keys())
+    assert 'python_foobar' in h._request_handlers
+    assert 'virtual://dummy_module:autocmd:BufEnter:*.py' in h._request_handlers
+    assert 'virtual://dummy_module:command:MyCommandSync' in h._request_handlers
+    assert 'virtual://dummy_module:function:MyFunction' in h._request_handlers
+
+
+def test_host_clientinfo(vim: Nvim):
     h = Host(vim)
     assert h._request_handlers.keys() == host_method_spec.keys()
     assert 'remote' == vim.api.get_chan_info(vim.channel_id)['client']['type']