test that the synchronizer installs and loads plugins

plugin_runner/plugin_runner.py

@@ -202,6 +202,40 @@ async def ReloadPlugins(
             yield ReloadPluginsResponse(success=True)
 
 
+async def synchronize_plugins(max_iterations: int = -1) -> None:
+    """Listen for messages on the pubsub channel that will indicate it is necessary to reinstall and reload plugins."""
+    client, pubsub = get_client()
+    await pubsub.psubscribe(settings.CHANNEL_NAME)
+    log.info("Listening for messages on pubsub channel")
+    iterations: int = 0
+    while iterations < max_iterations:
+        if max_iterations > 0:  # max_iterations == -1 means infinite iterations
+            iterations += 1
+        message = await pubsub.get_message(ignore_subscribe_messages=True)
+        if message is not None:
+            log.info("Received message from pubsub channel")
+
+            message_type = message.get("type", "")
+
+            if message_type != "pmessage":
+                continue
+
+            data = pickle.loads(message.get("data", pickle.dumps({})))
+
+            if "action" not in data:
+                continue
+
+            if data["action"] == "reload":
+                try:
+                    log.info(
+                        "plugin-synchronizer: installing and reloading plugins after receiving command"
+                    )
+                    install_plugins()
+                    load_plugins()
+                except Exception as e:
+                    print("plugin-synchronizer: `install_plugins` failed:", e)
+
+
 def validate_effects(effects: list[Effect]) -> list[Effect]:
     """Validates the effects based on predefined rules.
     Keeps only the first AUTOCOMPLETE_SEARCH_RESULTS effect and preserve all non-search-related effects.
@@ -285,37 +319,6 @@ def get_client() -> tuple[redis.Redis, redis.client.PubSub]:
     return client, pubsub
 
 
-async def synchronize_plugins() -> None:
-    """Listen for messages on the pubsub channel that will indicate it is necessary to reinstall and reload plugins."""
-    client, pubsub = get_client()
-    await pubsub.psubscribe(settings.CHANNEL_NAME)
-    log.info("Listening for messages on pubsub channel")
-    while True:
-        message = await pubsub.get_message(ignore_subscribe_messages=True)
-        if message is not None:
-            log.info("Received message from pubsub channel")
-
-            message_type = message.get("type", "")
-
-            if message_type != "pmessage":
-                continue
-
-            data = pickle.loads(message.get("data", pickle.dumps({})))
-
-            if "action" not in data:
-                continue
-
-            if data["action"] == "reload":
-                try:
-                    log.info(
-                        "plugin-synchronizer: installing and reloading plugins after receiving command"
-                    )
-                    install_plugins()
-                    load_plugins()
-                except Exception as e:
-                    print("plugin-synchronizer: `install_plugins` failed:", e)
-
-
 def load_or_reload_plugin(path: pathlib.Path) -> None:
     """Given a path, load or reload a plugin."""
     log.info(f"Loading {path}")

plugin_runner/tests/test_plugin_runner.py

@@ -1,4 +1,6 @@
+import asyncio
 import logging
+import pickle
 import shutil
 from collections.abc import Generator
 from pathlib import Path
@@ -15,6 +17,7 @@
     PluginRunner,
     load_or_reload_plugin,
     load_plugins,
+    synchronize_plugins,
 )
 
 
@@ -295,6 +298,37 @@ async def test_reload_plugins_event_handler_successfully_publishes_message(
     assert result[0].success is True
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("install_test_plugin", ["test_module_imports_plugin"], indirect=True)
+async def test_synchronize_plugins_calls_install_and_load_plugins(
+    install_test_plugin: Path, plugin_runner: PluginRunner
+) -> None:
+    """Test that synchronize_plugins calls install_plugins and load_plugins."""
+    with (
+        patch("plugin_runner.plugin_runner.get_client", new_callable=MagicMock) as mock_get_client,
+        patch(
+            "plugin_runner.plugin_runner.install_plugins", new_callable=AsyncMock
+        ) as mock_install_plugins,
+        patch(
+            "plugin_runner.plugin_runner.load_plugins", new_callable=AsyncMock
+        ) as mock_load_plugins,
+    ):
+        mock_client = AsyncMock()
+        mock_pubsub = AsyncMock()
+        mock_get_client.return_value = (mock_client, mock_pubsub)
+        mock_pubsub.get_message.return_value = {
+            "type": "pmessage",
+            "data": pickle.dumps({"action": "reload"}),
+        }
+
+        task = asyncio.create_task(synchronize_plugins(max_iterations=1))
+        await asyncio.sleep(0.1)  # Give some time for the coroutine to run
+        task.cancel()
+
+        mock_install_plugins.assert_called_once()
+        mock_load_plugins.assert_called_once()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("install_test_plugin", ["test_module_imports_plugin"], indirect=True)
 async def test_changes_to_plugin_modules_should_be_reflected_after_reload(