Test non-async-def sync operations

Author: dandavison

temporalio/nexus/handler.py

@@ -119,14 +119,17 @@ def to_workflow_handle(token: str, client: Client) -> WorkflowHandle[Any, O]:
         return client.get_workflow_handle(workflow_id, run_id=run_id)
 
 
+# TODO(dan): naming, visibility, make this less awkward
 def get_input_and_output_types_from_async_start_method(
     start_method: Callable[
         [S, I, nexusrpc.handler.StartOperationOptions],
         Awaitable[StartWorkflowOperationResult[O]],
     ],
 ) -> tuple[Type[I], Type[O]]:
     input_type, output_type = (
-        nexusrpc.handler.get_input_and_output_types_from_sync_start_method(start_method)
+        nexusrpc.handler.get_input_and_output_types_from_sync_operation_start_method(
+            start_method
+        )
     )
     origin_type = typing.get_origin(output_type)
     if not origin_type or not issubclass(origin_type, StartWorkflowOperationResult):

tests/worker/test_nexus.py

@@ -50,6 +50,7 @@ class OpDefinitionType(IntEnum):
 @dataclass
 class SyncResponse:
     op_definition_type: OpDefinitionType
+    use_async_def: bool
 
 
 @dataclass
@@ -86,6 +87,7 @@ class OpOutput:
 class ServiceInterface:
     sync_or_async_operation: nexusrpc.interface.Operation[OpInput, OpOutput]
     sync_operation: nexusrpc.interface.Operation[OpInput, OpOutput]
+    non_async_sync_operation: nexusrpc.interface.Operation[OpInput, OpOutput]
     async_operation: nexusrpc.interface.Operation[OpInput, OpOutput]
 
 
@@ -176,6 +178,16 @@ async def sync_operation(
             start_options_received_by_handler=options,
         )
 
+    @nexusrpc.handler.sync_operation
+    def non_async_sync_operation(
+        self, input: OpInput, options: nexusrpc.handler.StartOperationOptions
+    ) -> OpOutput:
+        assert isinstance(input.response_type, SyncResponse)
+        return OpOutput(
+            value="sync response",
+            start_options_received_by_handler=options,
+        )
+
     @temporalio.nexus.handler.workflow_run_operation
     async def async_operation(
         self, input: OpInput, options: nexusrpc.handler.StartOperationOptions
@@ -276,48 +288,73 @@ def _get_operation(
                 SyncResponse,
                 OpDefinitionType.SHORTHAND,
                 CallerReference.IMPL_WITH_INTERFACE,
+                True,
             ): ServiceImpl.sync_operation,
             (
                 SyncResponse,
                 OpDefinitionType.SHORTHAND,
                 CallerReference.INTERFACE,
+                True,
             ): ServiceInterface.sync_operation,
+            (
+                SyncResponse,
+                OpDefinitionType.SHORTHAND,
+                CallerReference.IMPL_WITH_INTERFACE,
+                False,
+            ): ServiceImpl.non_async_sync_operation,
+            (
+                SyncResponse,
+                OpDefinitionType.SHORTHAND,
+                CallerReference.INTERFACE,
+                False,
+            ): ServiceInterface.non_async_sync_operation,
             (
                 SyncResponse,
                 OpDefinitionType.LONGHAND,
                 CallerReference.IMPL_WITH_INTERFACE,
+                True,
             ): ServiceImpl.sync_or_async_operation,
             (
                 SyncResponse,
                 OpDefinitionType.LONGHAND,
                 CallerReference.INTERFACE,
+                True,
             ): ServiceInterface.sync_or_async_operation,
             (
                 AsyncResponse,
                 OpDefinitionType.SHORTHAND,
                 CallerReference.IMPL_WITH_INTERFACE,
+                True,
             ): ServiceImpl.async_operation,
             (
                 AsyncResponse,
                 OpDefinitionType.SHORTHAND,
                 CallerReference.INTERFACE,
+                True,
             ): ServiceInterface.async_operation,
             (
                 AsyncResponse,
                 OpDefinitionType.LONGHAND,
                 CallerReference.IMPL_WITH_INTERFACE,
+                True,
             ): ServiceImpl.sync_or_async_operation,
             (
                 AsyncResponse,
                 OpDefinitionType.LONGHAND,
                 CallerReference.INTERFACE,
+                True,
             ): ServiceInterface.sync_or_async_operation,
         }[
             {True: SyncResponse, False: AsyncResponse}[
                 isinstance(op_input.response_type, SyncResponse)
             ],
             op_input.response_type.op_definition_type,
             op_input.caller_reference,
+            (
+                op_input.response_type.use_async_def
+                if isinstance(op_input.response_type, SyncResponse)
+                else True
+            ),
         ]
 
 
@@ -400,7 +437,7 @@ async def test_sync_response(
             args=[
                 CallerWfInput(
                     op_input=OpInput(
-                        response_type=SyncResponse(op_definition_type),
+                        response_type=SyncResponse(op_definition_type, True),
                         start_options=nexusrpc.handler.StartOperationOptions(
                             headers={"header-key": "header-value"},
                         ),
@@ -579,7 +616,7 @@ async def test_untyped_caller(
             args=[
                 CallerWfInput(
                     op_input=OpInput(
-                        response_type=SyncResponse(op_definition_type),
+                        response_type=SyncResponse(op_definition_type, True),
                         start_options=nexusrpc.handler.StartOperationOptions(),
                         caller_reference=caller_reference,
                     ),