Add tests for update-with-start reattach behavior (#1000)

dandavison · web-flow · commit c8bc329f7630 · 2025-07-28T13:54:25.000-04:00
* Set test server version back to `default`

* Improve docstrings

* Remove more experimental notices

* Test UwS attach behaviors

* Use an enum: Literal doesn't seem to pass mypy

* Skip under JTS
diff --git a/temporalio/client.py b/temporalio/client.py
@@ -528,11 +528,12 @@ async def start_workflow(
                 retries and continue as new.
             run_timeout: Timeout of a single workflow run.
             task_timeout: Timeout of a single workflow task.
-            id_reuse_policy: How already-existing IDs are treated.
-            id_conflict_policy: How already-running workflows of the same ID are
-                treated. Default is unspecified which effectively means fail the
-                start attempt. This cannot be set if ``id_reuse_policy`` is set
-                to terminate if running.
+            id_conflict_policy: Behavior when a workflow is currently running with the same ID.
+                Default is UNSPECIFIED, which effectively means fail the start attempt.
+                Set to USE_EXISTING for idempotent deduplication on workflow ID.
+                Cannot be set if ``id_reuse_policy`` is set to TERMINATE_IF_RUNNING.
+            id_reuse_policy: Behavior when a closed workflow with the same ID exists.
+                Default is ALLOW_DUPLICATE.
             retry_policy: Retry policy for the workflow.
             cron_schedule: See https://docs.temporal.io/docs/content/what-is-a-temporal-cron-job/
             memo: Memo for the workflow.
@@ -2487,9 +2488,6 @@ class WithStartWorkflowOperation(Generic[SelfType, ReturnType]):
 
     Update-With-Start allows you to send an update to a workflow, while starting the
     workflow if necessary.
-
-    .. warning::
-        This API is experimental
     """
 
     # Overload for no-param workflow, with_start
@@ -2657,9 +2655,6 @@ def __init__(
     ) -> None:
         """Create a WithStartWorkflowOperation.
 
-        .. warning::
-           This API is experimental
-
         See :py:meth:`temporalio.client.Client.start_workflow` for documentation of the
         arguments.
         """
@@ -2700,11 +2695,7 @@ def __init__(
         self._used = False
 
     async def workflow_handle(self) -> WorkflowHandle[SelfType, ReturnType]:
-        """Wait until workflow is running and return a WorkflowHandle.
-
-        .. warning::
-           This API is experimental
-        """
+        """Wait until workflow is running and return a WorkflowHandle."""
         return await self._workflow_handle
 
 
diff --git a/tests/__init__.py b/tests/__init__.py
@@ -1,2 +1 @@
-# TODO: Change back to "default" after next CLI release
-DEV_SERVER_DOWNLOAD_VERSION = "v1.3.1-persistence-fix.0"
+DEV_SERVER_DOWNLOAD_VERSION = "default"
diff --git a/tests/worker/test_update_with_start.py b/tests/worker/test_update_with_start.py
@@ -1,10 +1,11 @@
 from __future__ import annotations
 
+import asyncio
 import uuid
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import timedelta
-from enum import Enum
+from enum import Enum, IntEnum
 from typing import Any, Iterator, Mapping, Optional
 from unittest.mock import patch
 
@@ -17,12 +18,14 @@
     OutboundInterceptor,
     StartWorkflowUpdateWithStartInput,
     WithStartWorkflowOperation,
+    WorkflowExecutionStatus,
     WorkflowUpdateFailedError,
     WorkflowUpdateHandle,
     WorkflowUpdateStage,
 )
 from temporalio.common import (
     WorkflowIDConflictPolicy,
+    WorkflowIDReusePolicy,
 )
 from temporalio.exceptions import ApplicationError, WorkflowAlreadyStartedError
 from temporalio.service import RPCError, RPCStatusCode, ServiceCall
@@ -859,3 +862,266 @@ async def __call__(
         assert err.value.status == RPCStatusCode.INTERNAL
         assert err.value.message == "empty details"
         assert len(err.value.grpc_status.details) == 0
+
+
+class ExecutionBehavior(IntEnum):
+    COMPLETES = 0
+    BLOCKS = 1
+
+
+@workflow.defn
+class WorkflowWithUpdate:
+    def __init__(self) -> None:
+        self._unblock_workflow = asyncio.Event()
+        self._unblock_update = asyncio.Event()
+
+    @workflow.run
+    async def run(self, behavior: ExecutionBehavior) -> str:
+        if behavior == ExecutionBehavior.BLOCKS:
+            await self._unblock_workflow.wait()
+        return str(workflow.uuid4())
+
+    @workflow.update(unfinished_policy=workflow.HandlerUnfinishedPolicy.ABANDON)
+    async def update(self, behavior: ExecutionBehavior) -> str:
+        if behavior == ExecutionBehavior.BLOCKS:
+            await self._unblock_update.wait()
+        return str(workflow.uuid4())
+
+    @workflow.signal
+    async def unblock_workflow(self):
+        self._unblock_workflow.set()
+
+    @workflow.signal
+    async def unblock_update(self):
+        self._unblock_update.set()
+
+
+@pytest.mark.parametrize(
+    "workflow_behavior_name",
+    [ExecutionBehavior.COMPLETES.name, ExecutionBehavior.BLOCKS.name],
+)
+@pytest.mark.parametrize(
+    "id_conflict_policy_name",
+    [
+        WorkflowIDConflictPolicy.USE_EXISTING.name,
+        WorkflowIDConflictPolicy.FAIL.name,
+    ],
+)
+@pytest.mark.parametrize(
+    "id_reuse_policy_name",
+    [
+        WorkflowIDReusePolicy.ALLOW_DUPLICATE.name,
+        WorkflowIDReusePolicy.REJECT_DUPLICATE.name,
+    ],
+)
+async def test_update_with_start_always_attaches_to_completed_update(
+    env: WorkflowEnvironment,
+    workflow_behavior_name: str,
+    id_conflict_policy_name: str,
+    id_reuse_policy_name: str,
+):
+    """
+    A workflow exists and contains a completed update. An update-with-start sent for that workflow ID and that
+    update ID attaches to the update if workflow is running. If the workflow is closed then it attaches iff
+    the update is completed. The behavior is unaffected by the conflict policy or id reuse policy (so, for
+    example, we attach to an update in an existing workflow even if the conflict policy is FAIL).
+    """
+    if env.supports_time_skipping:
+        pytest.skip("TODO: make update_with_start tests pass under Java test server")
+    client = env.client
+    id_conflict_policy = WorkflowIDConflictPolicy[id_conflict_policy_name]
+    id_reuse_policy = WorkflowIDReusePolicy[id_reuse_policy_name]
+    workflow_behavior = ExecutionBehavior[workflow_behavior_name]
+    shared_workflow_id = f"workflow-id-{uuid.uuid4()}"
+    shared_update_id = f"update-id-{uuid.uuid4()}"
+    async with new_worker(client, WorkflowWithUpdate) as worker:
+
+        def start_op():
+            return WithStartWorkflowOperation(
+                WorkflowWithUpdate.run,
+                workflow_behavior,
+                id=shared_workflow_id,
+                task_queue=worker.task_queue,
+                id_conflict_policy=id_conflict_policy,
+                id_reuse_policy=id_reuse_policy,
+            )
+
+        start_op_1 = start_op()
+        update_result_1 = await client.execute_update_with_start_workflow(
+            WorkflowWithUpdate.update,
+            ExecutionBehavior.COMPLETES,
+            id=shared_update_id,
+            start_workflow_operation=start_op_1,
+        )
+        wf_handle_1 = await start_op_1.workflow_handle()
+        assert (await wf_handle_1.describe()).status == (
+            WorkflowExecutionStatus.COMPLETED
+            if workflow_behavior == ExecutionBehavior.COMPLETES
+            else WorkflowExecutionStatus.RUNNING
+        )
+
+        # Whether or not the workflow closed, the update exists in the last workflow run and is completed, so
+        # we attach to it.
+
+        start_op_2 = start_op()
+        update_result_2 = await client.execute_update_with_start_workflow(
+            WorkflowWithUpdate.update,
+            ExecutionBehavior.COMPLETES,
+            id=shared_update_id,
+            start_workflow_operation=start_op_2,
+        )
+        wf_handle_2 = await start_op_2.workflow_handle()
+        assert wf_handle_1.first_execution_run_id == wf_handle_2.first_execution_run_id
+        assert update_result_1 == update_result_2
+
+
+@pytest.mark.parametrize(
+    "id_conflict_policy_name",
+    [
+        WorkflowIDConflictPolicy.USE_EXISTING.name,
+        WorkflowIDConflictPolicy.FAIL.name,
+    ],
+)
+@pytest.mark.parametrize(
+    "id_reuse_policy_name",
+    [
+        WorkflowIDReusePolicy.ALLOW_DUPLICATE.name,
+        WorkflowIDReusePolicy.REJECT_DUPLICATE.name,
+    ],
+)
+async def test_update_with_start_attaches_to_non_completed_update_in_running_workflow(
+    env: WorkflowEnvironment,
+    id_conflict_policy_name: str,
+    id_reuse_policy_name: str,
+):
+    """
+    A workflow exists and is running and contains a non-completed update. An update-with-start sent for that
+    workflow ID and that update ID attaches to the update. The behavior is unaffected by the conflict policy
+    or id reuse policy (so, for example, we attach to the update in an existing workflow even if the conflict
+    policy is FAIL).
+    """
+    if env.supports_time_skipping:
+        pytest.skip("TODO: make update_with_start tests pass under Java test server")
+    client = env.client
+    id_conflict_policy = WorkflowIDConflictPolicy[id_conflict_policy_name]
+    id_reuse_policy = WorkflowIDReusePolicy[id_reuse_policy_name]
+    shared_workflow_id = f"workflow-id-{uuid.uuid4()}"
+    shared_update_id = f"update-id-{uuid.uuid4()}"
+    async with new_worker(client, WorkflowWithUpdate) as worker:
+
+        def start_op():
+            return WithStartWorkflowOperation(
+                WorkflowWithUpdate.run,
+                ExecutionBehavior.BLOCKS,
+                id=shared_workflow_id,
+                task_queue=worker.task_queue,
+                id_conflict_policy=id_conflict_policy,
+                id_reuse_policy=id_reuse_policy,
+            )
+
+        start_op_1 = start_op()
+        update_handle_1 = await client.start_update_with_start_workflow(
+            WorkflowWithUpdate.update,
+            ExecutionBehavior.BLOCKS,
+            id=shared_update_id,
+            start_workflow_operation=start_op_1,
+            wait_for_stage=WorkflowUpdateStage.ACCEPTED,
+        )
+        wf_handle_1 = await start_op_1.workflow_handle()
+        assert (await wf_handle_1.describe()).status == WorkflowExecutionStatus.RUNNING
+
+        # The workflow is running with the update not-completed. We will attach to the update.
+
+        start_op_2 = start_op()
+
+        update_handle_2 = await client.start_update_with_start_workflow(
+            WorkflowWithUpdate.update,
+            ExecutionBehavior.COMPLETES,
+            id=shared_update_id,
+            start_workflow_operation=start_op_2,
+            wait_for_stage=WorkflowUpdateStage.ACCEPTED,
+        )
+        wf_handle_2 = await start_op_2.workflow_handle()
+        assert wf_handle_1.first_execution_run_id == wf_handle_2.first_execution_run_id
+        await wf_handle_1.signal(WorkflowWithUpdate.unblock_update)
+        assert (await update_handle_1.result()) == (await update_handle_2.result())
+
+
+@pytest.mark.parametrize(
+    "id_conflict_policy_name",
+    [
+        WorkflowIDConflictPolicy.USE_EXISTING.name,
+        WorkflowIDConflictPolicy.FAIL.name,
+    ],
+)
+@pytest.mark.parametrize(
+    "id_reuse_policy_name",
+    [
+        WorkflowIDReusePolicy.ALLOW_DUPLICATE.name,
+        WorkflowIDReusePolicy.REJECT_DUPLICATE.name,
+    ],
+)
+async def test_update_with_start_does_not_attach_to_non_completed_update_in_closed_workflow(
+    env: WorkflowEnvironment,
+    id_conflict_policy_name: str,
+    id_reuse_policy_name: str,
+):
+    """
+    A workflow exists but is closed and contains a non-completed update. An update-with-start sent for that workflow
+    ID and that update ID does not attach to the update. If the id reuse policy is ALLOW_DUPLICATE then a new
+    workflow is started and the update is issued.
+    """
+    if env.supports_time_skipping:
+        pytest.skip("TODO: make update_with_start tests pass under Java test server")
+    client = env.client
+    id_conflict_policy = WorkflowIDConflictPolicy[id_conflict_policy_name]
+    id_reuse_policy = WorkflowIDReusePolicy[id_reuse_policy_name]
+    shared_workflow_id = f"workflow-id-{uuid.uuid4()}"
+    shared_update_id = f"update-id-{uuid.uuid4()}"
+    async with new_worker(client, WorkflowWithUpdate) as worker:
+
+        def start_op():
+            return WithStartWorkflowOperation(
+                WorkflowWithUpdate.run,
+                ExecutionBehavior.COMPLETES,
+                id=shared_workflow_id,
+                task_queue=worker.task_queue,
+                id_conflict_policy=id_conflict_policy,
+                id_reuse_policy=id_reuse_policy,
+            )
+
+        start_op_1 = start_op()
+        await client.start_update_with_start_workflow(
+            WorkflowWithUpdate.update,
+            ExecutionBehavior.BLOCKS,
+            id=shared_update_id,
+            start_workflow_operation=start_op_1,
+            wait_for_stage=WorkflowUpdateStage.ACCEPTED,
+        )
+        wf_handle_1 = await start_op_1.workflow_handle()
+        assert (
+            await wf_handle_1.describe()
+        ).status == WorkflowExecutionStatus.COMPLETED
+
+        # The workflow closed with the update not-completed. We will start a new workflow and issue the update
+        # iff reuse_policy is ALLOW_DUPLICATE. Conflict policy is irrelevant.
+
+        start_op_2 = start_op()
+
+        async def _do_update() -> Any:
+            return await client.execute_update_with_start_workflow(
+                WorkflowWithUpdate.update,
+                ExecutionBehavior.COMPLETES,
+                id=shared_update_id,
+                start_workflow_operation=start_op_2,
+            )
+
+        if id_reuse_policy == WorkflowIDReusePolicy.ALLOW_DUPLICATE:
+            await _do_update()
+            wf_handle_2 = await start_op_2.workflow_handle()
+            assert (
+                wf_handle_1.first_execution_run_id != wf_handle_2.first_execution_run_id
+            )
+        elif id_reuse_policy == WorkflowIDReusePolicy.REJECT_DUPLICATE:
+            with pytest.raises(WorkflowAlreadyStartedError):
+                await _do_update()

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1 @@`
`1`		`-# TODO: Change back to "default" after next CLI release`
`2`		`-DEV_SERVER_DOWNLOAD_VERSION = "v1.3.1-persistence-fix.0"`
	`1`	`+DEV_SERVER_DOWNLOAD_VERSION = "default"`