|
21 | 21 | sync_operation, |
22 | 22 | ) |
23 | 23 |
|
| 24 | +import temporalio.api.enums.v1 |
24 | 25 | from temporalio import nexus, workflow |
25 | 26 | from temporalio.client import ( |
26 | 27 | CancelNexusOperationInput, |
|
36 | 37 | OutboundInterceptor, |
37 | 38 | StartNexusOperationInput, |
38 | 39 | TerminateNexusOperationInput, |
| 40 | + WorkflowHistory, |
39 | 41 | WorkflowUpdateStage, |
40 | 42 | ) |
41 | 43 | from temporalio.common import ( |
|
56 | 58 | from temporalio.testing import WorkflowEnvironment |
57 | 59 | from temporalio.worker import Worker |
58 | 60 | from tests.helpers import assert_eventually |
59 | | -from tests.helpers.nexus import make_nexus_endpoint_name |
| 61 | +from tests.helpers.nexus import ( |
| 62 | + assert_links_match, |
| 63 | + expected_nexus_operation_link, |
| 64 | + expected_workflow_event_link, |
| 65 | + links_from_workflow_execution_started_event, |
| 66 | + make_nexus_endpoint_name, |
| 67 | +) |
60 | 68 |
|
61 | 69 | # --------------------------------------------------------------------------- |
62 | 70 | # Data types |
@@ -259,6 +267,61 @@ async def test_start_async_operation_and_poll_result( |
259 | 267 | assert result.value == "async-hello" |
260 | 268 |
|
261 | 269 |
|
| 270 | +async def test_started_workflow_has_link_to_standalone_nexus_operation( |
| 271 | + client: Client, env: WorkflowEnvironment |
| 272 | +): |
| 273 | + """Start a workflow_run operation and verify its workflow links back to the Nexus op.""" |
| 274 | + if env.supports_time_skipping: |
| 275 | + pytest.skip( |
| 276 | + "Standalone Nexus Operation tests don't work with time-skipping server" |
| 277 | + ) |
| 278 | + |
| 279 | + task_queue = str(uuid.uuid4()) |
| 280 | + endpoint_name = make_nexus_endpoint_name(task_queue) |
| 281 | + service_handler = StandaloneTestServiceHandler() |
| 282 | + |
| 283 | + async with Worker( |
| 284 | + client, |
| 285 | + task_queue=task_queue, |
| 286 | + nexus_service_handlers=[service_handler], |
| 287 | + workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], |
| 288 | + ): |
| 289 | + await env.create_nexus_endpoint(endpoint_name, task_queue) |
| 290 | + |
| 291 | + nexus_client = client.create_nexus_client( |
| 292 | + service=StandaloneTestService, endpoint=endpoint_name |
| 293 | + ) |
| 294 | + op_id = str(uuid.uuid4()) |
| 295 | + input_value = f"link-test-{uuid.uuid4()}" |
| 296 | + workflow_id = f"blocking_async-{input_value}" |
| 297 | + |
| 298 | + handle = await nexus_client.start_operation( |
| 299 | + StandaloneTestService.blocking_async, |
| 300 | + EchoInput(value=input_value), |
| 301 | + id=op_id, |
| 302 | + id_reuse_policy=NexusOperationIDReusePolicy.REJECT_DUPLICATE, |
| 303 | + id_conflict_policy=NexusOperationIDConflictPolicy.FAIL, |
| 304 | + schedule_to_close_timeout=timedelta(seconds=30), |
| 305 | + ) |
| 306 | + |
| 307 | + await service_handler.started_blocking.wait() |
| 308 | + workflow_history = await _assert_workflow_started_with_nexus_operation_link( |
| 309 | + client, workflow_id, handle |
| 310 | + ) |
| 311 | + await _assert_nexus_operation_has_link_to_started_workflow( |
| 312 | + client, workflow_history, handle |
| 313 | + ) |
| 314 | + |
| 315 | + workflow_handle = client.get_workflow_handle(workflow_id) |
| 316 | + await workflow_handle.start_update( |
| 317 | + BlockingHandlerWorkflow.unblock, |
| 318 | + wait_for_stage=WorkflowUpdateStage.COMPLETED, |
| 319 | + ) |
| 320 | + result = await handle.result() |
| 321 | + assert isinstance(result, EchoOutput) |
| 322 | + assert result.value == input_value |
| 323 | + |
| 324 | + |
262 | 325 | async def test_execute_operation(client: Client, env: WorkflowEnvironment): |
263 | 326 | """Use execute_operation convenience method, verify it returns result directly.""" |
264 | 327 | if env.supports_time_skipping: |
@@ -949,3 +1012,52 @@ async def test_interceptor_receives_inputs(client: Client, env: WorkflowEnvironm |
949 | 1012 | count_input = interceptor.count_calls[-1] |
950 | 1013 | assert isinstance(count_input, CountNexusOperationsInput) |
951 | 1014 | assert count_input.query == query |
| 1015 | + |
| 1016 | + |
| 1017 | +async def _assert_workflow_started_with_nexus_operation_link( |
| 1018 | + client: Client, |
| 1019 | + workflow_id: str, |
| 1020 | + operation_handle: NexusOperationHandle[Any], |
| 1021 | +) -> WorkflowHistory: |
| 1022 | + history = await client.get_workflow_handle(workflow_id).fetch_history() |
| 1023 | + started_event = next( |
| 1024 | + ( |
| 1025 | + e |
| 1026 | + for e in history.events |
| 1027 | + if ( |
| 1028 | + e.event_type |
| 1029 | + == temporalio.api.enums.v1.EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED |
| 1030 | + ) |
| 1031 | + ), |
| 1032 | + None, |
| 1033 | + ) |
| 1034 | + assert started_event is not None |
| 1035 | + |
| 1036 | + assert operation_handle.run_id is not None |
| 1037 | + assert_links_match( |
| 1038 | + links_from_workflow_execution_started_event(started_event), |
| 1039 | + expected_nexus_operation_link( |
| 1040 | + namespace=client.namespace, |
| 1041 | + operation_id=operation_handle.operation_id, |
| 1042 | + run_id=operation_handle.run_id, |
| 1043 | + ), |
| 1044 | + ) |
| 1045 | + return history |
| 1046 | + |
| 1047 | + |
| 1048 | +async def _assert_nexus_operation_has_link_to_started_workflow( |
| 1049 | + client: Client, |
| 1050 | + workflow_history: WorkflowHistory, |
| 1051 | + operation_handle: NexusOperationHandle[Any], |
| 1052 | +) -> None: |
| 1053 | + desc = await operation_handle.describe() |
| 1054 | + assert_links_match( |
| 1055 | + desc.raw_description.links, |
| 1056 | + expected_workflow_event_link( |
| 1057 | + namespace=client.namespace, |
| 1058 | + workflow_id=workflow_history.workflow_id, |
| 1059 | + run_id=workflow_history.run_id, |
| 1060 | + event_type=temporalio.api.enums.v1.EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, |
| 1061 | + event_id=workflow_history.events[0].event_id, |
| 1062 | + ), |
| 1063 | + ) |
0 commit comments