fix(tests): refactor gcp pubsub ctx propagation (#3822)

Signed-off-by: Mike Nguyen <[email protected]>
Co-authored-by: Yaron Schneider <[email protected]>
Co-authored-by: Nelson Parente <[email protected]>
Co-authored-by: Cassie Coyle <[email protected]>

Author: 4 people

tests/certification/pubsub/gcp/pubsub/gcppubsub_helper.go

@@ -39,19 +39,18 @@ type DataMessage struct {
 
 type MessageFunc func(*DataMessage) error
 
-func NewTopicManager(projectID string) (*topicManager, error) {
+func NewTopicManager(ctx context.Context, projectID string) (*topicManager, error) {
 	tpm := &topicManager{
 		projectID: projectID,
 	}
-	err := tpm.connect()
+	err := tpm.connect(ctx)
 	if err != nil {
 		return nil, err
 	}
 	return tpm, nil
 }
 
-func (tm *topicManager) connect() error {
-	ctx := t.Context()
+func (tm *topicManager) connect(ctx context.Context) error {
 	client, err := pubsub.NewClient(ctx, tm.projectID)
 	if err != nil {
 		return fmt.Errorf("GCP pubsub.NewClient failed to connect: %v", err)
@@ -65,9 +64,8 @@ func (tm *topicManager) disconnect() error {
 	return tm.gcpClient.Close()
 }
 
-func (tp *topicManager) GetMessages(topicID string, msgTimeout time.Duration, subID string, fn MessageFunc) (int, error) {
-	ctx := t.Context()
-
+func (tp *topicManager) GetMessages(ctx context.Context, topicID string, msgTimeout time.Duration, subID string,
+	fn MessageFunc) (int, error) {
 	topic := tp.gcpClient.Topic(topicID)
 	cfg := &pubsub.SubscriptionConfig{
 		Topic: topic,
@@ -133,24 +131,23 @@ func getOrCreateSub(ctx context.Context, client *pubsub.Client, subID string, cf
 	return sub, nil
 }
 
-func deleteSubscriptions(projectID string, subs []string) error {
+func deleteSubscriptions(ctx context.Context, projectID string, subs []string) error {
 	for _, s := range subs {
 		fmt.Printf("Deleting subscription: %s\n", s)
-		deleteSubscription(os.Stdout, projectID, s)
+		deleteSubscription(ctx, os.Stdout, projectID, s)
 	}
 	return nil
 }
 
-func deleteTopics(projectID string, topics []string) error {
+func deleteTopics(ctx context.Context, projectID string, topics []string) error {
 	for _, t := range topics {
 		fmt.Printf("Deleting topics: %s\n", t)
-		deleteTopic(os.Stdout, projectID, t)
+		deleteTopic(ctx, os.Stdout, projectID, t)
 	}
 	return nil
 }
 
-func deleteSubscription(w io.Writer, projectID, subID string) error {
-	ctx := t.Context()
+func deleteSubscription(ctx context.Context, w io.Writer, projectID, subID string) error {
 	client, err := pubsub.NewClient(ctx, projectID)
 	if err != nil {
 		return fmt.Errorf("pubsub.NewClient: %v", err)
@@ -165,8 +162,7 @@ func deleteSubscription(w io.Writer, projectID, subID string) error {
 	return nil
 }
 
-func deleteTopic(w io.Writer, projectID, topicID string) error {
-	ctx := t.Context()
+func deleteTopic(ctx context.Context, w io.Writer, projectID, topicID string) error {
 	client, err := pubsub.NewClient(ctx, projectID)
 	if err != nil {
 		return fmt.Errorf("pubsub.NewClient: %v", err)

tests/certification/pubsub/gcp/pubsub/gcppubsub_test.go

@@ -457,7 +457,7 @@ func GCPPubSubMessageDeadLetter(t *testing.T) {
 			t := time.NewTicker(500 * time.Millisecond)
 			defer t.Stop()
 			counter := 1
-			tm, err := NewTopicManager(projectID)
+			tm, err := NewTopicManager(ctx, projectID)
 			if err != nil {
 				return fmt.Errorf("deadLetterReceiverApplication - NewTopicManager: %v", err)
 			}
@@ -474,11 +474,12 @@ func GCPPubSubMessageDeadLetter(t *testing.T) {
 					ctx.Logf("deadLetterReceiverApplication - timeout waiting for messages from (%q)", deadLetterTopicName)
 					return fmt.Errorf("deadLetterReceiverApplication - timeout waiting for messages from (%q)", deadLetterTopicName)
 				case <-t.C:
-					numMsgs, err := tm.GetMessages(deadLetterTopicName, msgTimeout, deadLetterSubcription, func(m *DataMessage) error {
-						ctx.Logf("deadLetterReceiverApplication - received message counter(%d) (%v)\n", counter, m.Data)
-						messagesWatcher.Observe(m.Data)
-						return nil
-					})
+					numMsgs, err := tm.GetMessages(ctx, deadLetterTopicName, msgTimeout, deadLetterSubcription,
+						func(m *DataMessage) error {
+							ctx.Logf("deadLetterReceiverApplication - received message counter(%d) (%v)\n", counter, m.Data)
+							messagesWatcher.Observe(m.Data)
+							return nil
+						})
 					if err != nil {
 						ctx.Logf("deadLetterReceiverApplication - failed to get messages from (%q) counter(%d) %v  - trying again\n", deadLetterTopicName, counter, err)
 						continue
@@ -797,11 +798,11 @@ func teardown(t *testing.T) {
 	t.Logf("GCP PubSub CertificationTests teardown...")
 	//Dapr runtime automatically creates the following subscriptions, topics
 	//so here they get deleted.
-	if err := deleteSubscriptions(projectID, subscriptions); err != nil {
+	if err := deleteSubscriptions(t.Context(), projectID, subscriptions); err != nil {
 		t.Log(err)
 	}
 
-	if err := deleteTopics(projectID, topics); err != nil {
+	if err := deleteTopics(t.Context(), projectID, topics); err != nil {
 		t.Log(err)
 	}