added or modified smoke test of four sources and gcp broker (#1161)

* modified smoke test of CloudPubSubSource

* fixed unit tests and renamed methods

* refactored duplicate code

* refactored the code

* added smoke test for storage schcheduler and pubsub sources

* removed redundant else

* added smoke test of gcp broker

* renaming

* added constants

* fixed typo

pkg/reconciler/events/auditlogs/auditlogs_test.go

@@ -661,6 +661,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceSink(sinkGVK, sinkName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -717,6 +718,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceSink(sinkGVK, sinkName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -773,6 +775,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceSink(sinkGVK, sinkName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -829,6 +832,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceSink(sinkGVK, sinkName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -896,6 +900,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceMethodName(testMethodName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -963,6 +968,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceMethodName(testMethodName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -1025,6 +1031,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceMethodName(testMethodName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),
@@ -1090,6 +1097,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudAuditLogsSourceServiceName(testServiceName),
 				WithCloudAuditLogsSourceSink(sinkGVK, sinkName),
 				WithCloudAuditLogsSourceProjectID(testProject),
+				WithCloudAuditLogsSourceSubscriptionID(SubscriptionID),
 				WithInitCloudAuditLogsSourceConditions,
 				WithCloudAuditLogsSourceTopicReady(testTopicID),
 				WithCloudAuditLogsSourcePullSubscriptionReady(),

pkg/reconciler/events/build/build_test.go

@@ -336,6 +336,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudBuildSourceConditions,
 					WithCloudBuildSourcePullSubscriptionReady(),
 					WithCloudBuildSourceSinkURI(pubsubSinkURL),
+					WithCloudBuildSourceSubscriptionID(SubscriptionID),
 				),
 			}, {
 				Object: NewCloudBuildSource(buildName, testNS,
@@ -346,6 +347,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudBuildSourceConditions,
 					WithCloudBuildSourcePullSubscriptionReady(),
 					WithCloudBuildSourceSinkURI(pubsubSinkURL),
+					WithCloudBuildSourceSubscriptionID(SubscriptionID),
 					WithCloudBuildSourceFinalizers("cloudbuildsources.events.cloud.google.com"),
 				),
 			}},

pkg/reconciler/events/pubsub/pubsub_test.go

@@ -335,6 +335,7 @@ func TestAllCases(t *testing.T) {
 				WithInitCloudPubSubSourceConditions,
 				WithCloudPubSubSourcePullSubscriptionReady(),
 				WithCloudPubSubSourceSinkURI(pubsubSinkURL),
+				WithCloudPubSubSourceSubscriptionID(SubscriptionID),
 			),
 		}, {
 			Object: NewCloudPubSubSource(pubsubName, testNS,
@@ -345,6 +346,7 @@ func TestAllCases(t *testing.T) {
 				WithInitCloudPubSubSourceConditions,
 				WithCloudPubSubSourcePullSubscriptionReady(),
 				WithCloudPubSubSourceSinkURI(pubsubSinkURL),
+				WithCloudPubSubSourceSubscriptionID(SubscriptionID),
 				WithCloudPubSubSourceFinalizers("cloudpubsubsources.events.cloud.google.com"),
 			),
 		}},

pkg/reconciler/events/scheduler/scheduler_test.go

@@ -705,6 +705,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobNotReady(reconciledFailedReason, fmt.Sprintf("%s: %s", failedToReconcileJobMsg, "create-client-induced-error")),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL)),
 			}},
@@ -766,6 +767,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobNotReady(reconciledFailedReason, fmt.Sprintf("%s: %s", failedToReconcileJobMsg, "get-job-induced-error")),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL)),
 			}},
@@ -827,6 +829,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobNotReady(reconciledFailedReason, fmt.Sprintf("%s: rpc error: code = %s desc = %s", failedToReconcileJobMsg, codes.Unknown, "get-job-induced-error")),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL)),
 			}},
@@ -889,6 +892,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobNotReady(reconciledFailedReason, fmt.Sprintf("%s: %s", failedToReconcileJobMsg, "create-job-induced-error")),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL)),
 			}},
@@ -950,6 +954,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobReady(jobName),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL)),
 			}},
@@ -1006,6 +1011,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobReady(jobName),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL)),
 			}},
@@ -1028,6 +1034,7 @@ func TestAllCases(t *testing.T) {
 					WithInitCloudSchedulerSourceConditions,
 					WithCloudSchedulerSourceTopicReady(testTopicID, testProject),
 					WithCloudSchedulerSourcePullSubscriptionReady(),
+					WithCloudSchedulerSourceSubscriptionID(SubscriptionID),
 					WithCloudSchedulerSourceJobReady(jobName),
 					WithCloudSchedulerSourceSinkURI(schedulerSinkURL),
 					WithCloudSchedulerSourceDeletionTimestamp,

pkg/reconciler/events/storage/storage_test.go

@@ -705,6 +705,7 @@ func TestAllCases(t *testing.T) {
 				WithCloudStorageSourceTopicReady(testTopicID),
 				WithCloudStorageSourceProjectID(testProject),
 				WithCloudStorageSourcePullSubscriptionReady(),
+				WithCloudStorageSourceSubscriptionID(SubscriptionID),
 				WithCloudStorageSourceSinkURI(storageSinkURL),
 				WithCloudStorageSourceNotificationNotReady(reconciledNotificationFailed, fmt.Sprintf("%s: %s", failedToReconcileNotificationMsg, "create-client-induced-error")),
 			),
@@ -773,6 +774,7 @@ func TestAllCases(t *testing.T) {
 					WithCloudStorageSourceTopicReady(testTopicID),
 					WithCloudStorageSourceProjectID(testProject),
 					WithCloudStorageSourcePullSubscriptionReady(),
+					WithCloudStorageSourceSubscriptionID(SubscriptionID),
 					WithCloudStorageSourceSinkURI(storageSinkURL),
 					WithCloudStorageSourceNotificationNotReady(reconciledNotificationFailed, fmt.Sprintf("%s: %s", failedToReconcileNotificationMsg, "bucket-notifications-induced-error")),
 				),
@@ -840,6 +842,7 @@ func TestAllCases(t *testing.T) {
 					WithCloudStorageSourceTopicReady(testTopicID),
 					WithCloudStorageSourceProjectID(testProject),
 					WithCloudStorageSourcePullSubscriptionReady(),
+					WithCloudStorageSourceSubscriptionID(SubscriptionID),
 					WithCloudStorageSourceSinkURI(storageSinkURL),
 					WithCloudStorageSourceNotificationNotReady(reconciledNotificationFailed, fmt.Sprintf("%s: %s", failedToReconcileNotificationMsg, "bucket-add-notification-induced-error")),
 				),
@@ -907,6 +910,7 @@ func TestAllCases(t *testing.T) {
 					WithCloudStorageSourceTopicReady(testTopicID),
 					WithCloudStorageSourceProjectID(testProject),
 					WithCloudStorageSourcePullSubscriptionReady(),
+					WithCloudStorageSourceSubscriptionID(SubscriptionID),
 					WithCloudStorageSourceSinkURI(storageSinkURL),
 					WithCloudStorageSourceNotificationReady(notificationId)),
 			}},

pkg/reconciler/intevents/reconciler.go

@@ -184,6 +184,7 @@ func (psb *PubSubBase) ReconcilePullSubscription(ctx context.Context, pubsubable
 		return ps, pkgreconciler.NewEvent(corev1.EventTypeWarning, PullSubscriptionStatusPropagateFailedReason, "Failed to propagate PullSubscription status: %s", err.Error())
 	}
 
+	status.SubscriptionID = ps.Status.SubscriptionID
 	status.SinkURI = ps.Status.SinkURI
 	return ps, nil
 }

pkg/reconciler/testing/auditlogs.go

@@ -133,6 +133,12 @@ func WithCloudAuditLogsSourceProjectID(projectID string) CloudAuditLogsSourceOpt
 	}
 }
 
+func WithCloudAuditLogsSourceSubscriptionID(subscriptionID string) CloudAuditLogsSourceOption {
+	return func(s *v1alpha1.CloudAuditLogsSource) {
+		s.Status.SubscriptionID = subscriptionID
+	}
+}
+
 func WithCloudAuditLogsSourceSinkID(sinkID string) CloudAuditLogsSourceOption {
 	return func(s *v1alpha1.CloudAuditLogsSource) {
 		s.Status.StackdriverSink = sinkID

pkg/reconciler/testing/build.go

@@ -133,6 +133,12 @@ func WithCloudBuildSourceSinkURI(url *apis.URL) CloudBuildSourceOption {
 	}
 }
 
+func WithCloudBuildSourceSubscriptionID(subscriptionID string) CloudBuildSourceOption {
+	return func(bs *v1alpha1.CloudBuildSource) {
+		bs.Status.SubscriptionID = subscriptionID
+	}
+}
+
 func WithCloudBuildSourceFinalizers(finalizers ...string) CloudBuildSourceOption {
 	return func(bs *v1alpha1.CloudBuildSource) {
 		bs.Finalizers = finalizers

pkg/reconciler/testing/pubsub.go

@@ -131,6 +131,12 @@ func WithCloudPubSubSourceSinkURI(url *apis.URL) CloudPubSubSourceOption {
 	}
 }
 
+func WithCloudPubSubSourceSubscriptionID(subscriptionID string) CloudPubSubSourceOption {
+	return func(ps *v1alpha1.CloudPubSubSource) {
+		ps.Status.SubscriptionID = subscriptionID
+	}
+}
+
 func WithCloudPubSubSourceFinalizers(finalizers ...string) CloudPubSubSourceOption {
 	return func(ps *v1alpha1.CloudPubSubSource) {
 		ps.Finalizers = finalizers

pkg/reconciler/testing/pullsubscription.go

@@ -33,6 +33,9 @@ import (
 // PullSubscriptionOption enables further configuration of a PullSubscription.
 type PullSubscriptionOption func(*v1alpha1.PullSubscription)
 
+const (
+	SubscriptionID = "subID"
+)
 // NewPullSubscription creates a PullSubscription with PullSubscriptionOptions
 func NewPullSubscription(name, namespace string, so ...PullSubscriptionOption) *v1alpha1.PullSubscription {
 	s := &v1alpha1.PullSubscription{
@@ -189,7 +192,7 @@ func WithPullSubscriptionReady(sink *apis.URL) PullSubscriptionOption {
 		s.Status.InitializeConditions()
 		s.Status.MarkSink(sink)
 		s.Status.MarkDeployed()
-		s.Status.MarkSubscribed("subID")
+		s.Status.MarkSubscribed(SubscriptionID)
 	}
 }
 

pkg/reconciler/testing/scheduler.go

@@ -185,6 +185,12 @@ func WithCloudSchedulerSourceSinkURI(url *apis.URL) CloudSchedulerSourceOption {
 	}
 }
 
+func WithCloudSchedulerSourceSubscriptionID(subscriptionID string) CloudSchedulerSourceOption {
+	return func(s *v1alpha1.CloudSchedulerSource) {
+		s.Status.SubscriptionID = subscriptionID
+	}
+}
+
 // WithCloudSchedulerSourceJobName sets the status for job Name
 func WithCloudSchedulerSourceJobName(jobName string) CloudSchedulerSourceOption {
 	return func(s *v1alpha1.CloudSchedulerSource) {

pkg/reconciler/testing/storage.go

@@ -204,6 +204,12 @@ func WithCloudStorageSourceProjectID(projectID string) CloudStorageSourceOption
 	}
 }
 
+func WithCloudStorageSourceSubscriptionID(subscriptionID string) CloudStorageSourceOption {
+	return func(s *v1alpha1.CloudStorageSource) {
+		s.Status.SubscriptionID = subscriptionID
+	}
+}
+
 func WithCloudStorageSourceStatusObservedGeneration(generation int64) CloudStorageSourceOption {
 	return func(s *v1alpha1.CloudStorageSource) {
 		s.Status.Status.ObservedGeneration = generation

test/e2e/e2e_test.go

@@ -174,7 +174,7 @@ func TestPullSubscriptionWithTarget(t *testing.T) {
 	PullSubscriptionWithTargetTestImpl(t, authConfig)
 }
 
-// TestSmokeCloudPubSubSource makes sure we can run tests on the CloudPubSubSource.
+// TestSmokeCloudPubSubSource we can create a CloudPubSubSource to ready state and we can delete a CloudPubSubSource and its underlying resources.
 func TestSmokeCloudPubSubSource(t *testing.T) {
 	cancel := logstream.Start(t)
 	defer cancel()
@@ -262,6 +262,13 @@ func TestCloudSchedulerSourceBrokerWithPubSubChannel(t *testing.T) {
 	SchedulerSourceBrokerWithPubSubChannelTestImpl(t, authConfig)
 }
 
+// TestSmokeCloudStorageSource tests if we can create a CloudStorageSource to ready state and delete a CloudStorageSource and its underlying resources.
+func TestSmokeCloudStorageSource(t *testing.T) {
+	cancel := logstream.Start(t)
+	defer cancel()
+	SmokeCloudStorageSourceTestImpl(t, authConfig)
+}
+
 // TestCloudStorageSourceWithTarget tests we can knock down a target from a CloudStorageSource.
 func TestCloudStorageSourceWithTarget(t *testing.T) {
 	cancel := logstream.Start(t)
@@ -277,18 +284,25 @@ func TestCloudStorageSourceStackDriverMetrics(t *testing.T) {
 	CloudStorageSourceWithTargetTestImpl(t, true /*assertMetrics */, authConfig)
 }
 
+// TestSmokeCloudAuditLogsSource tests if we can create a CloudAuditLogsSource to ready state and delete a CloudAuditLogsSource and its underlying resources.
+func TestSmokeCloudAuditLogsSource(t *testing.T) {
+	cancel := logstream.Start(t)
+	defer cancel()
+	SmokeCloudAuditLogsSourceTestImpl(t, authConfig)
+}
+
 // TestCloudAuditLogsSource tests we can knock down a target from an CloudAuditLogsSource.
 func TestCloudAuditLogsSourceWithTarget(t *testing.T) {
 	cancel := logstream.Start(t)
 	defer cancel()
 	CloudAuditLogsSourceWithTargetTestImpl(t, authConfig)
 }
 
-// TestSmokeCloudSchedulerSourceSetup tests if we can create a CloudSchedulerSource resource and get it to a ready state.
-func TestSmokeCloudSchedulerSourceSetup(t *testing.T) {
+// TestSmokeCloudSchedulerSource tests if we can create a CloudSchedulerSource to ready state and delete a CloudSchedulerSource and its underlying resources.
+func TestSmokeCloudSchedulerSource(t *testing.T) {
 	cancel := logstream.Start(t)
 	defer cancel()
-	SmokeCloudSchedulerSourceSetup(t, authConfig)
+	SmokeCloudSchedulerSourceTestImpl(t, authConfig)
 }
 
 // TestCloudSchedulerSourceWithTargetTestImpl tests if we can receive an event on a bespoke sink from a CloudSchedulerSource source.
@@ -298,6 +312,14 @@ func TestCloudSchedulerSourceWithTargetTestImpl(t *testing.T) {
 	CloudSchedulerSourceWithTargetTestImpl(t, authConfig)
 }
 
+// TestSmokeGCPBroker tests if we can create a GCPBroker to ready state and delete a GCPBroker and its underlying resources.
+func TestSmokeGCPBroker(t *testing.T) {
+	cancel := logstream.Start(t)
+	defer cancel()
+	SmokeGCPBrokerTestImpl(t, authConfig)
+}
+
+
 // TestGCPBroker tests we can knock a Knative Service from a gcp broker.
 func TestGCPBroker(t *testing.T) {
 	cancel := logstream.Start(t)

test/e2e/lib/auditlogs.go

@@ -17,11 +17,17 @@ limitations under the License.
 package lib
 
 import (
+	"context"
 	"fmt"
+	"google.golang.org/grpc/status"
+	"os"
+	"testing"
 
+	"cloud.google.com/go/logging/logadmin"
 	"github.com/google/knative-gcp/pkg/apis/events/v1alpha1"
 	kngcptesting "github.com/google/knative-gcp/pkg/reconciler/testing"
 	"github.com/google/knative-gcp/test/e2e/lib/resources"
+	"google.golang.org/grpc/codes"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -75,3 +81,24 @@ func MakeAuditLogsJobOrDie(client *Client, methodName, project, resourceName, se
 	}})
 	client.CreateJobOrFail(job, WithServiceForJob(targetName))
 }
+
+func StackdriverSinkExists(t *testing.T, sinkID string) bool {
+	t.Helper()
+	ctx := context.Background()
+	project := os.Getenv(ProwProjectKey)
+	client, err := logadmin.NewClient(ctx, project)
+	if err != nil {
+		t.Fatalf("failed to create LogAdmin client, %s", err.Error())
+	}
+	defer client.Close()
+
+	_, err = client.Sink(ctx, sinkID)
+	if err != nil {
+		if status.Code(err) == codes.NotFound {
+			return false
+		}
+
+		t.Fatalf("Failed from LogAdmin client while retrieving StackdriverSink %s with error %s", sinkID, err.Error())
+	}
+	return true
+}