Skip to content

Take activity tests off one-second clocks #1842

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
DABH wants to merge 4 commits into
base: main
Choose a base branch
from
+21 −13
Draft

Take activity tests off one-second clocks #1842

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4649730
Take test_workflow_cancel_activity off one-second clocks
DABH Sep 10, 2026
9db8b6c
Give manual activity tests time for the start handshake
DABH Sep 10, 2026
cc5be34
Share one cancel-wait helper in test_workflow_cancel_activity
DABH Sep 10, 2026
3aab22e
Merge remote-tracking branch 'origin/main' into flake/cancel-activity…
DABH Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions tests/test_activity.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -752,7 +752,7 @@ async def test_manual_completion(client: Client, env: WorkflowEnvironment):
ActivityInput(event_workflow_id=event_workflow_id),
id=activity_id,
task_queue=task_queue,
start_to_close_timeout=timedelta(seconds=5),
start_to_close_timeout=timedelta(minutes=1),
)

async with Worker(
Expand Down Expand Up @@ -794,7 +794,7 @@ async def test_manual_cancellation(client: Client, env: WorkflowEnvironment):
ActivityInput(event_workflow_id=event_workflow_id),
id=activity_id,
task_queue=task_queue,
start_to_close_timeout=timedelta(seconds=5),
start_to_close_timeout=timedelta(minutes=1),
)

async with Worker(
Expand Down Expand Up @@ -855,7 +855,7 @@ async def test_manual_failure(client: Client, env: WorkflowEnvironment):
ActivityInput(event_workflow_id=event_workflow_id),
id=activity_id,
task_queue=task_queue,
start_to_close_timeout=timedelta(seconds=5),
start_to_close_timeout=timedelta(minutes=1),
)
async with Worker(
client,
Expand Down Expand Up @@ -932,7 +932,7 @@ async def test_manual_heartbeat(client: Client, env: WorkflowEnvironment):
),
id=activity_id,
task_queue=task_queue,
start_to_close_timeout=timedelta(seconds=5),
start_to_close_timeout=timedelta(minutes=1),
)
wait_for_activity_start_wf_handle = await client.start_workflow(
EventWorkflow.wait,
Expand Down
26 changes: 17 additions & 9 deletions tests/worker/test_workflow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -946,7 +946,7 @@ async def run(self, params: CancelActivityWorkflowParams) -> None:
if params.local:
handle = workflow.start_local_activity_method(
ActivityWaitCancelNotify.wait_cancel,
schedule_to_close_timeout=timedelta(seconds=5),
schedule_to_close_timeout=timedelta(minutes=1),
cancellation_type=workflow.ActivityCancellationType[
params.cancellation_type
],
Expand All @@ -955,7 +955,7 @@ async def run(self, params: CancelActivityWorkflowParams) -> None:
handle = workflow.start_activity_method(
ActivityWaitCancelNotify.wait_cancel,
schedule_to_close_timeout=timedelta(seconds=5),
heartbeat_timeout=timedelta(seconds=1),
heartbeat_timeout=timedelta(seconds=5),
cancellation_type=workflow.ActivityCancellationType[
params.cancellation_type
],
Expand All @@ -976,14 +976,22 @@ def activity_result(self) -> str:

@pytest.mark.parametrize("local", [True, False])
async def test_workflow_cancel_activity(client: Client, local: bool):
# Need short task timeout to timeout LA task and longer assert timeout
# so the task can timeout
task_timeout = timedelta(seconds=1)
# Core completes the task holding a local activity at 80% of this timeout
task_timeout = timedelta(seconds=5)
assert_timeout = timedelta(seconds=10)
activity_inst = ActivityWaitCancelNotify()

async def wait_cancel_complete() -> None:
await asyncio.wait_for(
activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds()
)
activity_inst.wait_cancel_complete.clear()

async with new_worker(
client, CancelActivityWorkflow, activities=[activity_inst.wait_cancel]
client,
CancelActivityWorkflow,
activities=[activity_inst.wait_cancel],
max_heartbeat_throttle_interval=timedelta(milliseconds=300),
) as worker:
# Try cancel - confirm error and activity was sent the cancel
handle = await client.start_workflow(
Expand All @@ -1003,7 +1011,7 @@ async def activity_result() -> str:
await assert_eq_eventually(
"Error: CancelledError", activity_result, timeout=assert_timeout
)
await activity_inst.wait_cancel_complete.wait()
await wait_cancel_complete()
await handle.cancel()

# Wait cancel - confirm no error due to graceful cancel handling
Expand All @@ -1022,7 +1030,7 @@ async def activity_result() -> str:
activity_result,
timeout=assert_timeout,
)
await activity_inst.wait_cancel_complete.wait()
await wait_cancel_complete()
await handle.cancel()

# Abandon - confirm error and that activity stays running
Expand All @@ -1042,7 +1050,7 @@ async def activity_result() -> str:
await asyncio.sleep(0.5)
assert not activity_inst.wait_cancel_complete.is_set()
await handle.cancel()
await activity_inst.wait_cancel_complete.wait()
await wait_cancel_complete()


@workflow.defn
Expand Down
Loading