| // Copyright 2014 Google LLC |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| package pubsub |
| |
| import ( |
| "context" |
| "fmt" |
| "testing" |
| "time" |
| |
| "cloud.google.com/go/iam" |
| "cloud.google.com/go/internal" |
| "cloud.google.com/go/internal/testutil" |
| "cloud.google.com/go/internal/uid" |
| gax "github.com/googleapis/gax-go" |
| "google.golang.org/api/iterator" |
| "google.golang.org/api/option" |
| "google.golang.org/grpc" |
| "google.golang.org/grpc/codes" |
| "google.golang.org/grpc/status" |
| ) |
| |
| var ( |
| topicIDs = uid.NewSpace("topic", nil) |
| subIDs = uid.NewSpace("sub", nil) |
| ) |
| |
| // messageData is used to hold the contents of a message so that it can be compared against the contents |
| // of another message without regard to irrelevant fields. |
| type messageData struct { |
| ID string |
| Data []byte |
| Attributes map[string]string |
| } |
| |
| func extractMessageData(m *Message) *messageData { |
| return &messageData{ |
| ID: m.ID, |
| Data: m.Data, |
| Attributes: m.Attributes, |
| } |
| } |
| |
| func integrationTestClient(ctx context.Context, t *testing.T) *Client { |
| if testing.Short() { |
| t.Skip("Integration tests skipped in short mode") |
| } |
| projID := testutil.ProjID() |
| if projID == "" { |
| t.Skip("Integration tests skipped. See CONTRIBUTING.md for details") |
| } |
| ts := testutil.TokenSource(ctx, ScopePubSub, ScopeCloudPlatform) |
| if ts == nil { |
| t.Skip("Integration tests skipped. See CONTRIBUTING.md for details") |
| } |
| client, err := NewClient(ctx, projID, option.WithTokenSource(ts)) |
| if err != nil { |
| t.Fatalf("Creating client error: %v", err) |
| } |
| return client |
| } |
| |
| func TestIntegration_All(t *testing.T) { |
| t.Parallel() |
| ctx := context.Background() |
| client := integrationTestClient(ctx, t) |
| defer client.Close() |
| |
| topic, err := client.CreateTopic(ctx, topicIDs.New()) |
| if err != nil { |
| t.Errorf("CreateTopic error: %v", err) |
| } |
| defer topic.Stop() |
| exists, err := topic.Exists(ctx) |
| if err != nil { |
| t.Fatalf("TopicExists error: %v", err) |
| } |
| if !exists { |
| t.Errorf("topic %v should exist, but it doesn't", topic) |
| } |
| |
| var sub *Subscription |
| if sub, err = client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{Topic: topic}); err != nil { |
| t.Errorf("CreateSub error: %v", err) |
| } |
| exists, err = sub.Exists(ctx) |
| if err != nil { |
| t.Fatalf("SubExists error: %v", err) |
| } |
| if !exists { |
| t.Errorf("subscription %s should exist, but it doesn't", sub.ID()) |
| } |
| |
| for _, sync := range []bool{false, true} { |
| for _, maxMsgs := range []int{0, 3, -1} { // MaxOutstandingMessages = default, 3, unlimited |
| testPublishAndReceive(t, topic, sub, maxMsgs, sync) |
| } |
| } |
| if msg, ok := testIAM(ctx, topic.IAM(), "pubsub.topics.get"); !ok { |
| t.Errorf("topic IAM: %s", msg) |
| } |
| if msg, ok := testIAM(ctx, sub.IAM(), "pubsub.subscriptions.get"); !ok { |
| t.Errorf("sub IAM: %s", msg) |
| } |
| |
| snap, err := sub.CreateSnapshot(ctx, "") |
| if err != nil { |
| t.Fatalf("CreateSnapshot error: %v", err) |
| } |
| |
| timeoutCtx, cancel := context.WithTimeout(ctx, time.Minute) |
| defer cancel() |
| err = internal.Retry(timeoutCtx, gax.Backoff{}, func() (bool, error) { |
| snapIt := client.Snapshots(timeoutCtx) |
| for { |
| s, err := snapIt.Next() |
| if err == nil && s.name == snap.name { |
| return true, nil |
| } |
| if err == iterator.Done { |
| return false, fmt.Errorf("cannot find snapshot: %q", snap.name) |
| } |
| if err != nil { |
| return false, err |
| } |
| } |
| }) |
| if err != nil { |
| t.Error(err) |
| } |
| |
| err = internal.Retry(timeoutCtx, gax.Backoff{}, func() (bool, error) { |
| err := sub.SeekToSnapshot(timeoutCtx, snap.Snapshot) |
| return err == nil, err |
| }) |
| if err != nil { |
| t.Error(err) |
| } |
| |
| err = internal.Retry(timeoutCtx, gax.Backoff{}, func() (bool, error) { |
| err := sub.SeekToTime(timeoutCtx, time.Now()) |
| return err == nil, err |
| }) |
| if err != nil { |
| t.Error(err) |
| } |
| |
| err = internal.Retry(timeoutCtx, gax.Backoff{}, func() (bool, error) { |
| snapHandle := client.Snapshot(snap.ID()) |
| err := snapHandle.Delete(timeoutCtx) |
| return err == nil, err |
| }) |
| if err != nil { |
| t.Error(err) |
| } |
| |
| if err := sub.Delete(ctx); err != nil { |
| t.Errorf("DeleteSub error: %v", err) |
| } |
| |
| if err := topic.Delete(ctx); err != nil { |
| t.Errorf("DeleteTopic error: %v", err) |
| } |
| } |
| |
| func testPublishAndReceive(t *testing.T, topic *Topic, sub *Subscription, maxMsgs int, synchronous bool) { |
| ctx := context.Background() |
| var msgs []*Message |
| for i := 0; i < 10; i++ { |
| text := fmt.Sprintf("a message with an index %d", i) |
| attrs := make(map[string]string) |
| attrs["foo"] = "bar" |
| msgs = append(msgs, &Message{ |
| Data: []byte(text), |
| Attributes: attrs, |
| }) |
| } |
| |
| // Publish some messages. |
| type pubResult struct { |
| m *Message |
| r *PublishResult |
| } |
| var rs []pubResult |
| for _, m := range msgs { |
| r := topic.Publish(ctx, m) |
| rs = append(rs, pubResult{m, r}) |
| } |
| want := make(map[string]*messageData) |
| for _, res := range rs { |
| id, err := res.r.Get(ctx) |
| if err != nil { |
| t.Fatal(err) |
| } |
| md := extractMessageData(res.m) |
| md.ID = id |
| want[md.ID] = md |
| } |
| |
| sub.ReceiveSettings.MaxOutstandingMessages = maxMsgs |
| sub.ReceiveSettings.Synchronous = synchronous |
| |
| // Use a timeout to ensure that Pull does not block indefinitely if there are |
| // unexpectedly few messages available. |
| now := time.Now() |
| timeoutCtx, _ := context.WithTimeout(ctx, time.Minute) |
| gotMsgs, err := pullN(timeoutCtx, sub, len(want), func(ctx context.Context, m *Message) { |
| m.Ack() |
| }) |
| if err != nil { |
| if c := status.Convert(err); c.Code() == codes.Canceled { |
| if time.Now().Sub(now) >= time.Minute { |
| t.Fatal("pullN took too long") |
| } |
| } else { |
| t.Fatalf("Pull: %v", err) |
| } |
| } |
| got := make(map[string]*messageData) |
| for _, m := range gotMsgs { |
| md := extractMessageData(m) |
| got[md.ID] = md |
| } |
| if !testutil.Equal(got, want) { |
| t.Fatalf("MaxOutstandingMessages=%d, Synchronous=%t, messages got: %v, messages want: %v", |
| maxMsgs, synchronous, got, want) |
| } |
| } |
| |
| // IAM tests. |
| // NOTE: for these to succeed, the test runner identity must have the Pub/Sub Admin or Owner roles. |
| // To set, visit https://console.developers.google.com, select "IAM & Admin" from the top-left |
| // menu, choose the account, click the Roles dropdown, and select "Pub/Sub > Pub/Sub Admin". |
| // TODO(jba): move this to a testing package within cloud.google.com/iam, so we can re-use it. |
| func testIAM(ctx context.Context, h *iam.Handle, permission string) (msg string, ok bool) { |
| // Attempting to add an non-existent identity (e.g. "alice@example.com") causes the service |
| // to return an internal error, so use a real identity. |
| const member = "domain:google.com" |
| |
| var policy *iam.Policy |
| var err error |
| |
| if policy, err = h.Policy(ctx); err != nil { |
| return fmt.Sprintf("Policy: %v", err), false |
| } |
| // The resource is new, so the policy should be empty. |
| if got := policy.Roles(); len(got) > 0 { |
| return fmt.Sprintf("initially: got roles %v, want none", got), false |
| } |
| // Add a member, set the policy, then check that the member is present. |
| policy.Add(member, iam.Viewer) |
| if err := h.SetPolicy(ctx, policy); err != nil { |
| return fmt.Sprintf("SetPolicy: %v", err), false |
| } |
| if policy, err = h.Policy(ctx); err != nil { |
| return fmt.Sprintf("Policy: %v", err), false |
| } |
| if got, want := policy.Members(iam.Viewer), []string{member}; !testutil.Equal(got, want) { |
| return fmt.Sprintf("after Add: got %v, want %v", got, want), false |
| } |
| // Now remove that member, set the policy, and check that it's empty again. |
| policy.Remove(member, iam.Viewer) |
| if err := h.SetPolicy(ctx, policy); err != nil { |
| return fmt.Sprintf("SetPolicy: %v", err), false |
| } |
| if policy, err = h.Policy(ctx); err != nil { |
| return fmt.Sprintf("Policy: %v", err), false |
| } |
| if got := policy.Roles(); len(got) > 0 { |
| return fmt.Sprintf("after Remove: got roles %v, want none", got), false |
| } |
| // Call TestPermissions. |
| // Because this user is an admin, it has all the permissions on the |
| // resource type. Note: the service fails if we ask for inapplicable |
| // permissions (e.g. a subscription permission on a topic, or a topic |
| // create permission on a topic rather than its parent). |
| wantPerms := []string{permission} |
| gotPerms, err := h.TestPermissions(ctx, wantPerms) |
| if err != nil { |
| return fmt.Sprintf("TestPermissions: %v", err), false |
| } |
| if !testutil.Equal(gotPerms, wantPerms) { |
| return fmt.Sprintf("TestPermissions: got %v, want %v", gotPerms, wantPerms), false |
| } |
| return "", true |
| } |
| |
| func TestIntegration_CancelReceive(t *testing.T) { |
| t.Parallel() |
| ctx, cancel := context.WithCancel(context.Background()) |
| client := integrationTestClient(ctx, t) |
| defer client.Close() |
| |
| topic, err := client.CreateTopic(ctx, topicIDs.New()) |
| if err != nil { |
| t.Fatal(err) |
| } |
| defer topic.Delete(ctx) |
| defer topic.Stop() |
| |
| var sub *Subscription |
| if sub, err = client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{Topic: topic}); err != nil { |
| t.Fatal(err) |
| } |
| defer sub.Delete(ctx) |
| |
| sub.ReceiveSettings.MaxOutstandingMessages = -1 |
| sub.ReceiveSettings.MaxOutstandingBytes = -1 |
| sub.ReceiveSettings.NumGoroutines = 1 |
| |
| doneReceiving := make(chan struct{}) |
| |
| // Publish the messages. |
| go func() { |
| for { |
| select { |
| case <-doneReceiving: |
| return |
| default: |
| topic.Publish(ctx, &Message{Data: []byte("some msg")}) |
| time.Sleep(time.Second) |
| } |
| } |
| }() |
| |
| go func() { |
| defer close(doneReceiving) |
| err = sub.Receive(ctx, func(_ context.Context, msg *Message) { |
| cancel() |
| time.AfterFunc(5*time.Second, msg.Ack) |
| }) |
| if err != nil { |
| t.Error(err) |
| } |
| }() |
| |
| select { |
| case <-time.After(60 * time.Second): |
| t.Fatalf("Waited 60 seconds for Receive to finish, should have finished sooner") |
| case <-doneReceiving: |
| } |
| } |
| |
| func TestIntegration_UpdateSubscription(t *testing.T) { |
| t.Parallel() |
| ctx := context.Background() |
| client := integrationTestClient(ctx, t) |
| defer client.Close() |
| |
| topic, err := client.CreateTopic(ctx, topicIDs.New()) |
| if err != nil { |
| t.Fatalf("CreateTopic error: %v", err) |
| } |
| defer topic.Stop() |
| defer topic.Delete(ctx) |
| |
| var sub *Subscription |
| if sub, err = client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{Topic: topic}); err != nil { |
| t.Fatalf("CreateSub error: %v", err) |
| } |
| defer sub.Delete(ctx) |
| |
| got, err := sub.Config(ctx) |
| if err != nil { |
| t.Fatal(err) |
| } |
| want := SubscriptionConfig{ |
| Topic: topic, |
| AckDeadline: 10 * time.Second, |
| RetainAckedMessages: false, |
| RetentionDuration: defaultRetentionDuration, |
| } |
| if !testutil.Equal(got, want) { |
| t.Fatalf("\ngot %+v\nwant %+v", got, want) |
| } |
| // Add a PushConfig and change other fields. |
| projID := testutil.ProjID() |
| pc := PushConfig{ |
| Endpoint: "https://" + projID + ".appspot.com/_ah/push-handlers/push", |
| Attributes: map[string]string{"x-goog-version": "v1"}, |
| } |
| got, err = sub.Update(ctx, SubscriptionConfigToUpdate{ |
| PushConfig: &pc, |
| AckDeadline: 2 * time.Minute, |
| RetainAckedMessages: true, |
| RetentionDuration: 2 * time.Hour, |
| Labels: map[string]string{"label": "value"}, |
| }) |
| if err != nil { |
| t.Fatal(err) |
| } |
| want = SubscriptionConfig{ |
| Topic: topic, |
| PushConfig: pc, |
| AckDeadline: 2 * time.Minute, |
| RetainAckedMessages: true, |
| RetentionDuration: 2 * time.Hour, |
| Labels: map[string]string{"label": "value"}, |
| } |
| |
| if !testutil.Equal(got, want) { |
| t.Fatalf("\ngot %+v\nwant %+v", got, want) |
| } |
| |
| // Remove the PushConfig, turning the subscription back into pull mode. |
| // Change AckDeadline, remove labels. |
| pc = PushConfig{} |
| got, err = sub.Update(ctx, SubscriptionConfigToUpdate{ |
| PushConfig: &pc, |
| AckDeadline: 30 * time.Second, |
| Labels: map[string]string{}, |
| }) |
| if err != nil { |
| t.Fatal(err) |
| } |
| want.PushConfig = pc |
| want.AckDeadline = 30 * time.Second |
| want.Labels = nil |
| // service issue: PushConfig attributes are not removed. |
| // TODO(jba): remove when issue resolved. |
| want.PushConfig.Attributes = map[string]string{"x-goog-version": "v1"} |
| if !testutil.Equal(got, want) { |
| t.Fatalf("\ngot %+v\nwant %+v", got, want) |
| } |
| // If nothing changes, our client returns an error. |
| _, err = sub.Update(ctx, SubscriptionConfigToUpdate{}) |
| if err == nil { |
| t.Fatal("got nil, wanted error") |
| } |
| } |
| |
| // NOTE: This test should be skipped by open source contributors. It requires |
| // whitelisting, a (gsuite) organization project, and specific permissions. |
| func TestIntegration_UpdateTopic(t *testing.T) { |
| t.Parallel() |
| ctx := context.Background() |
| client := integrationTestClient(ctx, t) |
| defer client.Close() |
| |
| compareConfig := func(got TopicConfig, wantLabels map[string]string) bool { |
| if !testutil.Equal(got.Labels, wantLabels) { |
| return false |
| } |
| // For MessageStoragePolicy, we don't want to check for an exact set of regions. |
| // That set may change at any time. Instead, just make sure that the set isn't empty. |
| if len(got.MessageStoragePolicy.AllowedPersistenceRegions) == 0 { |
| return false |
| } |
| return true |
| } |
| |
| topic, err := client.CreateTopic(ctx, topicIDs.New()) |
| if err != nil { |
| t.Fatalf("CreateTopic error: %v", err) |
| } |
| defer topic.Stop() |
| defer topic.Delete(ctx) |
| |
| got, err := topic.Config(ctx) |
| if err != nil { |
| t.Fatal(err) |
| } |
| if !compareConfig(got, nil) { |
| t.Fatalf("\ngot %+v\nwant no labels", got) |
| } |
| |
| labels := map[string]string{"label": "value"} |
| got, err = topic.Update(ctx, TopicConfigToUpdate{Labels: labels}) |
| if err != nil { |
| t.Fatal(err) |
| } |
| if !compareConfig(got, labels) { |
| t.Fatalf("\ngot %+v\nwant labels %+v", got, labels) |
| } |
| // Remove all labels. |
| got, err = topic.Update(ctx, TopicConfigToUpdate{Labels: map[string]string{}}) |
| if err != nil { |
| t.Fatal(err) |
| } |
| if !compareConfig(got, nil) { |
| t.Fatalf("\ngot %+v\nwant no labels", got) |
| } |
| } |
| |
| func TestIntegration_PublicTopic(t *testing.T) { |
| t.Parallel() |
| ctx := context.Background() |
| client := integrationTestClient(ctx, t) |
| defer client.Close() |
| |
| sub, err := client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{ |
| Topic: client.TopicInProject("taxirides-realtime", "pubsub-public-data"), |
| }) |
| if err != nil { |
| t.Fatal(err) |
| } |
| defer sub.Delete(ctx) |
| // Confirm that Receive works. It doesn't matter if we actually get any |
| // messages. |
| ctxt, cancel := context.WithTimeout(ctx, 5*time.Second) |
| err = sub.Receive(ctxt, func(_ context.Context, msg *Message) { |
| msg.Ack() |
| cancel() |
| }) |
| if err != nil { |
| t.Fatal(err) |
| } |
| } |
| |
| func TestIntegration_Errors(t *testing.T) { |
| // Test various edge conditions. |
| t.Parallel() |
| ctx := context.Background() |
| client := integrationTestClient(ctx, t) |
| defer client.Close() |
| |
| topic, err := client.CreateTopic(ctx, topicIDs.New()) |
| if err != nil { |
| t.Fatalf("CreateTopic error: %v", err) |
| } |
| defer topic.Stop() |
| defer topic.Delete(ctx) |
| |
| // Out-of-range retention duration. |
| sub, err := client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{ |
| Topic: topic, |
| RetentionDuration: 1 * time.Second, |
| }) |
| if want := codes.InvalidArgument; grpc.Code(err) != want { |
| t.Errorf("got <%v>, want %s", err, want) |
| } |
| if err == nil { |
| sub.Delete(ctx) |
| } |
| |
| // Ack deadline less than minimum. |
| sub, err = client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{ |
| Topic: topic, |
| AckDeadline: 5 * time.Second, |
| }) |
| if want := codes.Unknown; grpc.Code(err) != want { |
| t.Errorf("got <%v>, want %s", err, want) |
| } |
| if err == nil { |
| sub.Delete(ctx) |
| } |
| |
| // Updating a non-existent subscription. |
| sub = client.Subscription(subIDs.New()) |
| _, err = sub.Update(ctx, SubscriptionConfigToUpdate{AckDeadline: 20 * time.Second}) |
| if want := codes.NotFound; grpc.Code(err) != want { |
| t.Errorf("got <%v>, want %s", err, want) |
| } |
| // Deleting a non-existent subscription. |
| err = sub.Delete(ctx) |
| if want := codes.NotFound; grpc.Code(err) != want { |
| t.Errorf("got <%v>, want %s", err, want) |
| } |
| |
| // Updating out-of-range retention duration. |
| sub, err = client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{Topic: topic}) |
| if err != nil { |
| t.Fatal(err) |
| } |
| defer sub.Delete(ctx) |
| _, err = sub.Update(ctx, SubscriptionConfigToUpdate{RetentionDuration: 1000 * time.Hour}) |
| if want := codes.InvalidArgument; grpc.Code(err) != want { |
| t.Errorf("got <%v>, want %s", err, want) |
| } |
| } |
| |
| // NOTE: This test should be skipped by open source contributors. It requires |
| // whitelisting, a (gsuite) organization project, and specific permissions. |
| // |
| // Googlers, see internal bug 77920644. Furthermore, be sure to add your |
| // service account as an owner of ps-geofencing-test. |
| func TestIntegration_MessageStoragePolicy(t *testing.T) { |
| // Verify that the message storage policy is populated. |
| if testing.Short() { |
| t.Skip("Integration tests skipped in short mode") |
| } |
| ctx := context.Background() |
| // The message storage policy depends on the Resource Location Restriction org policy. |
| // The usual testing project is in the google.com org, which has no resource location restrictions, |
| // so we will always see an empty MessageStoragePolicy. Use a project in another org that does |
| // have a restriction set ("us-east1"). |
| projID := "ps-geofencing-test" |
| // We can use the same creds as always because the service account of the default testing project |
| // has permission to use the above project. This test will fail if a different service account |
| // is used for testing. |
| ts := testutil.TokenSource(ctx, ScopePubSub, ScopeCloudPlatform) |
| if ts == nil { |
| t.Skip("Integration tests skipped. See CONTRIBUTING.md for details") |
| } |
| client, err := NewClient(ctx, projID, option.WithTokenSource(ts)) |
| if err != nil { |
| t.Fatalf("Creating client error: %v", err) |
| } |
| topic, err := client.CreateTopic(ctx, topicIDs.New()) |
| if err != nil { |
| t.Fatalf("CreateTopic error: %v", err) |
| } |
| defer topic.Stop() |
| defer topic.Delete(ctx) |
| |
| config, err := topic.Config(ctx) |
| if err != nil { |
| t.Fatal(err) |
| } |
| got := config.MessageStoragePolicy.AllowedPersistenceRegions |
| want := []string{"us-east1"} |
| if !testutil.Equal(got, want) { |
| t.Errorf("got %v, want %v", got, want) |
| } |
| } |