Activate nexus worker in worker tests

Author: dandavison

tests/worker/test_worker.py

@@ -7,8 +7,10 @@
 from typing import Any, Awaitable, Callable, Optional, Sequence
 from urllib.request import urlopen
 
+import nexusrpc
+
 import temporalio.api.enums.v1
-import temporalio.client
+import temporalio.nexus
 import temporalio.worker._worker
 from temporalio import activity, workflow
 from temporalio.api.workflowservice.v1 import (
@@ -80,6 +82,21 @@ async def run(self) -> None:
         raise NotImplementedError
 
 
+@nexusrpc.handler.service_handler
+class NeverRunService:
+    @nexusrpc.handler.sync_operation
+    async def never_run_operation(
+        self, _ctx: nexusrpc.handler.StartOperationContext, _input: None
+    ) -> None:
+        raise NotImplementedError
+
+    @temporalio.nexus.workflow_run_operation
+    async def never_run_workflow_run_operation(
+        self, _ctx: temporalio.nexus.WorkflowRunOperationContext, _input: None
+    ) -> temporalio.nexus.WorkflowHandle[None]:
+        raise NotImplementedError
+
+
 async def test_worker_fatal_error_run(client: Client):
     # Run worker with injected workflow poll error
     worker = create_worker(client)
@@ -1143,6 +1160,7 @@ def create_worker(
         task_queue=f"task-queue-{uuid.uuid4()}",
         activities=[never_run_activity],
         workflows=[NeverRunWorkflow],
+        nexus_service_handlers=[NeverRunService()],
         on_fatal_error=on_fatal_error,
     )