ID string

Data string

Attributes map[string]string

return messageData{

ID: m.ID,

Data: string(m.Data),

Attributes: m.Attributes,

}

grpcHeadersEnforcer := &testutil.HeadersEnforcer{

OnFailure: t.Errorf,

Checkers: []*testutil.HeaderChecker{

testutil.XGoogClientHeaderChecker,

},

}

return append(grpcHeadersEnforcer.CallOptions(), opts...)

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")

}

opts = append(withGRPCHeadersAssertion(t, option.WithTokenSource(ts)), opts...)

client, err := NewClient(ctx, projID, opts...)

if err != nil {

t.Fatalf("Creating client error: %v", err)

}

return 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")

}

opts = append(withGRPCHeadersAssertion(t, option.WithTokenSource(ts)), opts...)

sc, err := NewSchemaClient(ctx, projID, opts...)

if err != nil {

t.Fatalf("Creating client error: %v", err)

}

return sc

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())

}

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)

}

labels := map[string]string{"foo": "bar"}

sc, err := snap.SetLabels(ctx, labels)

if err != nil {

t.Fatalf("Snapshot.SetLabels error: %v", err)

}

if diff := testutil.Diff(sc.Labels, labels); diff != "" {

t.Fatalf("\ngot: - want: +\n%s", diff)

}

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)

}

ctx := context.Background()

client := integrationTestClient(ctx, t)

for _, sync := range []bool{false, true} {

for _, maxMsgs := range []int{0, 3, -1} { // MaxOutstandingMessages = default, 3, unlimited

testPublishAndReceive(t, client, maxMsgs, sync, false, 10, 0)

}

// Tests for large messages (larger than the 4MB gRPC limit).

testPublishAndReceive(t, client, 0, sync, false, 1, 5*1024*1024)

}

ctxMD, _ := metadata.FromOutgoingContext(ctx)

kv := []string{

"gl-go",

version.Go(),

"gax",

gax.Version,

"grpc",

grpc.Version,

}

allMDs := append([]metadata.MD{ctxMD}, metadata.Pairs("x-goog-api-client", gax.XGoogHeader(kv...)))

return metadata.NewOutgoingContext(ctx, metadata.Join(allMDs...))

t.Run(fmt.Sprintf("maxMsgs:%d,synchronous:%t,exactlyOnceDelivery:%t,numMsgs:%d", maxMsgs, synchronous, exactlyOnceDelivery, numMsgs), func(t *testing.T) {

t.Parallel()

ctx := context.Background()

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)

}

sub, err := client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{

Topic: topic,

EnableExactlyOnceDelivery: exactlyOnceDelivery,

})

if 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())

}

var msgs []*Message

for i := 0; i < numMsgs; i++ {

text := fmt.Sprintf("a message with an index %d - %s", i, strings.Repeat(".", extraBytes))

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()

timeout := 3 * time.Minute

timeoutCtx, cancel := context.WithTimeout(ctx, timeout)

defer cancel()

gotMsgs, err := pullN(timeoutCtx, sub, len(want), 2*time.Second, func(ctx context.Context, m *Message) {

if exactlyOnceDelivery {

if _, err := m.AckWithResult().Get(ctx); err != nil {

t.Fatalf("failed to ack message with exactly once delivery: %v", err)

}

return

}

m.Ack()

})

if err != nil {

if c := status.Convert(err); c.Code() == codes.Canceled {

if time.Since(now) >= timeout {

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)

}

})

// Manually adding withGoogleClientInfo here because this code only takes

// a handle with a grpc.ClientConn that has the "x-goog-api-client" header enforcer,

// but unfortunately not the underlying infrastructure that takes pre-set headers.

ctx = withGoogleClientInfo(ctx)

// 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

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.Delete(ctx)

defer topic.Stop()

// Calculate the largest possible message length that is still valid.

// First, calculate the max length of the encoded message accounting for the topic name.

length := MaxPublishRequestBytes - calcFieldSizeString(topic.String())

// Next, account for the overhead from encoding an individual PubsubMessage,

// and the inner PubsubMessage.Data field.

pbMsgOverhead := 1 + protowire.SizeVarint(uint64(length))

dataOverhead := 1 + protowire.SizeVarint(uint64(length-pbMsgOverhead))

maxLengthSingleMessage := length - pbMsgOverhead - dataOverhead

publishReq := &pb.PublishRequest{

Topic: topic.String(),

Messages: []*pb.PubsubMessage{

{

Data: bytes.Repeat([]byte{'A'}, maxLengthSingleMessage),

},

},

}

if got := proto.Size(publishReq); got != MaxPublishRequestBytes {

t.Fatalf("Created request size of %d bytes,\nwant %f bytes", got, MaxPublishRequestBytes)

}

// Publishing the max length message by itself should succeed.

msg := &Message{

Data: bytes.Repeat([]byte{'A'}, maxLengthSingleMessage),

}

topic.PublishSettings.FlowControlSettings.LimitExceededBehavior = FlowControlSignalError

r := topic.Publish(ctx, msg)

if _, err := r.Get(ctx); err != nil {

t.Fatalf("Failed to publish max length message: %v", err)

}

// Publish a small message first and make sure the max length message

// is added to its own bundle.

smallMsg := &Message{

Data: []byte{'A'},

}

topic.Publish(ctx, smallMsg)

r = topic.Publish(ctx, msg)

if _, err := r.Get(ctx); err != nil {

t.Fatalf("Failed to publish max length message after a small message: %v", err)

}

// Increase the data byte string by 1 byte, which should cause the request to fail,

// specifically due to exceeding the bundle byte limit.

msg.Data = append(msg.Data, 'A')

r = topic.Publish(ctx, msg)

if _, err := r.Get(ctx); err != ErrOversizedMessage {

t.Fatalf("Should throw item size too large error, got %v", err)

}

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:

}

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.Delete(ctx)

defer topic.Stop()

cfg := SubscriptionConfig{

Topic: topic,

ExpirationPolicy: time.Duration(0),

}

var sub *Subscription

if sub, err = client.CreateSubscription(ctx, subIDs.New(), cfg); err != nil {

t.Fatalf("CreateSub error: %v", err)

}

defer sub.Delete(ctx)

got, err := sub.Config(ctx)

if err != nil {

t.Fatal(err)

}

want := time.Duration(0)

if got.ExpirationPolicy != want {

t.Fatalf("config.ExpirationPolicy mismatch, got: %v, want: %v\n", got.ExpirationPolicy, want)

}

jwtConf, err := testutil.JWTConfig()

if err == nil && jwtConf != nil {

return jwtConf.Email

}

creds := testutil.Credentials(ctx, ScopePubSub, ScopeCloudPlatform)

if creds == nil {

t.Fatal("Failed to retrieve credentials")

}

if len(creds.JSON) == 0 {

t.Skip("No JWTConfig JSON was present so can't get serviceAccountEmail")

}

jwtConf, err = google.JWTConfigFromJSON(creds.JSON)

if err != nil {

t.Fatalf("Failed to parse Google JWTConfig from JSON: %v", err)

}

return jwtConf.Email

t.Parallel()

ctx := context.Background()

client := integrationTestClient(ctx, t)

defer client.Close()

serviceAccountEmail := findServiceAccountEmail(ctx, t)

topic, err := client.CreateTopic(ctx, topicIDs.New())

if err != nil {

t.Fatalf("CreateTopic error: %v", err)

}

defer topic.Delete(ctx)

defer topic.Stop()

var sub *Subscription

projID := testutil.ProjID()

sCfg := SubscriptionConfig{

Topic: topic,

PushConfig: PushConfig{

Endpoint: "https://" + projID + ".appspot.com/_ah/push-handlers/push",

AuthenticationMethod: &OIDCToken{

Audience: "client-12345",

ServiceAccountEmail: serviceAccountEmail,

},

},

}

if sub, err = client.CreateSubscription(ctx, subIDs.New(), sCfg); 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,

ExpirationPolicy: defaultExpirationPolicy,

PushConfig: PushConfig{

Endpoint: "https://" + projID + ".appspot.com/_ah/push-handlers/push",

AuthenticationMethod: &OIDCToken{

Audience: "client-12345",

ServiceAccountEmail: serviceAccountEmail,

},

},

State: SubscriptionStateActive,

}

opt := cmpopts.IgnoreUnexported(SubscriptionConfig{})

if diff := testutil.Diff(got, want, opt); diff != "" {

t.Fatalf("\ngot: - want: +\n%s", diff)

}

// Add a PushConfig and change other fields.

pc := PushConfig{

Endpoint: "https://" + projID + ".appspot.com/_ah/push-handlers/push",

Attributes: map[string]string{"x-goog-version": "v1"},

AuthenticationMethod: &OIDCToken{

Audience: "client-updated-54321",

ServiceAccountEmail: serviceAccountEmail,

},

}

got, err = sub.Update(ctx, SubscriptionConfigToUpdate{

PushConfig: &pc,

AckDeadline: 2 * time.Minute,

RetainAckedMessages: true,

RetentionDuration: 2 * time.Hour,

Labels: map[string]string{"label": "value"},

ExpirationPolicy: 25 * time.Hour,

})

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"},

ExpirationPolicy: 25 * time.Hour,

State: SubscriptionStateActive,

}

if !testutil.Equal(got, want, opt) {

t.Fatalf("\ngot %+v\nwant %+v", got, want)

}

// Update ExpirationPolicy to never expire.

got, err = sub.Update(ctx, SubscriptionConfigToUpdate{

ExpirationPolicy: time.Duration(0),

})

if err != nil {

t.Fatal(err)

}

want.ExpirationPolicy = time.Duration(0)

if !testutil.Equal(got, want, opt) {

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, opt) {

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")

}

res := topic.Publish(ctx, msg)

_, err := res.Get(ctx)

if err != nil {

t.Fatalf("publishSync err: %v", err)

}

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.Delete(ctx)

defer topic.Stop()

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)

// Set ExpirationPolicy within the valid range.

got, err := sub.Update(ctx, SubscriptionConfigToUpdate{

RetentionDuration: 2 * time.Hour,

ExpirationPolicy: 25 * time.Hour,

AckDeadline: 2 * time.Minute,

})

if err != nil {

t.Fatal(err)

}

want := 25 * time.Hour

if got.ExpirationPolicy != want {

t.Fatalf("config.ExpirationPolicy mismatch; got: %v, want: %v", got.ExpirationPolicy, want)

}

// ExpirationPolicy to never expire.

got, err = sub.Update(ctx, SubscriptionConfigToUpdate{

ExpirationPolicy: time.Duration(0),

})

if err != nil {

t.Fatalf("Unexpected error: %v\n", err)

}

want = time.Duration(0)

if diff := testutil.Diff(got.ExpirationPolicy, want); diff != "" {

t.Fatalf("\ngot: - want: +\n%s", diff)

}

// ExpirationPolicy when nil is passed in, should not cause any updates.

got, err = sub.Update(ctx, SubscriptionConfigToUpdate{

ExpirationPolicy: nil,

})

if err == nil || err.Error() != "pubsub: UpdateSubscription call with nothing to update" {

t.Fatalf("Expected no attributes to be updated, error: %v", err)

}

// ExpirationPolicy of nil, with the previous value having been a non-zero value.

_, err = sub.Update(ctx, SubscriptionConfigToUpdate{

ExpirationPolicy: 26 * time.Hour,

})

if err != nil {

t.Fatal(err)

}

// Now examine what setting it to nil produces.

_, err = sub.Update(ctx, SubscriptionConfigToUpdate{

ExpirationPolicy: nil,

})

if err == nil || err.Error() != "pubsub: UpdateSubscription call with nothing to update" {

t.Fatalf("Expected no attributes to be updated, error: %v", err)

}

t.Parallel()

ctx := context.Background()

client := integrationTestClient(ctx, t)

defer client.Close()

compareConfig := func(got TopicConfig, wantLabels map[string]string) bool {

return testutil.Equal(got.Labels, wantLabels)

}

topic, err := client.CreateTopic(ctx, topicIDs.New())

if err != nil {

t.Fatalf("CreateTopic error: %v", err)

}

defer topic.Delete(ctx)

defer topic.Stop()

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)

}

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)

}

sub.Delete(ctx)

// 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.Delete(ctx)

defer topic.Stop()

// Out-of-range retention duration.

sub, err := client.CreateSubscription(ctx, subIDs.New(), SubscriptionConfig{

Topic: topic,

RetentionDuration: 1 * time.Second,

})

if want := codes.InvalidArgument; status.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; status.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; status.Code(err) != want {

t.Errorf("got <%v>, want %s", err, want)

}

// Deleting a non-existent subscription.

err = sub.Delete(ctx)

if want := codes.NotFound; status.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; status.Code(err) != want {

t.Errorf("got <%v>, want %s", err, want)

}

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.Delete(ctx)

defer topic.Stop()

// Specify some regions to set.

regions := []string{"asia-east1", "us-east1"}

cfg, err := topic.Update(ctx, TopicConfigToUpdate{

MessageStoragePolicy: &MessageStoragePolicy{

AllowedPersistenceRegions: regions,

},

})

if err != nil {

t.Fatal(err)

}

got := cfg.MessageStoragePolicy.AllowedPersistenceRegions

want := regions

if !testutil.Equal(got, want) {

t.Fatalf("\ngot %+v\nwant regions%+v", got, want)

}

// Removing all regions should fail

updateCfg := TopicConfigToUpdate{

MessageStoragePolicy: &MessageStoragePolicy{

AllowedPersistenceRegions: []string{},

},

}

if _, err = topic.Update(ctx, updateCfg); err == nil {

t.Fatalf("Unexpected succeeded in removing all regions\n%+v\n", got)

}

// Verify that the message storage policy is populated.

if testing.Short() {

t.Skip("Integration tests skipped in short mode")

}

t.Parallel()

ctx := context.Background()

// If a message storage policy is not set on a topic, the policy depends on the Resource Location

// Restriction which is specified on an organization level. The usual testing project is in the

// google.com org, which has no resource location restrictions. 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")