Merge pull request #49 from barney-s/master

Adding more tests (postgres, cloudsql)

Makefile

@@ -72,5 +72,6 @@ docker-push: docker-build
 
 e2e-test:
 	kubectl get namespace airflowop-system || kubectl create namespace airflowop-system
+	kubectl apply -f hack/sample/cloudsql-celery/sqlproxy-secret.yaml -n airflowop-system
 	go test -v -timeout 20m test/e2e/base_test.go --namespace airflowop-system
 	go test -v -timeout 20m test/e2e/cluster_test.go --namespace airflowop-system

pkg/apis/airflow/v1alpha1/airflow.go

@@ -261,6 +261,7 @@ USE $(SQL_DB);
 CREATE USER IF NOT EXISTS '$(SQL_USER)'@'%' IDENTIFIED BY '$(SQL_PASSWORD)';
 GRANT ALL ON $(SQL_DB).* TO '$(SQL_USER)'@'%' ;
 FLUSH PRIVILEGES;
+SHOW GRANTS FOR $(SQL_USER);
 EOSQL
 `},
 		},

test/e2e/base_test.go

@@ -71,6 +71,18 @@ var _ = Describe(CRName+" controller tests", func() {
 		ctx = nil
 	})
 
+	It("creating a "+CRName+" with cloudsql", func() {
+		ctx = f.NewContext().WithCR(airflowBase(SampleDir + "cloudsql-celery/base.yaml"))
+		cr := ctx.CR.(*v1alpha1.AirflowBase)
+		By("creating a new " + CRName + ": " + cr.Name)
+		ctx.CreateCR()
+		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-cloudsql", 1, 1)
+		ctx.WithTimeout(10).CheckService(cr.Name+"-sql", map[string]int32{"mysql": 3306})
+		//ctx.WithTimeout(10).CheckSecret(name)
+		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-nfs", 1, 1)
+		ctx.WithTimeout(100).CheckCR(isBaseReady)
+	})
+
 	It("creating a "+CRName+" with mysql", func() {
 		ctx = f.NewContext().WithCR(airflowBase(SampleDir + "mysql-celery/base.yaml"))
 		cr := ctx.CR.(*v1alpha1.AirflowBase)

test/e2e/cluster_test.go

@@ -30,6 +30,7 @@ const (
 
 var f *test.Framework
 var ctx, basectx *test.Context
+var deleteBase bool
 
 func airflowBase(file string) *v1alpha1.AirflowBase {
 	cr := &v1alpha1.AirflowBase{}
@@ -80,12 +81,84 @@ func isClusterReady(cr interface{}) bool {
 	return stts.IsReady()
 }
 
+func checkLocal(cr *v1alpha1.AirflowCluster) {
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-airflowui", 1, 1)
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-scheduler", 1, 1)
+	ctx.WithTimeout(200).CheckCR(isClusterReady)
+}
+func checkCelery(cr *v1alpha1.AirflowCluster) {
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-airflowui", 1, 1)
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-flower", 1, 1)
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-redis", 1, 1)
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-scheduler", 1, 1)
+	ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-worker", 2, 1)
+	ctx.WithTimeout(10).CheckService(cr.Name+"-redis", map[string]int32{"redis": 6379})
+	ctx.WithTimeout(200).CheckCR(isClusterReady)
+}
+
 var _ = Describe(CRName+" controller tests", func() {
 	AfterEach(func() {
 		ctx.DeleteCR()
+		if deleteBase {
+			deleteBase = false
+			basectx.DeleteCR()
+		}
 		ctx = nil
 	})
 
+	// CloudSQL
+	// BROKEN  TODO
+	// https://github.com/GoogleCloudPlatform/airflow-operator/issues/48<Paste>
+	/*
+		It("creating a "+CRName+" with cloudsql, celery executor", func() {
+			basectx = f.NewContext().WithCR(airflowBase(SampleDir + "cloudsql-celery/base.yaml"))
+			ctx = f.NewContext().WithCR(airflowCluster(SampleDir + "cloudsql-celery/cluster.yaml"))
+			basecr := basectx.CR.(*v1alpha1.AirflowBase)
+			cr := ctx.CR.(*v1alpha1.AirflowCluster)
+			By("creating a base " + basecr.Name)
+			basectx.CreateCR()
+			basectx.WithTimeout(200).CheckCR(isBaseReady)
+
+			By("creating a new " + CRName + ": " + cr.Name)
+			ctx.CreateCR()
+			checkCelery(cr)
+		})
+
+		It("creating a "+CRName+" with cloudsql, local executor", func() {
+			ctx = f.NewContext().WithCR(airflowCluster(SampleDir + "cloudsql-local/cluster.yaml"))
+			cr := ctx.CR.(*v1alpha1.AirflowCluster)
+			By("creating a new " + CRName + ": " + cr.Name)
+			ctx.CreateCR()
+			checkLocal(cr)
+			deleteBase = true
+		})
+	*/
+
+	// Postgres
+	It("creating a "+CRName+" with postgres, celery executor", func() {
+		basectx = f.NewContext().WithCR(airflowBase(SampleDir + "postgres-celery/base.yaml"))
+		ctx = f.NewContext().WithCR(airflowCluster(SampleDir + "postgres-celery/cluster.yaml"))
+		basecr := basectx.CR.(*v1alpha1.AirflowBase)
+		cr := ctx.CR.(*v1alpha1.AirflowCluster)
+		By("creating a base " + basecr.Name)
+		basectx.CreateCR()
+		basectx.WithTimeout(200).CheckCR(isBaseReady)
+
+		By("creating a new " + CRName + ": " + cr.Name)
+		ctx.CreateCR()
+		checkCelery(cr)
+	})
+
+	It("creating a "+CRName+" with postgres, local executor", func() {
+		ctx = f.NewContext().WithCR(airflowCluster(SampleDir + "postgres-local/cluster.yaml"))
+		cr := ctx.CR.(*v1alpha1.AirflowCluster)
+		By("creating a new " + CRName + ": " + cr.Name)
+		ctx.CreateCR()
+		checkLocal(cr)
+		deleteBase = true
+	})
+
+	// Mysql
 	It("creating a "+CRName+" with mysql, celery executor", func() {
 		basectx = f.NewContext().WithCR(airflowBase(SampleDir + "mysql-celery/base.yaml"))
 		ctx = f.NewContext().WithCR(airflowCluster(SampleDir + "mysql-celery/cluster.yaml"))
@@ -97,22 +170,16 @@ var _ = Describe(CRName+" controller tests", func() {
 
 		By("creating a new " + CRName + ": " + cr.Name)
 		ctx.CreateCR()
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-airflowui", 1, 1)
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-flower", 1, 1)
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-redis", 1, 1)
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-scheduler", 1, 1)
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-worker", 2, 1)
-		ctx.WithTimeout(10).CheckService(cr.Name+"-redis", map[string]int32{"redis": 6379})
-		ctx.WithTimeout(200).CheckCR(isClusterReady)
+		checkCelery(cr)
 	})
 
 	It("creating a "+CRName+" with mysql, local executor", func() {
 		ctx = f.NewContext().WithCR(airflowCluster(SampleDir + "mysql-local/cluster.yaml"))
 		cr := ctx.CR.(*v1alpha1.AirflowCluster)
 		By("creating a new " + CRName + ": " + cr.Name)
 		ctx.CreateCR()
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-airflowui", 1, 1)
-		ctx.WithTimeout(200).CheckStatefulSet(cr.Name+"-scheduler", 1, 1)
-		ctx.WithTimeout(200).CheckCR(isClusterReady)
+		checkLocal(cr)
+		deleteBase = true
 	})
+
 })