Some PR updates, refactoring tests, added name

Author: tconley1428

temporalio/client.py

@@ -228,11 +228,14 @@ def __init__(
         for plugin in reversed(list(plugins)):
             root_plugin = plugin.init_client_plugin(root_plugin)
 
-        self._config = root_plugin.on_create_client(config)
+        self._init_from_config(root_plugin.on_create_client(config))
+
+    def _init_from_config(self, config: ClientConfig):
+        self._config = config
 
         # Iterate over interceptors in reverse building the impl
         self._impl: OutboundInterceptor = _ClientImpl(self)
-        for interceptor in reversed(list(interceptors)):
+        for interceptor in reversed(list(self._config["interceptors"])):
             self._impl = interceptor.intercept_client(self._impl)
 
     def config(self) -> ClientConfig:
@@ -7388,6 +7391,9 @@ async def _decode_user_metadata(
 
 
 class Plugin:
+    def name(self) -> str:
+        return type(self).__module__ + "." + type(self).__qualname__
+
     def init_client_plugin(self, next: Plugin) -> Plugin:
         self.next_client_plugin = next
         return self

temporalio/worker/_worker.py

@@ -97,6 +97,9 @@ def _to_bridge(self) -> temporalio.bridge.worker.PollerBehavior:
 
 
 class Plugin:
+    def name(self) -> str:
+        return type(self).__module__ + "." + type(self).__qualname__
+
     def init_worker_plugin(self, next: Plugin) -> Plugin:
         self.next_worker_plugin = next
         return self
@@ -388,6 +391,11 @@ def __init__(
             List[Plugin],
             [p for p in client.config()["plugins"] if isinstance(p, Plugin)],
         )
+        for client_plugin in plugins_from_client:
+            if type(client_plugin) in [type(p) for p in plugins]:
+                warnings.warn(
+                    f"The same plugin type {type(client_plugin)} is present from both client and worker. It may run twice and may not be the intended behavior."
+                )
         plugins = plugins_from_client + list(plugins)
 
         root_plugin: Plugin = _RootPlugin()

tests/test_client.py

@@ -1501,30 +1501,3 @@ async def test_cloud_client_simple():
         GetNamespaceRequest(namespace=os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"])
     )
     assert os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"] == result.namespace.namespace
-
-
-class MyPlugin(Plugin):
-    def on_create_client(self, config: ClientConfig) -> ClientConfig:
-        config["namespace"] = "replaced_namespace"
-        return super().on_create_client(config)
-
-    async def connect_service_client(
-        self, config: temporalio.service.ConnectConfig
-    ) -> temporalio.service.ServiceClient:
-        config.api_key = "replaced key"
-        return await super().connect_service_client(config)
-
-
-async def test_client_plugin(client: Client, env: WorkflowEnvironment):
-    if env.supports_time_skipping:
-        pytest.skip("Client connect is only designed for local")
-
-    config = client.config()
-    config["plugins"] = [MyPlugin()]
-    new_client = Client(**config)
-    assert new_client.namespace == "replaced_namespace"
-
-    new_client = await Client.connect(
-        client.service_client.config.target_host, plugins=[MyPlugin()]
-    )
-    assert new_client.service_client.config.api_key == "replaced key"

tests/test_plugins.py

@@ -0,0 +1,112 @@
+import warnings
+
+import pytest
+
+import temporalio.client
+import temporalio.worker
+from temporalio.client import Client, ClientConfig, OutboundInterceptor, Plugin
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker, WorkerConfig
+from tests.worker.test_worker import never_run_activity
+
+
+class TestClientInterceptor(temporalio.client.Interceptor):
+    intercepted = False
+
+    def intercept_client(self, next: OutboundInterceptor) -> OutboundInterceptor:
+        self.intercepted = True
+        return super().intercept_client(next)
+
+
+class MyClientPlugin(Plugin):
+    def __init__(self):
+        self.interceptor = TestClientInterceptor()
+
+    def on_create_client(self, config: ClientConfig) -> ClientConfig:
+        config["namespace"] = "replaced_namespace"
+        config["interceptors"] = list(config.get("interceptors") or []) + [
+            self.interceptor
+        ]
+        return super().on_create_client(config)
+
+    async def connect_service_client(
+        self, config: temporalio.service.ConnectConfig
+    ) -> temporalio.service.ServiceClient:
+        config.api_key = "replaced key"
+        return await super().connect_service_client(config)
+
+
+async def test_client_plugin(client: Client, env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip("Client connect is only designed for local")
+
+    plugin = MyClientPlugin()
+    config = client.config()
+    config["plugins"] = [plugin]
+    new_client = Client(**config)
+    assert new_client.namespace == "replaced_namespace"
+    assert plugin.interceptor.intercepted
+    assert plugin.name() == "tests.test_plugins.MyClientPlugin"
+
+    new_client = await Client.connect(
+        client.service_client.config.target_host, plugins=[MyClientPlugin()]
+    )
+    assert new_client.service_client.config.api_key == "replaced key"
+
+
+class MyCombinedPlugin(temporalio.client.Plugin, temporalio.worker.Plugin):
+    def on_create_worker(self, config: WorkerConfig) -> WorkerConfig:
+        config["task_queue"] = "combined"
+        return super().on_create_worker(config)
+
+
+class MyWorkerPlugin(temporalio.worker.Plugin):
+    def on_create_worker(self, config: WorkerConfig) -> WorkerConfig:
+        config["task_queue"] = "replaced_queue"
+        return super().on_create_worker(config)
+
+    async def run_worker(self, worker: Worker) -> None:
+        await super().run_worker(worker)
+
+
+async def test_worker_plugin_basic_config(client: Client) -> None:
+    worker = Worker(
+        client,
+        task_queue="queue",
+        activities=[never_run_activity],
+        plugins=[MyWorkerPlugin()],
+    )
+    assert worker.config().get("task_queue") == "replaced_queue"
+
+    # Test client plugin propagation to worker plugins
+    new_config = client.config()
+    new_config["plugins"] = [MyCombinedPlugin()]
+    client = Client(**new_config)
+    worker = Worker(client, task_queue="queue", activities=[never_run_activity])
+    assert worker.config().get("task_queue") == "combined"
+
+    # Test both. Client propagated plugins are called first, so the worker plugin overrides in this case
+    worker = Worker(
+        client,
+        task_queue="queue",
+        activities=[never_run_activity],
+        plugins=[MyWorkerPlugin()],
+    )
+    assert worker.config().get("task_queue") == "replaced_queue"
+
+
+async def test_worker_duplicated_plugin(client: Client) -> None:
+    new_config = client.config()
+    new_config["plugins"] = [MyCombinedPlugin()]
+    client = Client(**new_config)
+
+    with warnings.catch_warnings(record=True) as warning_list:
+        worker = Worker(
+            client,
+            task_queue="queue",
+            activities=[never_run_activity],
+            plugins=[MyCombinedPlugin()],
+        )
+
+    assert len(warning_list) == 1
+    assert "The same plugin type" in str(warning_list[0].message)

tests/worker/test_worker.py

@@ -1189,46 +1189,3 @@ def shutdown(self) -> None:
         if self.next_exception_task:
             self.next_exception_task.cancel()
         setattr(self.worker._bridge_worker, self.attr, self.orig_poll_call)
-
-
-class MyCombinedPlugin(temporalio.client.Plugin, temporalio.worker.Plugin):
-    def on_create_worker(self, config: WorkerConfig) -> WorkerConfig:
-        print("Create worker combined plugin")
-        config["task_queue"] = "combined"
-        return super().on_create_worker(config)
-
-
-class MyWorkerPlugin(temporalio.worker.Plugin):
-    def on_create_worker(self, config: WorkerConfig) -> WorkerConfig:
-        print("Create worker worker plugin")
-        config["task_queue"] = "replaced_queue"
-        return super().on_create_worker(config)
-
-    async def run_worker(self, worker: Worker) -> None:
-        await super().run_worker(worker)
-
-
-async def test_worker_plugin(client: Client) -> None:
-    worker = Worker(
-        client,
-        task_queue="queue",
-        activities=[never_run_activity],
-        plugins=[MyWorkerPlugin()],
-    )
-    assert worker.config().get("task_queue") == "replaced_queue"
-
-    # Test client plugin propagation to worker plugins
-    new_config = client.config()
-    new_config["plugins"] = [MyCombinedPlugin()]
-    client = Client(**new_config)
-    worker = Worker(client, task_queue="queue", activities=[never_run_activity])
-    assert worker.config().get("task_queue") == "combined"
-
-    # Test both. Client propagated plugins are called first, so the worker plugin overrides in this case
-    worker = Worker(
-        client,
-        task_queue="queue",
-        activities=[never_run_activity],
-        plugins=[MyWorkerPlugin()],
-    )
-    assert worker.config().get("task_queue") == "replaced_queue"