Test execute_operation w/ output type and async untyped

Author: dandavison

tests/worker/test_nexus.py

@@ -384,13 +384,24 @@ async def run(
             op_name = "sync_operation"
         else:
             raise TypeError
-        op_handle = await self.nexus_service.start_operation(
-            op_name,
-            op_input,
-            headers=op_input.start_options.headers,
-            output_type=OpOutput,
-        )
-        op_output = await op_handle
+
+        arbitrary_condition = isinstance(op_input.response_type, SyncResponse)
+
+        if arbitrary_condition:
+            op_handle = await self.nexus_service.start_operation(
+                op_name,
+                op_input,
+                headers=op_input.start_options.headers,
+                output_type=OpOutput,
+            )
+            op_output = await op_handle
+        else:
+            op_output = await self.nexus_service.execute_operation(
+                op_name,
+                op_input,
+                headers=op_input.start_options.headers,
+                output_type=OpOutput,
+            )
         return CallerWfOutput(
             op_output=OpOutput(
                 value=op_output.value,
@@ -596,27 +607,37 @@ async def _start_wf_and_nexus_op(
     "caller_reference",
     [CallerReference.IMPL_WITH_INTERFACE, CallerReference.INTERFACE],
 )
+@pytest.mark.parametrize("response_type", [SyncResponse, AsyncResponse])
 async def test_untyped_caller(
     client: Client,
     op_definition_type: OpDefinitionType,
     caller_reference: CallerReference,
+    response_type: ResponseType,
 ):
     task_queue = str(uuid.uuid4())
     async with Worker(
         client,
-        workflows=[UntypedCallerWorkflow],
+        workflows=[UntypedCallerWorkflow, HandlerWorkflow],
         nexus_services=[ServiceImpl()],
         task_queue=task_queue,
         workflow_runner=UnsandboxedWorkflowRunner(),
         workflow_failure_exception_types=[Exception],
     ):
+        if response_type == SyncResponse:
+            response_type = SyncResponse(op_definition_type, True)
+        else:
+            response_type = AsyncResponse(
+                str(uuid.uuid4()),
+                False,
+                op_definition_type,
+            )
         await create_nexus_endpoint(task_queue, client)
         caller_wf_handle = await client.start_workflow(
             UntypedCallerWorkflow.run,
             args=[
                 CallerWfInput(
                     op_input=OpInput(
-                        response_type=SyncResponse(op_definition_type, True),
+                        response_type=response_type,
                         start_options=nexusrpc.handler.StartOperationOptions(),
                         caller_reference=caller_reference,
                     ),
@@ -628,7 +649,11 @@ async def test_untyped_caller(
             task_queue=task_queue,
         )
         result = await caller_wf_handle.result()
-        assert result.op_output.value == "sync response"
+        assert result.op_output.value == (
+            "sync response"
+            if isinstance(response_type, SyncResponse)
+            else "workflow result"
+        )
         assert result.op_output.start_options_received_by_handler