fix(PubSub): fix test failures (#3079)

Fixes: #3076

Author: cy-yun

pubsub/api/Pubsub.Samples.Tests/CreateTopicWithAwsMskIngestionTest.cs

@@ -12,36 +12,28 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Google.Cloud.PubSub.V1;
+using Grpc.Core;
 using Xunit;
 
 [Collection(nameof(PubsubFixture))]
 public class CreateTopicWithAwsMskIngestionTest
 {
-  private readonly PubsubFixture _pubsubFixture;
-  private readonly CreateTopicWithAwsMskIngestionSample _createTopicWithAwsMskIngestionSample;
+    private readonly PubsubFixture _pubsubFixture;
+    private readonly CreateTopicWithAwsMskIngestionSample _createTopicWithAwsMskIngestionSample;
 
-  public CreateTopicWithAwsMskIngestionTest(PubsubFixture pubsubFixture)
-  {
-    _pubsubFixture = pubsubFixture;
-    _createTopicWithAwsMskIngestionSample = new CreateTopicWithAwsMskIngestionSample();
-  }
+    public CreateTopicWithAwsMskIngestionTest(PubsubFixture pubsubFixture)
+    {
+        _pubsubFixture = pubsubFixture;
+        _createTopicWithAwsMskIngestionSample = new CreateTopicWithAwsMskIngestionSample();
+    }
 
-  [Fact]
-  public void CreateTopicWithAwsMskIngestion()
-  {
-    string topicId = _pubsubFixture.RandomTopicId();
-    var (clusterArn, mskTopic, awsRoleArn, gcpServiceAccount) = _pubsubFixture.RandomAwsMskIngestionParams();
-    Topic createdTopic = _createTopicWithAwsMskIngestionSample.CreateTopicWithAwsMskIngestion(_pubsubFixture.ProjectId, topicId, clusterArn, mskTopic, awsRoleArn, gcpServiceAccount);
-
-    // Confirm that the created topic and topic retrieved by ID are equal
-    Topic retrievedTopic = _pubsubFixture.GetTopic(topicId);
-    Assert.Equal(createdTopic, retrievedTopic);
-
-    // Confirm that all Amazon MSK Ingestion params are equal to expected values
-    Assert.Equal(clusterArn, createdTopic.IngestionDataSourceSettings.AwsMsk.ClusterArn);
-    Assert.Equal(mskTopic, createdTopic.IngestionDataSourceSettings.AwsMsk.Topic);
-    Assert.Equal(awsRoleArn, createdTopic.IngestionDataSourceSettings.AwsMsk.AwsRoleArn);
-    Assert.Equal(gcpServiceAccount, createdTopic.IngestionDataSourceSettings.AwsMsk.GcpServiceAccount);
-  }
-}
+    [Fact]
+    public void CreateTopicWithAwsMskIngestion()
+    {
+        string topicId = _pubsubFixture.RandomTopicId();
+        var (clusterArn, mskTopic, awsRoleArn, gcpServiceAccount) = _pubsubFixture.AwsMskIngestionParams();
+        var exception = Assert.Throws<RpcException>(() => _createTopicWithAwsMskIngestionSample.CreateTopicWithAwsMskIngestion(_pubsubFixture.ProjectId, topicId, clusterArn, mskTopic, awsRoleArn, gcpServiceAccount));
+        Assert.Equal(StatusCode.PermissionDenied, exception.Status.StatusCode);
+        Assert.Equal(_pubsubFixture.PermissionDeniedMessage, exception.Status.Detail);
+    }
+}

pubsub/api/Pubsub.Samples.Tests/CreateTopicWithAzureEventHubsIngestionTest.cs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Google.Cloud.PubSub.V1;
+using Grpc.Core;
 using Xunit;
 
 [Collection(nameof(PubsubFixture))]
@@ -31,20 +31,9 @@ public CreateTopicWithAzureEventHubsIngestionTest(PubsubFixture pubsubFixture)
     public void CreateTopicWithAzureEventHubsIngestion()
     {
         string topicId = _pubsubFixture.RandomTopicId();
-        var (resourceGroup, nameSpace, eventHub, clientId, tenantId, subscriptionId, gcpServiceAccount) = _pubsubFixture.RandomAzureEventHubsIngestionParams();
-        Topic createdTopic = _createTopicWithAzureEventHubsIngestionSample.CreateTopicWithAzureEventHubsIngestion(_pubsubFixture.ProjectId, topicId, resourceGroup, nameSpace, eventHub, clientId, tenantId, subscriptionId, gcpServiceAccount);
-
-        // Confirm that the created topic and topic retrieved by ID are equal
-        Topic retrievedTopic = _pubsubFixture.GetTopic(topicId);
-        Assert.Equal(createdTopic, retrievedTopic);
-
-        // Confirm that all Ingestion params are equal to expected values
-        Assert.Equal(resourceGroup, createdTopic.IngestionDataSourceSettings.AzureEventHubs.ResourceGroup);
-        Assert.Equal(nameSpace, createdTopic.IngestionDataSourceSettings.AzureEventHubs.Namespace);
-        Assert.Equal(eventHub, createdTopic.IngestionDataSourceSettings.AzureEventHubs.EventHub);
-        Assert.Equal(clientId, createdTopic.IngestionDataSourceSettings.AzureEventHubs.ClientId);
-        Assert.Equal(tenantId, createdTopic.IngestionDataSourceSettings.AzureEventHubs.TenantId);
-        Assert.Equal(subscriptionId, createdTopic.IngestionDataSourceSettings.AzureEventHubs.SubscriptionId);
-        Assert.Equal(gcpServiceAccount, createdTopic.IngestionDataSourceSettings.AzureEventHubs.GcpServiceAccount);
+        var (resourceGroup, nameSpace, eventHub, clientId, tenantId, subscriptionId, gcpServiceAccount) = _pubsubFixture.AzureEventHubsIngestionParams();
+        var exception = Assert.Throws<RpcException>(() => _createTopicWithAzureEventHubsIngestionSample.CreateTopicWithAzureEventHubsIngestion(_pubsubFixture.ProjectId, topicId, resourceGroup, nameSpace, eventHub, clientId, tenantId, subscriptionId, gcpServiceAccount));
+        Assert.Equal(StatusCode.NotFound, exception.Status.StatusCode);
+        Assert.Equal($"Cloud Pub/Sub encountered a not-found error while trying to connect to the ingestion data source: Failed to resolve bootstrap server {nameSpace}.servicebus.windows.net to an IP.", exception.Status.Detail);
     }
 }

pubsub/api/Pubsub.Samples.Tests/CreateTopicWithCloudStorageIngestionTest.cs

@@ -33,7 +33,7 @@ public CreateTopicWithCloudStorageIngestionTest(PubsubFixture pubsubFixture)
     public void CreateTopicWithCloudStorageIngestion()
     {
         string topicId = _pubsubFixture.RandomTopicId();
-        string bucket = _pubsubFixture.RandomName("Bucket");
+        string bucket = _pubsubFixture.CloudStorageBucketName;
         string inputFormat = "text";
         string textDelimiter = "\n";
         string matchGlob = "**.txt";

pubsub/api/Pubsub.Samples.Tests/CreateTopicWithConfluentCloudIngestionTest.cs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Google.Cloud.PubSub.V1;
+using Grpc.Core;
 using Xunit;
 
 [Collection(nameof(PubsubFixture))]
@@ -31,18 +31,9 @@ public CreateTopicWithConfluentCloudIngestionTest(PubsubFixture pubsubFixture)
     public void CreateTopicWithConfluentCloudIngestion()
     {
         string topicId = _pubsubFixture.RandomTopicId();
-        var (bootstrapServer, clusterId, confluentTopic, identityPoolId, gcpServiceAccount) = _pubsubFixture.RandomConfluentCloudIngestionParams();
-        Topic createdTopic = _createTopicWithConfluentCloudIngestionSample.CreateTopicWithConfluentCloudIngestion(_pubsubFixture.ProjectId, topicId, bootstrapServer, clusterId, confluentTopic, identityPoolId, gcpServiceAccount);
-
-        // Confirm that the created topic and topic retrieved by ID are equal
-        Topic retrievedTopic = _pubsubFixture.GetTopic(topicId);
-        Assert.Equal(createdTopic, retrievedTopic);
-
-        // Confirm that all Confluent Cloud Ingestion params are equal to expected values
-        Assert.Equal(bootstrapServer, createdTopic.IngestionDataSourceSettings.ConfluentCloud.BootstrapServer);
-        Assert.Equal(clusterId, createdTopic.IngestionDataSourceSettings.ConfluentCloud.ClusterId);
-        Assert.Equal(confluentTopic, createdTopic.IngestionDataSourceSettings.ConfluentCloud.Topic);
-        Assert.Equal(identityPoolId, createdTopic.IngestionDataSourceSettings.ConfluentCloud.IdentityPoolId);
-        Assert.Equal(gcpServiceAccount, createdTopic.IngestionDataSourceSettings.ConfluentCloud.GcpServiceAccount);
+        var (bootstrapServer, clusterId, confluentTopic, identityPoolId, gcpServiceAccount) = _pubsubFixture.ConfluentCloudIngestionParams();
+        var exception = Assert.Throws<RpcException>(() => _createTopicWithConfluentCloudIngestionSample.CreateTopicWithConfluentCloudIngestion(_pubsubFixture.ProjectId, topicId, bootstrapServer, clusterId, confluentTopic, identityPoolId, gcpServiceAccount));
+        Assert.Equal(StatusCode.InvalidArgument, exception.Status.StatusCode);
+        Assert.Equal("Cloud Pub/Sub received invalid argument/s for the ingestion data source: Unreachable bootstrap server.", exception.Status.Detail);
     }
 }

pubsub/api/Pubsub.Samples.Tests/CreateTopicWithKinesisIngestionTest.cs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Google.Cloud.PubSub.V1;
+using Grpc.Core;
 using Xunit;
 
 [Collection(nameof(PubsubFixture))]
@@ -31,17 +31,9 @@ public CreateTopicWithKinesisIngestionTest(PubsubFixture pubsubFixture)
     public void CreateTopicWithKinesisIngestion()
     {
         string topicId = _pubsubFixture.RandomTopicId();
-        var (streamArn, consumerArn, awsRoleArn, gcpServiceAccount) = _pubsubFixture.RandomKinesisIngestionParams();
-        Topic createdTopic = _createTopicWithKinesisIngestionSample.CreateTopicWithKinesisIngestion(_pubsubFixture.ProjectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
-
-        // Confirm that the created topic and topic retrieved by ID are equal
-        Topic retrievedTopic = _pubsubFixture.GetTopic(topicId);
-        Assert.Equal(createdTopic, retrievedTopic);
-
-        // Confirm that all AWSKinesis Ingestion params are equal to expected values
-        Assert.Equal(streamArn, createdTopic.IngestionDataSourceSettings.AwsKinesis.StreamArn);
-        Assert.Equal(consumerArn, createdTopic.IngestionDataSourceSettings.AwsKinesis.ConsumerArn);
-        Assert.Equal(awsRoleArn, createdTopic.IngestionDataSourceSettings.AwsKinesis.AwsRoleArn);
-        Assert.Equal(gcpServiceAccount, createdTopic.IngestionDataSourceSettings.AwsKinesis.GcpServiceAccount);
+        var (streamArn, consumerArn, awsRoleArn, gcpServiceAccount) = _pubsubFixture.KinesisIngestionParams();
+        var exception = Assert.Throws<RpcException>(() => _createTopicWithKinesisIngestionSample.CreateTopicWithKinesisIngestion(_pubsubFixture.ProjectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount));
+        Assert.Equal(StatusCode.PermissionDenied, exception.Status.StatusCode);
+        Assert.Equal(_pubsubFixture.PermissionDeniedMessage, exception.Status.Detail);
     }
 }

pubsub/api/Pubsub.Samples.Tests/OptimisticSubscribeTest.cs

@@ -22,12 +22,14 @@ public class OptimisticSubscribeTest
     private readonly PubsubFixture _pubsubFixture;
     private readonly PublishMessagesAsyncSample _publishMessagesAsyncSample;
     private readonly OptimisticSubscribeSample _optimisticSubscribeSample;
+    private readonly DeleteSubscriptionSample _deleteSubscriptionSample;
 
     public OptimisticSubscribeTest(PubsubFixture pubsubFixture)
     {
         _pubsubFixture = pubsubFixture;
         _publishMessagesAsyncSample = new PublishMessagesAsyncSample();
         _optimisticSubscribeSample = new OptimisticSubscribeSample();
+        _deleteSubscriptionSample = new DeleteSubscriptionSample();
     }
 
     [Fact]
@@ -38,10 +40,17 @@ public async Task SubscribeOptimistically()
 
         _pubsubFixture.CreateTopic(topicId);
 
+        // Publish messages and call OptimisticSubscribe before subscription creation
         await _publishMessagesAsyncSample.PublishMessagesAsync(_pubsubFixture.ProjectId, topicId, messages);
+        var result1 = await _optimisticSubscribeSample.OptimisticSubscribe(_pubsubFixture.ProjectId, topicId, subscriptionId);
+        // Existing messages before subscription creation would not be received
+        Assert.Equal(0, result1);
 
-        var result = await _optimisticSubscribeSample.OptimisticSubscribe(_pubsubFixture.ProjectId, topicId, subscriptionId);
+        await _publishMessagesAsyncSample.PublishMessagesAsync(_pubsubFixture.ProjectId, topicId, messages);
+        var result2 = await _optimisticSubscribeSample.OptimisticSubscribe(_pubsubFixture.ProjectId, topicId, subscriptionId);
+        // Messages published after subscription creation should be received
+        Assert.Equal(messages.Count, result2);
 
-        Assert.Equal(messages.Count, result);
+        _deleteSubscriptionSample.DeleteSubscription(_pubsubFixture.ProjectId, subscriptionId);
     }
 }

pubsub/api/Pubsub.Samples.Tests/PubsubFixture.cs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+using Google.Apis.Auth.OAuth2;
 using Google.Apis.Bigquery.v2.Data;
 using Google.Apis.Storage.v1.Data;
 using Google.Cloud.BigQuery.V2;
@@ -23,11 +24,12 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Runtime.CompilerServices;
+using System.Threading.Tasks;
 using Xunit;
 using Xunit.Sdk;
 
 [CollectionDefinition(nameof(PubsubFixture))]
-public class PubsubFixture : IDisposable, ICollectionFixture<PubsubFixture>
+public class PubsubFixture : IDisposable, ICollectionFixture<PubsubFixture>, IAsyncLifetime
 {
     public string ProjectId { get; }
     public List<string> TempTopicIds { get; } = new List<string>();
@@ -44,6 +46,9 @@ public class PubsubFixture : IDisposable, ICollectionFixture<PubsubFixture>
     public string BigQueryTableId { get; } = $"testTable{Guid.NewGuid().ToString().Substring(24)}";
     public string BigQueryTableName { get; }
     public string CloudStorageBucketName { get; }
+    public string PermissionDeniedMessage { get; } = "Cloud Pub/Sub does not have the necessary permissions settings on the ingestion data source: AccessDenied: Not authorized to perform sts:AssumeRoleWithWebIdentity.";
+    private const string AwsRoleArn = "arn:aws:iam::111111111111:role/fake-role-name";
+    private string _gcpServiceAccount = "";
 
     public RetryRobot Pull { get; } = new RetryRobot
     {
@@ -256,27 +261,27 @@ public List<Schema> ListSchemaRevisions(string schemaId)
         return ($"test{caller}Topic{randomName}", $"test{caller}Subscription{randomName}", $"test{caller}Schema{randomName}");
     }
 
-    public (string streamArn, string consumerArn, string awsRoleArn, string gcpServiceAccount) RandomKinesisIngestionParams([CallerMemberName] string caller = null)
-    {
-        var randomName = RandomName();
-        return ($"test{caller}StreamArn{randomName}", $"test{caller}ConsumerArn{randomName}", $"test{caller}AwsRoleArn{randomName}", $"test{caller}GcpServiceAccount{randomName}");
-    }
+    public (string streamArn, string consumerArn, string awsRoleArn, string gcpServiceAccount) KinesisIngestionParams() => ("arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name", "arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/consumer/consumer-1:1111111111", AwsRoleArn, _gcpServiceAccount);
 
-    public (string clusterArn, string mskTopic, string awsRoleArn, string gcpServiceAccount) RandomAwsMskIngestionParams([CallerMemberName] string caller = null)
-    {
-        var randomName = RandomName();
-        return ($"test{caller}ClusterArn{randomName}", $"test{caller}MskTopic{randomName}", $"test{caller}AwsRoleArn{randomName}", $"test{caller}GcpServiceAccount{randomName}");
-    }
+    public (string clusterArn, string mskTopic, string awsRoleArn, string gcpServiceAccount) AwsMskIngestionParams() => ("arn:aws:kafka:us-east-1:111111111111:cluster/fake-cluster-name/11111111-1111-1", "fake-msk-topic-name", AwsRoleArn, _gcpServiceAccount);
 
-    public (string bootstrapServer, string clusterId, string confluentTopic, string identityPoolId, string gcpServiceAccount) RandomConfluentCloudIngestionParams([CallerMemberName] string caller = null)
-    {
-        var randomName = RandomName();
-        return ($"test{caller}BootstrapServer{randomName}", $"test{caller}ClusterId{randomName}", $"test{caller}ConfluentTopic{randomName}", $"test{caller}IdentityPoolId{randomName}", $"test{caller}GcpServiceAccount{randomName}");
-    }
+    public (string bootstrapServer, string clusterId, string confluentTopic, string identityPoolId, string gcpServiceAccount) ConfluentCloudIngestionParams() => ("fake-bootstrap-server-id.us-south1.gcp.confluent.cloud:9092", "fake-cluster-id", "fake-confluent-topic-name", "fake-identity-pool-id", _gcpServiceAccount);
+
+    public (string resourceGroup, string nameSpace, string eventHub, string clientId, string tenantId, string subscriptionId, string gcpServiceAccount) AzureEventHubsIngestionParams() => ("fake-resource-group", "fake-namespace", "fake-event-hub", "11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222", "33333333-3333-3333-3333-333333333333", _gcpServiceAccount);
 
-    public (string resourceGroup, string nameSpace, string eventHub, string clientId, string tenantId, string subscriptionId, string gcpServiceAccount) RandomAzureEventHubsIngestionParams([CallerMemberName] string caller = null)
+    public async Task InitializeAsync()
     {
-        var randomName = RandomName();
-        return ($"test{caller}ResourceGroup{randomName}", $"test{caller}Namespace{randomName}", $"test{caller}EventHub{randomName}", $"test{caller}ClientId{randomName}", $"test{caller}TenantId{randomName}", $"test{caller}SubscriptionId{randomName}", $"test{caller}GcpServiceAccount{randomName}");
+        GoogleCredential appDefaultCredentials = GoogleCredential.GetApplicationDefault();
+        if (appDefaultCredentials.UnderlyingCredential is ServiceAccountCredential sac)
+        {
+            _gcpServiceAccount = sac.Id;
+        }
+        else if (appDefaultCredentials.UnderlyingCredential is ComputeCredential cc)
+        {
+            _gcpServiceAccount = await cc.GetDefaultServiceAccountEmailAsync();
+        }
+        return;
     }
+
+    public Task DisposeAsync() => Task.CompletedTask;
 }

pubsub/api/Pubsub.Samples.Tests/UpdateTopicTypeTest.cs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Google.Cloud.PubSub.V1;
+using Grpc.Core;
 using Xunit;
 
 [Collection(nameof(PubsubFixture))]
@@ -33,9 +33,9 @@ public void UpdateTopicType()
         string topicId = _pubsubFixture.RandomTopicId();
         _pubsubFixture.CreateTopic(topicId);
 
-        var (streamArn, consumerArn, awsRoleArn, gcpServiceAccount) = _pubsubFixture.RandomKinesisIngestionParams();
-        Topic updatedTopic = _updateTopicTypeSample.UpdateTopicType(_pubsubFixture.ProjectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
-        Topic retrievedTopic = _pubsubFixture.GetTopic(topicId);
-        Assert.Equal(updatedTopic, retrievedTopic);
+        var (streamArn, consumerArn, awsRoleArn, gcpServiceAccount) = _pubsubFixture.KinesisIngestionParams();
+        var exception = Assert.Throws<RpcException>(() => _updateTopicTypeSample.UpdateTopicType(_pubsubFixture.ProjectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount));
+        Assert.Equal(StatusCode.PermissionDenied, exception.Status.StatusCode);
+        Assert.Equal(_pubsubFixture.PermissionDeniedMessage, exception.Status.Detail);
     }
 }

pubsub/api/Pubsub.Samples/CreateTopicWithKinesisIngestion.cs

@@ -21,8 +21,6 @@ public class CreateTopicWithKinesisIngestionSample
 {
     public Topic CreateTopicWithKinesisIngestion(string projectId, string topicId, string streamArn, string consumerArn, string awsRoleArn, string gcpServiceAccount)
     {
-        PublisherServiceApiClient publisher = PublisherServiceApiClient.Create();
-
         // Define settings for Kinesis ingestion
         IngestionDataSourceSettings ingestionDataSourceSettings = new IngestionDataSourceSettings
         {
@@ -35,12 +33,12 @@ public Topic CreateTopicWithKinesisIngestion(string projectId, string topicId, s
             }
         };
 
+        PublisherServiceApiClient publisher = PublisherServiceApiClient.Create();
         Topic topic = new Topic()
         {
             Name = TopicName.FormatProjectTopic(projectId, topicId),
             IngestionDataSourceSettings = ingestionDataSourceSettings
         };
-
         Topic createdTopic = publisher.CreateTopic(topic);
         Console.WriteLine($"Topic {topic.Name} created.");