go.temporal.io/server/tests/matching_utils.go
77 LOC · 0 covered · 77 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"context"
5
"testing"
6
"time"
7
8
"github.com/stretchr/testify/require"
9
enumspb "go.temporal.io/api/enums/v1"
10
deploymentspb "go.temporal.io/server/api/deployment/v1"
11
"go.temporal.io/server/api/matchingservice/v1"
12
"go.temporal.io/server/common/namespace"
13
"go.temporal.io/server/common/primitives/timestamp"
14
"go.temporal.io/server/common/testing/testvars"
15
"go.temporal.io/server/tests/testcore"
16
)
17
18
// runWithMatchingBehaviors runs a test with all combinations of matching behaviors.
19
func runWithMatchingBehaviors(
20
t *testing.T,
21
baseOpts []testcore.TestOption,
22
subtest func(s *testcore.TestEnv, behavior testcore.MatchingBehavior),
23
) {
24
for _, behavior := range testcore.AllMatchingBehaviors() {
25
t.Run(behavior.Name(), func(t *testing.T) {
26
opts := append([]testcore.TestOption{}, baseOpts...)
27
opts = append(opts, behavior.Options()...)
28
29
env := testcore.NewEnv(t, opts...)
30
behavior.InjectHooks(env)
31
32
subtest(env, behavior)
33
})
34
}
35
}
36
37
// syncDeploymentVersionToTaskQueues sends a SyncDeploymentUserData request to the matching service
38
// to register a deployment version as current for the specified task queue types, then waits for
39
// the data to propagate to all partitions using CheckTaskQueueUserDataPropagation.
40
func syncDeploymentVersionToTaskQueues(
41
t testing.TB,
42
matchingClient matchingservice.MatchingServiceClient,
43
namespaceID namespace.ID,
44
tv *testvars.TestVars,
45
tqTypes ...enumspb.TaskQueueType,
46
) {
47
t.Helper()
48
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
49
defer cancel()
50
51
now := timestamp.TimePtr(time.Now())
52
resp, err := matchingClient.SyncDeploymentUserData(
53
ctx, &matchingservice.SyncDeploymentUserDataRequest{
54
NamespaceId: namespaceID.String(),
55
TaskQueue: tv.TaskQueue().GetName(),
56
TaskQueueTypes: tqTypes,
57
Operation: &matchingservice.SyncDeploymentUserDataRequest_UpdateVersionData{
58
UpdateVersionData: &deploymentspb.DeploymentVersionData{
59
Version: tv.DeploymentVersion(),
60
RoutingUpdateTime: now,
61
CurrentSinceTime: now,
62
},
63
},
64
},
65
)
66
require.NoError(t, err)
67
68
// Wait for the data to propagate to all partitions.
69
_, err = matchingClient.CheckTaskQueueUserDataPropagation(
70
ctx, &matchingservice.CheckTaskQueueUserDataPropagationRequest{
71
NamespaceId: namespaceID.String(),
72
TaskQueue: tv.TaskQueue().GetName(),
73
Version: resp.GetVersion(),
74
},
75
)
76
require.NoError(t, err)
77
}