@@ -4,10 +4,13 @@ import (
44 "context"
55 "fmt"
66
7- "cloud.google.com/go/pubsub"
7+ "cloud.google.com/go/pubsub/v2"
8+ "cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
89 "github.com/gruntwork-io/terratest/modules/logger"
910 "github.com/gruntwork-io/terratest/modules/testing"
1011 "github.com/stretchr/testify/require"
12+ "google.golang.org/grpc/codes"
13+ "google.golang.org/grpc/status"
1114)
1215
1316// AssertTopicExists checks if the given Pub/Sub topic exists and fails the test if it does not.
@@ -51,13 +54,15 @@ func AssertTopicExistsContextE(t testing.TestingT, ctx context.Context, projectI
5154// (see pubsub_unit_test.go for the pattern).
5255// The ctx parameter supports cancellation and timeouts.
5356func AssertTopicExistsWithClient (ctx context.Context , client * pubsub.Client , topicName string ) error {
54- exists , err := client .Topic (topicName ).Exists (ctx )
57+ projectID := client .Project ()
58+ _ , err := client .TopicAdminClient .GetTopic (ctx , & pubsubpb.GetTopicRequest {
59+ Topic : topicResource (projectID , topicName ),
60+ })
5561 if err != nil {
56- return fmt .Errorf ("failed to check if Pub/Sub topic %s exists in project %s: %w" , topicName , client .Project (), err )
57- }
58-
59- if ! exists {
60- return fmt .Errorf ("Pub/Sub topic %s does not exist in project %s" , topicName , client .Project ())
62+ if status .Code (err ) == codes .NotFound {
63+ return fmt .Errorf ("Pub/Sub topic %s does not exist in project %s" , topicName , projectID )
64+ }
65+ return fmt .Errorf ("failed to check if Pub/Sub topic %s exists in project %s: %w" , topicName , projectID , err )
6166 }
6267
6368 return nil
@@ -104,13 +109,15 @@ func AssertSubscriptionExistsContextE(t testing.TestingT, ctx context.Context, p
104109// (see pubsub_unit_test.go for the pattern).
105110// The ctx parameter supports cancellation and timeouts.
106111func AssertSubscriptionExistsWithClient (ctx context.Context , client * pubsub.Client , subscriptionName string ) error {
107- exists , err := client .Subscription (subscriptionName ).Exists (ctx )
112+ projectID := client .Project ()
113+ _ , err := client .SubscriptionAdminClient .GetSubscription (ctx , & pubsubpb.GetSubscriptionRequest {
114+ Subscription : subscriptionResource (projectID , subscriptionName ),
115+ })
108116 if err != nil {
109- return fmt .Errorf ("failed to check if Pub/Sub subscription %s exists in project %s: %w" , subscriptionName , client .Project (), err )
110- }
111-
112- if ! exists {
113- return fmt .Errorf ("Pub/Sub subscription %s does not exist in project %s" , subscriptionName , client .Project ())
117+ if status .Code (err ) == codes .NotFound {
118+ return fmt .Errorf ("Pub/Sub subscription %s does not exist in project %s" , subscriptionName , projectID )
119+ }
120+ return fmt .Errorf ("failed to check if Pub/Sub subscription %s exists in project %s: %w" , subscriptionName , projectID , err )
114121 }
115122
116123 return nil
@@ -157,8 +164,11 @@ func CreateTopicContextE(t testing.TestingT, ctx context.Context, projectID stri
157164// (see pubsub_unit_test.go for the pattern).
158165// The ctx parameter supports cancellation and timeouts.
159166func CreateTopicWithClient (ctx context.Context , client * pubsub.Client , topicName string ) error {
160- if _ , err := client .CreateTopic (ctx , topicName ); err != nil {
161- return fmt .Errorf ("failed to create Pub/Sub topic %s in project %s: %w" , topicName , client .Project (), err )
167+ projectID := client .Project ()
168+ if _ , err := client .TopicAdminClient .CreateTopic (ctx , & pubsubpb.Topic {
169+ Name : topicResource (projectID , topicName ),
170+ }); err != nil {
171+ return fmt .Errorf ("failed to create Pub/Sub topic %s in project %s: %w" , topicName , projectID , err )
162172 }
163173
164174 return nil
@@ -205,8 +215,11 @@ func DeleteTopicContextE(t testing.TestingT, ctx context.Context, projectID stri
205215// (see pubsub_unit_test.go for the pattern).
206216// The ctx parameter supports cancellation and timeouts.
207217func DeleteTopicWithClient (ctx context.Context , client * pubsub.Client , topicName string ) error {
208- if err := client .Topic (topicName ).Delete (ctx ); err != nil {
209- return fmt .Errorf ("failed to delete Pub/Sub topic %s in project %s: %w" , topicName , client .Project (), err )
218+ projectID := client .Project ()
219+ if err := client .TopicAdminClient .DeleteTopic (ctx , & pubsubpb.DeleteTopicRequest {
220+ Topic : topicResource (projectID , topicName ),
221+ }); err != nil {
222+ return fmt .Errorf ("failed to delete Pub/Sub topic %s in project %s: %w" , topicName , projectID , err )
210223 }
211224
212225 return nil
@@ -253,10 +266,12 @@ func CreateSubscriptionContextE(t testing.TestingT, ctx context.Context, project
253266// (see pubsub_unit_test.go for the pattern).
254267// The ctx parameter supports cancellation and timeouts.
255268func CreateSubscriptionWithClient (ctx context.Context , client * pubsub.Client , subscriptionName string , topicName string ) error {
256- if _ , err := client .CreateSubscription (ctx , subscriptionName , pubsub.SubscriptionConfig {
257- Topic : client .Topic (topicName ),
269+ projectID := client .Project ()
270+ if _ , err := client .SubscriptionAdminClient .CreateSubscription (ctx , & pubsubpb.Subscription {
271+ Name : subscriptionResource (projectID , subscriptionName ),
272+ Topic : topicResource (projectID , topicName ),
258273 }); err != nil {
259- return fmt .Errorf ("failed to create Pub/Sub subscription %s on topic %s in project %s: %w" , subscriptionName , topicName , client . Project () , err )
274+ return fmt .Errorf ("failed to create Pub/Sub subscription %s on topic %s in project %s: %w" , subscriptionName , topicName , projectID , err )
260275 }
261276
262277 return nil
@@ -303,8 +318,11 @@ func DeleteSubscriptionContextE(t testing.TestingT, ctx context.Context, project
303318// (see pubsub_unit_test.go for the pattern).
304319// The ctx parameter supports cancellation and timeouts.
305320func DeleteSubscriptionWithClient (ctx context.Context , client * pubsub.Client , subscriptionName string ) error {
306- if err := client .Subscription (subscriptionName ).Delete (ctx ); err != nil {
307- return fmt .Errorf ("failed to delete Pub/Sub subscription %s in project %s: %w" , subscriptionName , client .Project (), err )
321+ projectID := client .Project ()
322+ if err := client .SubscriptionAdminClient .DeleteSubscription (ctx , & pubsubpb.DeleteSubscriptionRequest {
323+ Subscription : subscriptionResource (projectID , subscriptionName ),
324+ }); err != nil {
325+ return fmt .Errorf ("failed to delete Pub/Sub subscription %s in project %s: %w" , subscriptionName , projectID , err )
308326 }
309327
310328 return nil
@@ -319,3 +337,15 @@ func newPubSubClient(ctx context.Context, projectID string) (*pubsub.Client, err
319337
320338 return client , nil
321339}
340+
341+ // topicResource returns the fully-qualified Pub/Sub topic resource name ("projects/<p>/topics/<t>"),
342+ // required by the v2 admin client APIs.
343+ func topicResource (projectID , topicName string ) string {
344+ return fmt .Sprintf ("projects/%s/topics/%s" , projectID , topicName )
345+ }
346+
347+ // subscriptionResource returns the fully-qualified Pub/Sub subscription resource name
348+ // ("projects/<p>/subscriptions/<s>"), required by the v2 admin client APIs.
349+ func subscriptionResource (projectID , subscriptionName string ) string {
350+ return fmt .Sprintf ("projects/%s/subscriptions/%s" , projectID , subscriptionName )
351+ }
0 commit comments