mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
test: add SendToChannel helper and replace bare channel sends in functional tests (#9876)
## What changed?
Added a `SendToChannel(ctx, ch)` helper to `FunctionalTestBase`,
mirroring the existing `WaitForChannel`, and replaced all bare `ch <-
struct{}{}` sends in functional test bodies with calls to it.
## Why?
Bare channel sends in test bodies silently hang when the receiving
goroutine has stalled or died, causing tests to block until the overall
test timeout rather than failing fast with a useful message.
`WaitForChannel` already existed to solve this for receives,
`SendToChannel` closes the gap for sends.
## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)
## Potential risks
None, this is a pure test-helper addition.
This commit is contained in:
@@ -115,7 +115,7 @@ func (s *ActivityAPIPauseClientTestSuite) TestActivityPauseApi_WhileRunning() {
|
||||
}, 5*time.Second, 500*time.Millisecond)
|
||||
|
||||
// unblock the activity
|
||||
activityPausedCn <- struct{}{}
|
||||
env.SendToChannel(ctx, activityPausedCn)
|
||||
// make sure activity is paused on server and completed on the worker
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
description, err := env.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
|
||||
@@ -264,7 +264,7 @@ func (s *ActivityAPIPauseClientTestSuite) TestActivityPauseApi_IncreaseAttemptsO
|
||||
}, 5*time.Second, 500*time.Millisecond)
|
||||
|
||||
// End the activity
|
||||
activityPausedCn <- struct{}{}
|
||||
env.SendToChannel(ctx, activityPausedCn)
|
||||
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
description, err := env.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
|
||||
@@ -632,7 +632,7 @@ func (s *ActivityAPIPauseClientTestSuite) TestActivityPauseApi_WithReset() {
|
||||
}, 5*time.Second, 100*time.Millisecond)
|
||||
|
||||
// let activity finish
|
||||
activityCompleteCn <- struct{}{}
|
||||
env.SendToChannel(ctx, activityCompleteCn)
|
||||
|
||||
// wait for workflow to finish
|
||||
var out string
|
||||
|
||||
@@ -1765,7 +1765,7 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationCancelBeforeStarted_Cancelati
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
canStartCh <- struct{}{}
|
||||
env.SendToChannel(ctx, canStartCh)
|
||||
env.WaitForChannel(ctx, cancelSentCh)
|
||||
|
||||
// Terminate the workflow for good measure.
|
||||
|
||||
@@ -233,7 +233,7 @@ func (s *PauseWorkflowExecutionSuite) TestPauseUnpauseWorkflowExecution() {
|
||||
}, 5*time.Second, 200*time.Millisecond)
|
||||
|
||||
// Unblock the activity to complete the workflow.
|
||||
s.activityCompletedCh <- struct{}{}
|
||||
s.SendToChannel(ctx, s.activityCompletedCh)
|
||||
|
||||
// assert that the workflow completes now.
|
||||
s.EventuallyWithT(func(t *assert.CollectT) {
|
||||
@@ -591,7 +591,7 @@ func (s *PauseWorkflowExecutionSuite) TestQueryWorkflowWhenPaused() {
|
||||
s.NotNil(unpauseResp)
|
||||
|
||||
// Unblock the activity and send the signal to complete the workflow.
|
||||
s.activityCompletedCh <- struct{}{}
|
||||
s.SendToChannel(ctx, s.activityCompletedCh)
|
||||
err = s.SdkClient().SignalWorkflow(ctx, workflowID, runID, s.testEndSignal, "test end signal")
|
||||
s.NoError(err)
|
||||
|
||||
@@ -818,7 +818,7 @@ func (s *PauseWorkflowExecutionSuite) TestPauseWorkflowExecutionAlreadyPaused()
|
||||
}, 5*time.Second, 200*time.Millisecond)
|
||||
|
||||
// Unblock the activity and send the signal to complete the workflow.
|
||||
s.activityCompletedCh <- struct{}{}
|
||||
s.SendToChannel(ctx, s.activityCompletedCh)
|
||||
err = s.SdkClient().SignalWorkflow(ctx, workflowID, runID, s.testEndSignal, "test end signal")
|
||||
s.NoError(err)
|
||||
|
||||
|
||||
@@ -659,6 +659,15 @@ func (s *FunctionalTestBase) WaitForChannel(ctx context.Context, ch chan struct{
|
||||
}
|
||||
}
|
||||
|
||||
func (s *FunctionalTestBase) SendToChannel(ctx context.Context, ch chan struct{}) {
|
||||
s.T().Helper()
|
||||
select {
|
||||
case ch <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
s.FailNow("context timeout while sending to channel")
|
||||
}
|
||||
}
|
||||
|
||||
// TODO (alex): change to nsName namespace.Name
|
||||
func (s *FunctionalTestBase) SendSignal(nsName string, execution *commonpb.WorkflowExecution, signalName string,
|
||||
input *commonpb.Payloads, identity string) error {
|
||||
|
||||
@@ -2070,34 +2070,34 @@ func (s *VersioningIntegSuite) TestDispatchActivityUpgrade() {
|
||||
s.WaitForChannel(ctx, startedWf)
|
||||
rule2 := s.addRedirectRule(ctx, tq, v1, v11)
|
||||
s.waitForRedirectRulePropagation(ctx, tq, rule2)
|
||||
proceedWf <- struct{}{}
|
||||
s.SendToChannel(ctx, proceedWf)
|
||||
|
||||
s.WaitForChannel(ctx, started11)
|
||||
// wf assigned build ID should be updated by activity redirect
|
||||
s.validateWorkflowBuildIds(ctx, run.GetID(), run.GetRunID(), v11, true, v1, "", []string{v1})
|
||||
// let activity finish
|
||||
proceed11 <- struct{}{}
|
||||
s.SendToChannel(ctx, proceed11)
|
||||
|
||||
// wf replays on 1.1 so need to unblock it an extra time
|
||||
s.WaitForChannel(ctx, startedWf)
|
||||
proceedWf <- struct{}{}
|
||||
s.SendToChannel(ctx, proceedWf)
|
||||
|
||||
s.WaitForChannel(ctx, startedWf)
|
||||
rule2 = s.addRedirectRule(ctx, tq, v11, v12)
|
||||
s.waitForRedirectRulePropagation(ctx, tq, rule2)
|
||||
proceedWf <- struct{}{}
|
||||
s.SendToChannel(ctx, proceedWf)
|
||||
|
||||
s.WaitForChannel(ctx, started12)
|
||||
// wf assigned build ID should not be updated by independent activity redirect
|
||||
s.validateWorkflowBuildIds(ctx, run.GetID(), run.GetRunID(), v11, true, v11, "", []string{v1})
|
||||
// let activity finish
|
||||
proceed12 <- struct{}{}
|
||||
s.SendToChannel(ctx, proceed12)
|
||||
|
||||
// wf replays on 1.2 so need to unblock it two extra times
|
||||
s.WaitForChannel(ctx, startedWf)
|
||||
proceedWf <- struct{}{}
|
||||
s.SendToChannel(ctx, proceedWf)
|
||||
s.WaitForChannel(ctx, startedWf)
|
||||
proceedWf <- struct{}{}
|
||||
s.SendToChannel(ctx, proceedWf)
|
||||
|
||||
var out string
|
||||
s.NoError(run.Get(ctx, &out))
|
||||
|
||||
Reference in New Issue
Block a user