Fix tests

Author: dandavison

tests/worker/test_nexus.py

@@ -287,10 +287,15 @@ async def run(
             headers=op_input.start_options.headers,
         )
         self._nexus_operation_started = True
-        if isinstance(input.op_input.response_type, SyncResponse):
-            assert op_handle.operation_token is None
-        else:
-            assert op_handle.operation_token
+        if not input.op_input.response_type.exception_in_operation_start:
+            if isinstance(input.op_input.response_type, SyncResponse):
+                assert (
+                    op_handle.operation_token is None
+                ), "operation_token should be absent after a sync response"
+            else:
+                assert (
+                    op_handle.operation_token
+                ), "operation_token should be present after an async response"
 
         if request_cancel:
             # Even for SyncResponse, the op_handle future is not done at this point; that
@@ -561,19 +566,6 @@ async def test_async_response(
             op_definition_type,
             caller_reference,
         )
-        # TODO(dan): race here? How do we know it hasn't been canceled already?
-        handler_wf_info = await handler_wf_handle.describe()
-        assert handler_wf_info.status in [
-            WorkflowExecutionStatus.RUNNING,
-            WorkflowExecutionStatus.COMPLETED,
-        ]
-        await assert_caller_workflow_has_link_to_handler_workflow(
-            caller_wf_handle, handler_wf_handle, handler_wf_info.run_id
-        )
-        await assert_handler_workflow_has_link_to_caller_workflow(
-            caller_wf_handle, handler_wf_handle
-        )
-
         if exception_in_operation_start:
             with pytest.raises(WorkflowFailureError) as ei:
                 await caller_wf_handle.result()
@@ -590,7 +582,22 @@ async def test_async_response(
                 if op_definition_type == OpDefinitionType.SHORTHAND
                 else "sync_or_async_operation"
             )
-        elif request_cancel:
+            return
+
+        # TODO(dan): race here? How do we know it hasn't been canceled already?
+        handler_wf_info = await handler_wf_handle.describe()
+        assert handler_wf_info.status in [
+            WorkflowExecutionStatus.RUNNING,
+            WorkflowExecutionStatus.COMPLETED,
+        ]
+        await assert_caller_workflow_has_link_to_handler_workflow(
+            caller_wf_handle, handler_wf_handle, handler_wf_info.run_id
+        )
+        await assert_handler_workflow_has_link_to_caller_workflow(
+            caller_wf_handle, handler_wf_handle
+        )
+
+        if request_cancel:
             # The operation response was asynchronous and so request_cancel is honored. See
             # explanation below.
             with pytest.raises(WorkflowFailureError) as ei:
@@ -638,7 +645,7 @@ async def _start_wf_and_nexus_op(
     """
     await create_nexus_endpoint(task_queue, client)
     # TODO(dan): this workflow ID is for xray; change to uuid before merge
-    operation_workflow_id = "default-workflow-id"
+    operation_workflow_id = str(uuid.uuid4())
 
     # Start the caller workflow and wait until it confirms the Nexus operation has started.
     block_forever_waiting_for_cancellation = request_cancel