diff --git a/tests/e2e/asserts_test.go b/tests/e2e/asserts_test.go index 15c23609c0..0aeab445c8 100644 --- a/tests/e2e/asserts_test.go +++ b/tests/e2e/asserts_test.go @@ -32,7 +32,7 @@ import ( "github.com/thoas/go-funk" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" - v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/util/retry" "k8s.io/utils/strings/slices" @@ -504,56 +504,33 @@ func insertRecordIntoTable(tableName string, value int, conn *sql.DB) { Expect(err).ToNot(HaveOccurred()) } -// AssertDatabaseExists assert if database exists -func AssertDatabaseExists(pod *corev1.Pod, databaseName string, expectedValue bool) { - By(fmt.Sprintf("verifying if database %v exists", databaseName), func() { - query := fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_database WHERE lower(datname) = lower('%v'));", databaseName) - Eventually(func(g Gomega) { - stdout, stderr, err := env.ExecQueryInInstancePod( - testsUtils.PodLocator{ - Namespace: pod.Namespace, - PodName: pod.Name, - }, - testsUtils.PostgresDBName, - query) - if err != nil { - GinkgoWriter.Printf("stdout: %v\nstderr: %v", stdout, stderr) - } - g.Expect(err).ToNot(HaveOccurred()) - - if expectedValue { - g.Expect(strings.Trim(stdout, "\n")).To(BeEquivalentTo("t")) - } else { - g.Expect(strings.Trim(stdout, "\n")).To(BeEquivalentTo("f")) - } - }, 60).Should(Succeed()) - }) +func AssertQueryEventuallyMatchExpectation( + pod *corev1.Pod, + dbname testsUtils.DatabaseName, + query string, + expectedOutput string, +) { + Eventually(func(g Gomega) { + stdout, stderr, err := env.ExecQueryInInstancePod( + testsUtils.PodLocator{ + Namespace: pod.Namespace, + PodName: pod.Name, + }, dbname, query) + if err != nil { + GinkgoWriter.Printf("stdout: %v\nstderr: %v", stdout, stderr) + } + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(strings.Trim(stdout, "\n")).To(BeEquivalentTo(expectedOutput), + fmt.Sprintf("expected query %q to return %q", query, expectedOutput)) + }, 30).Should(Succeed()) } -// AssertUserExists assert if user exists -func AssertUserExists(pod *corev1.Pod, userName string, expectedValue bool) { - By(fmt.Sprintf("verifying if user %v exists", userName), func() { - query := fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE lower(rolname) = lower('%v'));", userName) - Eventually(func(g Gomega) { - stdout, stderr, err := env.ExecQueryInInstancePod( - testsUtils.PodLocator{ - Namespace: pod.Namespace, - PodName: pod.Name, - }, - testsUtils.PostgresDBName, - query) - if err != nil { - GinkgoWriter.Printf("stdout: %v\nstderr: %v", stdout, stderr) - } - g.Expect(err).ToNot(HaveOccurred()) +func roleExistsQuery(roleName string) string { + return fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='%v')", roleName) +} - if expectedValue { - g.Expect(strings.Trim(stdout, "\n")).To(BeEquivalentTo("t")) - } else { - g.Expect(strings.Trim(stdout, "\n")).To(BeEquivalentTo("f")) - } - }, 60).Should(Succeed()) - }) +func databaseExistsQuery(dbName string) string { + return fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname='%v')", dbName) } // AssertDataExpectedCount verifies that an expected amount of rows exists on the table @@ -834,7 +811,7 @@ func AssertScheduledBackupsAreScheduled(namespace string, backupYAMLPath string, Name: scheduledBackupName, } - Eventually(func() (*v1.Time, error) { + Eventually(func() (*metav1.Time, error) { scheduledBackup := &apiv1.ScheduledBackup{} err := env.Client.Get(env.Ctx, scheduledBackupNamespacedName, scheduledBackup) @@ -893,11 +870,6 @@ func getScheduledBackupCompleteBackupsCount(namespace string, scheduledBackupNam // AssertPgRecoveryMode verifies if the target pod recovery mode is enabled or disabled func AssertPgRecoveryMode(pod *corev1.Pod, expectedValue bool) { By(fmt.Sprintf("verifying that postgres recovery mode is %v", expectedValue), func() { - stringExpectedValue := "f" - if expectedValue { - stringExpectedValue = "t" - } - Eventually(func() (string, error) { stdOut, stdErr, err := env.ExecQueryInInstancePod( testsUtils.PodLocator{ @@ -910,10 +882,18 @@ func AssertPgRecoveryMode(pod *corev1.Pod, expectedValue bool) { GinkgoWriter.Printf("stdout: %v\ntderr: %v\n", stdOut, stdErr) } return strings.Trim(stdOut, "\n"), err - }, 300, 10).Should(BeEquivalentTo(stringExpectedValue)) + }, 300, 10).Should(BeEquivalentTo(boolPGOutput(expectedValue))) }) } +func boolPGOutput(expectedValue bool) string { + stringExpectedValue := "f" + if expectedValue { + stringExpectedValue = "t" + } + return stringExpectedValue +} + // AssertReplicaModeCluster checks that, after inserting some data in a source cluster, // a replica cluster can be bootstrapped using pg_basebackup and is properly replicating // from the source cluster @@ -993,8 +973,10 @@ func AssertReplicaModeCluster( // verify the replica database created followed the source database, rather than // default to the "app" db and user By("checking that in replica cluster there is no database app and user app", func() { - AssertDatabaseExists(primaryReplicaCluster, "app", false) - AssertUserExists(primaryReplicaCluster, "app", false) + AssertQueryEventuallyMatchExpectation(primaryReplicaCluster, testsUtils.PostgresDBName, + databaseExistsQuery("app"), "f") + AssertQueryEventuallyMatchExpectation(primaryReplicaCluster, testsUtils.PostgresDBName, + roleExistsQuery("app"), "f") }) } } @@ -1074,8 +1056,10 @@ func AssertDetachReplicaModeCluster( By("verifying the replica database doesn't exist in the replica cluster", func() { // Application database configuration is skipped for replica clusters, // so we expect these to not be present - AssertDatabaseExists(primaryReplicaCluster, replicaDatabaseName, false) - AssertUserExists(primaryReplicaCluster, replicaUserName, false) + AssertQueryEventuallyMatchExpectation(primaryReplicaCluster, testsUtils.PostgresDBName, + databaseExistsQuery(replicaDatabaseName), "f") + AssertQueryEventuallyMatchExpectation(primaryReplicaCluster, testsUtils.PostgresDBName, + roleExistsQuery(replicaUserName), "f") }) By("writing some new data to the source cluster", func() { @@ -1686,7 +1670,7 @@ func AssertScheduledBackupsImmediate(namespace, backupYAMLPath, scheduledBackupN Namespace: namespace, Name: scheduledBackupName, } - Eventually(func() (*v1.Time, error) { + Eventually(func() (*metav1.Time, error) { scheduledBackup := &apiv1.ScheduledBackup{} err = env.Client.Get(env.Ctx, scheduledBackupNamespacedName, scheduledBackup) @@ -2607,7 +2591,7 @@ func AssertBackupConditionTimestampChangedInClusterStatus( namespace, clusterName string, clusterConditionType apiv1.ClusterConditionType, - lastTransactionTimeStamp *v1.Time, + lastTransactionTimeStamp *metav1.Time, ) { By(fmt.Sprintf("waiting for backup condition status in cluster '%v'", clusterName), func() { Eventually(func() (bool, error) { diff --git a/tests/e2e/cluster_microservice_test.go b/tests/e2e/cluster_microservice_test.go index 6019086aa3..bb0240cd6c 100644 --- a/tests/e2e/cluster_microservice_test.go +++ b/tests/e2e/cluster_microservice_test.go @@ -261,7 +261,8 @@ func assertTableAndDataOnImportedCluster( }) By("verifying the user named 'micro' on source is not in imported database", func() { - AssertUserExists(pod, "micro", false) + AssertQueryEventuallyMatchExpectation(pod, testsUtils.PostgresDBName, + roleExistsQuery("micro"), "f") }) }) } @@ -330,8 +331,10 @@ func assertImportRenamesSelectedDatabase( importedPrimaryPod, err := env.GetClusterPrimary(namespace, importedClusterName) Expect(err).ToNot(HaveOccurred()) - AssertUserExists(importedPrimaryPod, "db2", false) - AssertUserExists(importedPrimaryPod, "app", true) + AssertQueryEventuallyMatchExpectation(importedPrimaryPod, testsUtils.PostgresDBName, + roleExistsQuery("db2"), "f") + AssertQueryEventuallyMatchExpectation(importedPrimaryPod, testsUtils.PostgresDBName, + roleExistsQuery("app"), "t") }) By("cleaning up the clusters", func() { diff --git a/tests/e2e/declarative_database_management_test.go b/tests/e2e/declarative_database_management_test.go index 1eafd7bab5..02adf88cd8 100644 --- a/tests/e2e/declarative_database_management_test.go +++ b/tests/e2e/declarative_database_management_test.go @@ -24,7 +24,7 @@ import ( apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" "github.com/cloudnative-pg/cloudnative-pg/tests" - "github.com/cloudnative-pg/cloudnative-pg/tests/utils" + testsUtils "github.com/cloudnative-pg/cloudnative-pg/tests/utils" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -75,7 +75,7 @@ var _ = Describe("Declarative database management", Label(tests.LabelSmoke, test db.Spec.Name, db.Spec.Encoding, db.Spec.LcCtype, db.Spec.LcCollate) Eventually(func(g Gomega) { stdout, _, err := env.ExecQueryInInstancePod( - utils.PodLocator{ + testsUtils.PodLocator{ Namespace: namespace, PodName: primaryPod, }, @@ -119,20 +119,22 @@ var _ = Describe("Declarative database management", Label(tests.LabelSmoke, test primaryPodInfo, err := env.GetClusterPrimary(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) - AssertDatabaseExists(primaryPodInfo, dbname, true) + AssertQueryEventuallyMatchExpectation(primaryPodInfo, testsUtils.PostgresDBName, + databaseExistsQuery(dbname), "t") assertDatabaseHasExpectedFields(namespace, primaryPodInfo.Name, database) }) By("removing the Database object", func() { - Expect(utils.DeleteObject(env, &database)).To(Succeed()) + Expect(testsUtils.DeleteObject(env, &database)).To(Succeed()) }) By("verifying the retention policy in the postgres database", func() { primaryPodInfo, err := env.GetClusterPrimary(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) - AssertDatabaseExists(primaryPodInfo, dbname, retainOnDeletion) + AssertQueryEventuallyMatchExpectation(primaryPodInfo, testsUtils.PostgresDBName, + databaseExistsQuery(dbname), boolPGOutput(retainOnDeletion)) }) } diff --git a/tests/e2e/managed_roles_test.go b/tests/e2e/managed_roles_test.go index fc5dd5f314..a04ea653f2 100644 --- a/tests/e2e/managed_roles_test.go +++ b/tests/e2e/managed_roles_test.go @@ -24,13 +24,13 @@ import ( "github.com/lib/pq" corev1 "k8s.io/api/core/v1" - v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" "github.com/cloudnative-pg/cloudnative-pg/tests" - "github.com/cloudnative-pg/cloudnative-pg/tests/utils" + testsUtils "github.com/cloudnative-pg/cloudnative-pg/tests/utils" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -92,11 +92,11 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic ) mem ON member = oid WHERE rolname =` + pq.QuoteLiteral(roleName) stdout, _, err := env.ExecQueryInInstancePod( - utils.PodLocator{ + testsUtils.PodLocator{ Namespace: namespace, PodName: primaryPod, }, - utils.PostgresDBName, + testsUtils.PostgresDBName, query) if err != nil { return []string{ERROR} @@ -112,11 +112,11 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic Expect(err).ToNot(HaveOccurred()) Eventually(func() string { stdout, _, err := env.ExecQueryInInstancePod( - utils.PodLocator{ + testsUtils.PodLocator{ Namespace: namespace, PodName: primaryPod.Name, }, - utils.PostgresDBName, + testsUtils.PostgresDBName, query) if err != nil { return "" @@ -139,10 +139,14 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic primaryPod, err := env.GetClusterPrimary(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) - AssertUserExists(primaryPod, username, true) - AssertUserExists(primaryPod, userWithPerpetualPass, true) - AssertUserExists(primaryPod, userWithHashedPassword, true) - AssertUserExists(primaryPod, unrealizableUser, false) + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(username), "t") + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(userWithPerpetualPass), "t") + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(userWithHashedPassword), "t") + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(unrealizableUser), "f") query := fmt.Sprintf("SELECT true FROM pg_roles WHERE rolname='%s' and rolcanlogin=%v and rolsuper=%v "+ "and rolcreatedb=%v and rolcreaterole=%v and rolinherit=%v and rolreplication=%v "+ @@ -152,11 +156,11 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic for _, q := range []string{query, query2} { stdout, _, err := env.ExecQueryInInstancePod( - utils.PodLocator{ + testsUtils.PodLocator{ Namespace: primaryPod.Namespace, PodName: primaryPod.Name, }, - utils.PostgresDBName, + testsUtils.PostgresDBName, q) Expect(err).ToNot(HaveOccurred()) Expect(stdout).To(Equal("t\n")) @@ -164,17 +168,20 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic }) By("Verifying connectivity of new managed role", func() { - rwService := utils.GetReadWriteServiceName(clusterName) + rwService := testsUtils.GetReadWriteServiceName(clusterName) // assert connectable use username and password defined in secrets - AssertConnection(namespace, rwService, utils.PostgresDBName, username, password, env) - AssertConnection(namespace, rwService, utils.PostgresDBName, userWithHashedPassword, userWithHashedPassword, env) + AssertConnection(namespace, rwService, testsUtils.PostgresDBName, + username, password, env) + AssertConnection(namespace, rwService, testsUtils.PostgresDBName, + userWithHashedPassword, userWithHashedPassword, env) }) By("ensuring the app role has been granted createdb in the managed stanza", func() { primaryPodInfo, err := env.GetClusterPrimary(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) - AssertUserExists(primaryPodInfo, appUsername, true) + AssertQueryEventuallyMatchExpectation(primaryPodInfo, testsUtils.PostgresDBName, + roleExistsQuery(appUsername), "t") query := fmt.Sprintf("SELECT rolcreatedb and rolvaliduntil='infinity' "+ "FROM pg_roles WHERE rolname='%s'", appUsername) @@ -186,7 +193,7 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic Expect(err).NotTo(HaveOccurred()) appUserSecret := corev1.Secret{} - err = utils.GetObject( + err = testsUtils.GetObject( env, types.NamespacedName{Name: cluster.GetApplicationSecretName(), Namespace: namespace}, &appUserSecret, @@ -194,9 +201,9 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic Expect(err).NotTo(HaveOccurred()) pass := string(appUserSecret.Data["password"]) - rwService := utils.GetReadWriteServiceName(clusterName) + rwService := testsUtils.GetReadWriteServiceName(clusterName) // assert connectable use username and password defined in secrets - AssertConnection(namespace, rwService, utils.PostgresDBName, appUsername, pass, env) + AssertConnection(namespace, rwService, testsUtils.PostgresDBName, appUsername, pass, env) }) By("Verify show unrealizable role configurations in the status", func() { @@ -220,7 +227,7 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic expectedCreateDB := false expectedCreateRole := true expectedConnLmt := int64(10) - rwService := utils.GetReadWriteServiceName(clusterName) + rwService := testsUtils.GetReadWriteServiceName(clusterName) By("updating role attribute in spec", func() { cluster, err := env.GetCluster(namespace, clusterName) @@ -243,8 +250,8 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic }) By("the connection should fail since we disabled the login", func() { - forwardConn, conn, err := utils.ForwardPSQLServiceConnection(env, namespace, rwService, - utils.PostgresDBName, username, password) + forwardConn, conn, err := testsUtils.ForwardPSQLServiceConnection(env, namespace, rwService, + testsUtils.PostgresDBName, username, password) defer func() { _ = conn.Close() forwardConn.Close() @@ -274,9 +281,9 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic }) By("the connectivity should be success again", func() { - rwService := utils.GetReadWriteServiceName(clusterName) + rwService := testsUtils.GetReadWriteServiceName(clusterName) // assert connectable use username and password defined in secrets - AssertConnection(namespace, rwService, utils.PostgresDBName, username, password, env) + AssertConnection(namespace, rwService, testsUtils.PostgresDBName, username, password, env) }) }) @@ -370,7 +377,8 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic Expect(err).ToNot(HaveOccurred()) return len(cluster.Status.ManagedRolesStatus.CannotReconcile) }, 30).Should(Equal(0)) - AssertUserExists(primaryPod, unrealizableUser, true) + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(unrealizableUser), "t") }) By("Add role in InRole for role new_role and verify in database", func() { @@ -431,7 +439,8 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic err = env.Client.Patch(env.Ctx, updated, client.MergeFrom(cluster)) Expect(err).ToNot(HaveOccurred()) // user not changed - AssertUserExists(primaryPod, unrealizableUser, true) + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(unrealizableUser), "t") Eventually(func() int { cluster, err := env.GetCluster(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) @@ -464,9 +473,9 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic }) By("Verify connectivity using changed password in secret", func() { - rwService := utils.GetReadWriteServiceName(clusterName) + rwService := testsUtils.GetReadWriteServiceName(clusterName) // assert connectable use username and password defined in secrets - AssertConnection(namespace, rwService, utils.PostgresDBName, username, newPassword, env) + AssertConnection(namespace, rwService, testsUtils.PostgresDBName, username, newPassword, env) }) By("Update password in database", func() { @@ -474,18 +483,18 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic username, pq.QuoteLiteral(newPassword)) _, _, err = env.ExecQueryInInstancePod( - utils.PodLocator{ + testsUtils.PodLocator{ Namespace: namespace, PodName: primaryPod.Name, }, - utils.PostgresDBName, + testsUtils.PostgresDBName, query) Expect(err).ToNot(HaveOccurred()) }) By("Verify password in secrets is still valid", func() { - rwService := utils.GetReadWriteServiceName(clusterName) - AssertConnection(namespace, rwService, utils.PostgresDBName, username, newPassword, env) + rwService := testsUtils.GetReadWriteServiceName(clusterName) + AssertConnection(namespace, rwService, testsUtils.PostgresDBName, username, newPassword, env) }) }) @@ -497,12 +506,12 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic updated := cluster.DeepCopy() for i, r := range updated.Spec.Managed.Roles { if r.Name == newUserName { - updated.Spec.Managed.Roles[i].ValidUntil = &v1.Time{} + updated.Spec.Managed.Roles[i].ValidUntil = &metav1.Time{} } if r.Name == username { tt, err := time.Parse(time.RFC3339Nano, newValidUntilString) Expect(err).ToNot(HaveOccurred()) - nt := v1.NewTime(tt) + nt := metav1.NewTime(tt) updated.Spec.Managed.Roles[i].ValidUntil = &nt } } @@ -544,7 +553,8 @@ var _ = Describe("Managed roles tests", Label(tests.LabelSmoke, tests.LabelBasic By("Verify new_role not existed in db", func() { primaryPod, err := env.GetClusterPrimary(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) - AssertUserExists(primaryPod, newUserName, false) + AssertQueryEventuallyMatchExpectation(primaryPod, testsUtils.PostgresDBName, + roleExistsQuery(newUserName), "f") }) }) }) diff --git a/tests/e2e/publication_subscription_test.go b/tests/e2e/publication_subscription_test.go index a8af3da01d..6bc14cd150 100644 --- a/tests/e2e/publication_subscription_test.go +++ b/tests/e2e/publication_subscription_test.go @@ -24,23 +24,12 @@ import ( apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" "github.com/cloudnative-pg/cloudnative-pg/tests" - testUtils "github.com/cloudnative-pg/cloudnative-pg/tests/utils" + testsUtils "github.com/cloudnative-pg/cloudnative-pg/tests/utils" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) -// This struct contains info of publication or subscription object, -// in order to maintain just an assert function -type objectInfo struct { - namespace string - name string - clusterName string - dbName string - query string - expected bool -} - // - spinning up a cluster, apply a declarative publication/subscription on it // Set of tests in which we use the declarative publication and subscription CRDs on an existing cluster @@ -103,7 +92,7 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub destPrimaryPod, err := env.GetClusterPrimary(namespace, destinationClusterName) Expect(err).ToNot(HaveOccurred()) _, _, err = env.EventuallyExecQueryInInstancePod( - testUtils.PodLocator{ + testsUtils.PodLocator{ Namespace: destPrimaryPod.Namespace, PodName: destPrimaryPod.Name, }, @@ -117,7 +106,7 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub sourcePrimaryPod, err := env.GetClusterPrimary(namespace, sourceClusterName) Expect(err).ToNot(HaveOccurred()) _, _, err = env.EventuallyExecQueryInInstancePod( - testUtils.PodLocator{ + testsUtils.PodLocator{ Namespace: sourcePrimaryPod.Namespace, PodName: sourcePrimaryPod.Name, }, @@ -130,32 +119,12 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub Expect(DeleteResourcesFromFile(namespace, destinationDatabaseManifest)).To(Succeed()) Expect(DeleteResourcesFromFile(namespace, sourceDatabaseManifest)).To(Succeed()) - AssertDatabaseExists(destPrimaryPod, dbname, false) - AssertDatabaseExists(sourcePrimaryPod, dbname, false) + AssertQueryEventuallyMatchExpectation(sourcePrimaryPod, testsUtils.PostgresDBName, + databaseExistsQuery(dbname), "f") + AssertQueryEventuallyMatchExpectation(destPrimaryPod, testsUtils.PostgresDBName, + databaseExistsQuery(dbname), "f") }) - assertObjectExists := func(obj objectInfo) { - primaryPodInfo, err := env.GetClusterPrimary(namespace, obj.clusterName) - Expect(err).ToNot(HaveOccurred()) - - Eventually(func(g Gomega) { - stdout, _, err := env.ExecQueryInInstancePod( - testUtils.PodLocator{ - Namespace: primaryPodInfo.Namespace, - PodName: primaryPodInfo.Name, - }, - testUtils.DatabaseName(obj.dbName), obj.query) - g.Expect(err).ToNot(HaveOccurred()) - if obj.expected { - g.Expect(stdout).To(ContainSubstring("1"), - fmt.Sprintf("expected %q to be present", obj.name)) - } else { - g.Expect(stdout).To(ContainSubstring("0"), - fmt.Sprintf("expected %q to be not present", obj.name)) - } - }, 30).Should(Succeed()) - } - assertCreateDatabase := func(namespace, clusterName, databaseManifest string) { databaseObject := &apiv1.Database{} databaseObjectName, err := env.GetResourceNameFromYAML(databaseManifest) @@ -182,13 +151,13 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub primaryPodInfo, err := env.GetClusterPrimary(namespace, clusterName) Expect(err).ToNot(HaveOccurred()) - AssertDatabaseExists(primaryPodInfo, databaseObject.Spec.Name, true) + AssertQueryEventuallyMatchExpectation(primaryPodInfo, testsUtils.PostgresDBName, + databaseExistsQuery(databaseObject.Spec.Name), "t") }) } // nolint:dupl assertCreatePublication := func(namespace, clusterName, publicationManifest string) { - pub := &apiv1.Publication{} pubObjectName, err := env.GetResourceNameFromYAML(publicationManifest) Expect(err).NotTo(HaveOccurred()) @@ -204,6 +173,7 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub } Eventually(func(g Gomega) { + pub := &apiv1.Publication{} err := env.Client.Get(env.Ctx, pubNamespacedName, pub) Expect(err).ToNot(HaveOccurred()) g.Expect(pub.Status.Applied).Should(HaveValue(BeTrue())) @@ -211,23 +181,16 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub }) By("verifying new publication has been created", func() { - publicationName := pub.Spec.Name - query := fmt.Sprintf("select count(*) from pg_publication where pubname = '%s'", publicationName) - pub := objectInfo{ - namespace: namespace, - name: publicationName, - clusterName: clusterName, - dbName: pub.Spec.DBName, - query: query, - expected: true, - } - assertObjectExists(pub) + primaryPodInfo, err := env.GetClusterPrimary(namespace, clusterName) + Expect(err).ToNot(HaveOccurred()) + + AssertQueryEventuallyMatchExpectation(primaryPodInfo, dbname, + publicationExistsQuery(pubName), "t") }) } // nolint:dupl assertCreateSubscription := func(namespace, clusterName, subscriptionManifest string) { - sub := &apiv1.Subscription{} subObjectName, err := env.GetResourceNameFromYAML(subscriptionManifest) Expect(err).NotTo(HaveOccurred()) @@ -243,6 +206,7 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub } Eventually(func(g Gomega) { + sub := &apiv1.Subscription{} err := env.Client.Get(env.Ctx, pubNamespacedName, sub) Expect(err).ToNot(HaveOccurred()) g.Expect(sub.Status.Applied).Should(HaveValue(BeTrue())) @@ -250,17 +214,11 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub }) By("verifying new subscription has been created", func() { - subscriptionName := sub.Spec.Name - query := fmt.Sprintf("select count(*) from pg_subscription where subname = '%s'", subscriptionName) - sub := objectInfo{ - namespace: namespace, - name: subscriptionName, - clusterName: clusterName, - dbName: sub.Spec.DBName, - query: query, - expected: true, - } - assertObjectExists(sub) + primaryPodInfo, err := env.GetClusterPrimary(namespace, clusterName) + Expect(err).ToNot(HaveOccurred()) + + AssertQueryEventuallyMatchExpectation(primaryPodInfo, dbname, + subscriptionExistsQuery(subName), "t") }) } @@ -279,7 +237,7 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub By("creating an empty table inside the destination database", func() { query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %v (column1 int) ;", tableName) - _, err = testUtils.RunExecOverForward(env, namespace, destinationClusterName, dbname, + _, err = testsUtils.RunExecOverForward(env, namespace, destinationClusterName, dbname, apiv1.ApplicationUserSecretSuffix, query) Expect(err).ToNot(HaveOccurred()) }) @@ -297,7 +255,6 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub if retainOnDeletion { publicationReclaimPolicy = apiv1.PublicationReclaimRetain subscriptionReclaimPolicy = apiv1.SubscriptionReclaimRetain - } // Get the object names pubObjectName, err := env.GetResourceNameFromYAML(pubManifest) @@ -306,13 +263,13 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub Expect(err).NotTo(HaveOccurred()) Eventually(func(g Gomega) { - publication, err = testUtils.GetPublicationObject(namespace, pubObjectName, env) + publication, err = testsUtils.GetPublicationObject(namespace, pubObjectName, env) g.Expect(err).ToNot(HaveOccurred()) publication.Spec.ReclaimPolicy = publicationReclaimPolicy err = env.Client.Update(env.Ctx, publication) g.Expect(err).ToNot(HaveOccurred()) - subscription, err = testUtils.GetSubscriptionObject(namespace, subObjectName, env) + subscription, err = testsUtils.GetSubscriptionObject(namespace, subObjectName, env) g.Expect(err).ToNot(HaveOccurred()) subscription.Spec.ReclaimPolicy = subscriptionReclaimPolicy err = env.Client.Update(env.Ctx, subscription) @@ -331,36 +288,24 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub }) By("removing the objects", func() { - Expect(testUtils.DeleteObject(env, publication)).To(Succeed()) - Expect(testUtils.DeleteObject(env, subscription)).To(Succeed()) + Expect(testsUtils.DeleteObject(env, publication)).To(Succeed()) + Expect(testsUtils.DeleteObject(env, subscription)).To(Succeed()) }) - By("verifying the retention policy in the postgres database", func() { - publicationName := publication.Spec.Name - pubQuery := fmt.Sprintf("select count(*) from pg_publication where pubname = '%s'", - publicationName) - pub := objectInfo{ - namespace: namespace, - name: publication.Name, - clusterName: sourceClusterName, - dbName: publication.Spec.DBName, - query: pubQuery, - expected: retainOnDeletion, - } - assertObjectExists(pub) - - subscriptionName := subscription.Spec.Name - subQuery := fmt.Sprintf("select count(*) from pg_subscription where subname = '%s'", - subscriptionName) - sub := objectInfo{ - namespace: namespace, - name: subscription.Name, - clusterName: destinationClusterName, - dbName: subscription.Spec.DBName, - query: subQuery, - expected: retainOnDeletion, - } - assertObjectExists(sub) + By("verifying the publication reclaim policy outcome", func() { + primaryPodInfo, err := env.GetClusterPrimary(namespace, sourceClusterName) + Expect(err).ToNot(HaveOccurred()) + + AssertQueryEventuallyMatchExpectation(primaryPodInfo, dbname, + publicationExistsQuery(pubName), boolPGOutput(retainOnDeletion)) + }) + + By("verifying the subscription reclaim policy outcome", func() { + primaryPodInfo, err := env.GetClusterPrimary(namespace, destinationClusterName) + Expect(err).ToNot(HaveOccurred()) + + AssertQueryEventuallyMatchExpectation(primaryPodInfo, dbname, + subscriptionExistsQuery(subName), boolPGOutput(retainOnDeletion)) }) } @@ -377,3 +322,11 @@ var _ = Describe("Publication and Subscription", Label(tests.LabelDeclarativePub }) }) }) + +func publicationExistsQuery(pubName string) string { + return fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_publication WHERE pubname='%s')", pubName) +} + +func subscriptionExistsQuery(subName string) string { + return fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_subscription WHERE subname='%s')", subName) +}