diff --git a/tests/helpers/__init__.py b/tests/helpers/__init__.py index fe37296e9..242bac1e6 100644 --- a/tests/helpers/__init__.py +++ b/tests/helpers/__init__.py @@ -239,7 +239,7 @@ async def check_workflow_exists() -> bool: async def assert_pending_activity_exists_eventually( handle: WorkflowHandle, activity_id: str, - timeout: timedelta = timedelta(seconds=5), + timeout: timedelta = timedelta(seconds=10), ) -> PendingActivityInfo: """Wait until a pending activity with the given ID exists and return it.""" @@ -351,7 +351,11 @@ async def check_paused() -> None: async def unpause_and_assert(client: Client, handle: WorkflowHandle, activity_id: str): - """Unpause the given activity and assert it is not paused.""" + """Unpause the given activity and assert it is no longer paused. + + An unpaused activity may retry and close before we observe it, so an + activity that is no longer pending also counts as unpaused. + """ desc = await handle.describe() req = UnpauseActivityRequest( namespace=client.namespace, @@ -363,10 +367,9 @@ async def unpause_and_assert(client: Client, handle: WorkflowHandle, activity_id ) await client.workflow_service.unpause_activity(req) - # Assert eventually not paused async def check_unpaused() -> None: - info = await assert_pending_activity_exists_eventually(handle, activity_id) - assert not info.paused, f"Activity {activity_id} still paused" + info = await get_pending_activity_info(handle, activity_id) + assert info is None or not info.paused, f"Activity {activity_id} still paused" await assert_eventually(check_unpaused) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index c4fe53271..d2a041e64 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -8941,7 +8941,6 @@ async def run( True, activity_id=activity_id, start_to_close_timeout=timedelta(seconds=10), - heartbeat_timeout=timedelta(seconds=2), retry_policy=RetryPolicy(maximum_attempts=1), ) ) @@ -8951,7 +8950,6 @@ async def run( True, activity_id=f"{activity_id}-2", start_to_close_timeout=timedelta(seconds=10), - heartbeat_timeout=timedelta(seconds=2), retry_policy=RetryPolicy(maximum_attempts=1), ) ) @@ -8970,6 +8968,8 @@ async def test_activity_pause_cancellation_details( workflows=[ActivityHeartbeatWorkflow], activities=[heartbeat_activity, sync_heartbeat_activity], activity_executor=executor, + max_heartbeat_throttle_interval=timedelta(milliseconds=300), + default_heartbeat_throttle_interval=timedelta(milliseconds=300), ) as worker: test_activity_id = f"heartbeat-activity-{uuid.uuid4()}" @@ -9022,7 +9022,6 @@ async def run( False, activity_id=activity_id, start_to_close_timeout=timedelta(seconds=10), - heartbeat_timeout=timedelta(seconds=1), retry_policy=RetryPolicy(maximum_attempts=2), ) ) @@ -9032,7 +9031,6 @@ async def run( False, activity_id=f"{activity_id}-2", start_to_close_timeout=timedelta(seconds=10), - heartbeat_timeout=timedelta(seconds=1), retry_policy=RetryPolicy(maximum_attempts=2), ) ) @@ -9130,7 +9128,6 @@ async def run(self, activity_id: str) -> None: external_activity_heartbeat, activity_id=activity_id, start_to_close_timeout=timedelta(seconds=10), - heartbeat_timeout=timedelta(seconds=1), retry_policy=RetryPolicy(maximum_attempts=2), ) @@ -9170,12 +9167,11 @@ async def test_external_activity_cancellation_details( # Pause activity then assert it is paused await pause_and_assert(client, wf_handle, activity_info.activity_id) - try: + with pytest.raises(AsyncActivityCancelledError) as err: await external_activity_handle.heartbeat() - except AsyncActivityCancelledError as err: - assert err.details == temporalio.activity.ActivityCancellationDetails( - paused=True - ) + assert err.value.details == temporalio.activity.ActivityCancellationDetails( + paused=True + ) @activity.defn