mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Add a test to validate unpaused workflows does not impact Activity (#9573)
## What changed? WISOTT ## Why? Codifying the interaction between workflow unpause w/ paused activities. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [X] added new functional test(s) ## Potential risks Test only
This commit is contained in:
@@ -414,6 +414,109 @@ func (s *PauseWorkflowExecutionSuite) TestPauseWorkflowAndActivity() {
|
||||
}, 10*time.Second, 200*time.Millisecond)
|
||||
}
|
||||
|
||||
// TestUnpauseWorkflowKeepsActivityPaused tests that unpausing a workflow does not unpause its paused activities.
|
||||
// 1. Start a workflow with a failing activity
|
||||
// 2. Pause the activity
|
||||
// 3. Pause the workflow
|
||||
// 4. Unpause the workflow (while the activity is still paused)
|
||||
// 5. Verify the activity remains paused and the workflow is running
|
||||
func (s *PauseWorkflowExecutionSuite) TestUnpauseWorkflowKeepsActivityPaused() {
|
||||
ctx := testcore.NewContext()
|
||||
|
||||
s.activityShouldSucceed.Store(false)
|
||||
|
||||
activityID := "failing-activity"
|
||||
|
||||
workflowOptions := sdkclient.StartWorkflowOptions{
|
||||
ID: testcore.RandomizeStr("unpause-wf-keeps-activity-paused-" + s.T().Name()),
|
||||
TaskQueue: s.TaskQueue(),
|
||||
}
|
||||
|
||||
workflowRun, err := s.SdkClient().ExecuteWorkflow(ctx, workflowOptions, s.workflowWithFailingActivity)
|
||||
s.NoError(err)
|
||||
workflowID := workflowRun.GetID()
|
||||
runID := workflowRun.GetRunID()
|
||||
|
||||
// Wait for activity to fail at least once
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
desc, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowID, runID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, desc.PendingActivities, 1)
|
||||
require.NotNil(t, desc.PendingActivities[0].LastFailure)
|
||||
}, 5*time.Second, 200*time.Millisecond)
|
||||
|
||||
// Pause the activity
|
||||
_, err = s.FrontendClient().PauseActivity(ctx, &workflowservice.PauseActivityRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
|
||||
Activity: &workflowservice.PauseActivityRequest_Id{Id: activityID},
|
||||
Identity: s.pauseIdentity,
|
||||
Reason: "pausing activity for test",
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
desc, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowID, runID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, desc.PendingActivities, 1)
|
||||
require.True(t, desc.PendingActivities[0].Paused)
|
||||
}, 5*time.Second, 200*time.Millisecond)
|
||||
|
||||
// Pause the workflow
|
||||
_, err = s.FrontendClient().PauseWorkflowExecution(ctx, &workflowservice.PauseWorkflowExecutionRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
WorkflowId: workflowID,
|
||||
RunId: runID,
|
||||
Identity: s.pauseIdentity,
|
||||
Reason: s.pauseReason,
|
||||
RequestId: uuid.NewString(),
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
s.assertWorkflowIsPaused(ctx, t, workflowID, runID)
|
||||
}, 5*time.Second, 200*time.Millisecond)
|
||||
|
||||
// Unpause the workflow only
|
||||
_, err = s.FrontendClient().UnpauseWorkflowExecution(ctx, &workflowservice.UnpauseWorkflowExecutionRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
WorkflowId: workflowID,
|
||||
RunId: runID,
|
||||
Identity: s.pauseIdentity,
|
||||
Reason: s.pauseReason,
|
||||
RequestId: uuid.NewString(),
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
// Verify the workflow is running but the activity remains paused
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
desc, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowID, runID)
|
||||
require.NoError(t, err)
|
||||
info := desc.GetWorkflowExecutionInfo()
|
||||
require.NotNil(t, info)
|
||||
require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, info.GetStatus(),
|
||||
"workflow should be running after unpause, got: %s", info.GetStatus())
|
||||
require.Len(t, desc.PendingActivities, 1)
|
||||
require.True(t, desc.PendingActivities[0].Paused, "activity should still be paused after workflow unpause")
|
||||
}, 5*time.Second, 200*time.Millisecond)
|
||||
|
||||
// Cleanup: unblock and unpause the activity so the workflow can complete
|
||||
s.activityShouldSucceed.Store(true)
|
||||
_, err = s.FrontendClient().UnpauseActivity(ctx, &workflowservice.UnpauseActivityRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
|
||||
Activity: &workflowservice.UnpauseActivityRequest_Id{Id: activityID},
|
||||
Identity: s.pauseIdentity,
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
desc, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowID, runID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, desc.GetWorkflowExecutionInfo().GetStatus())
|
||||
}, 10*time.Second, 200*time.Millisecond)
|
||||
}
|
||||
|
||||
func (s *PauseWorkflowExecutionSuite) TestQueryWorkflowWhenPaused() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
Reference in New Issue
Block a user