test superuser-only flags and password modes with a non-superuser admin

The flag test now covers `superuser`, `replication`, and `bypassrls`. Two tests run `passwordEncryption: server` and a pre-hashed SCRAM-SHA-256 verifier under the non-superuser admin.

Author: ThoSap

operator/src/test/java/it/aboutbits/postgresql/crd/role/RoleReconcilerNonSuperuserTest.java

@@ -15,9 +15,14 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.stream.Stream;
 
 import static it.aboutbits.postgresql.core.KubernetesService.SECRET_DATA_BASIC_AUTH_PASSWORD_KEY;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -206,18 +211,22 @@ void nonSuperuserAdmin_updatesLoginAndMembership() {
         adminDsl.execute("drop role if exists {0}", role(parentRole));
     }
 
-    @Test
-    @DisplayName("When the admin is not a superuser and the Role asks for SUPERUSER, the status should be ERROR")
-    void nonSuperuserAdmin_superuserFlag_setsError() {
+    @ParameterizedTest(name = "flag {0}")
+    @MethodSource("provideSuperuserOnlyFlags")
+    @DisplayName("When the admin is not a superuser and the Role asks for a superuser-only flag, the status should be ERROR")
+    void nonSuperuserAdmin_superuserOnlyFlag_setsError(
+            String flagName,
+            BiConsumer<RoleSpec.Flags, Boolean> flagSetter
+    ) {
         // given
-        var rootConnection = givenRootClusterConnection("test-connection-root-superuser-flag");
+        var rootConnection = givenRootClusterConnection("test-connection-root-%s-flag".formatted(flagName));
         var rootDsl = postgreSQLContextFactory.getDSLContext(rootConnection);
-        var adminConnection = givenNonSuperuserClusterConnection(rootDsl, "test-connection-non-superuser-flag");
+        var adminConnection = givenNonSuperuserClusterConnection(rootDsl, "test-connection-non-superuser-%s-flag".formatted(flagName));
 
-        var roleName = "test-non-superuser-role-superuser-flag";
+        var roleName = "test-non-superuser-role-%s-flag".formatted(flagName);
 
         var flags = new RoleSpec.Flags();
-        flags.setSuperuser(true);
+        flagSetter.accept(flags, true);
 
         // when
         var role = given.one()
@@ -242,6 +251,84 @@ void nonSuperuserAdmin_superuserFlag_setsError() {
         assertThat(roleService.roleExists(rootDsl, role.getSpec())).isFalse();
     }
 
+    @Test
+    @DisplayName("When the admin is not a superuser and passwordEncryption is 'server', the server should hash the password")
+    void nonSuperuserAdmin_serverPasswordEncryption_letsTheServerHash() {
+        // given
+        var rootConnection = givenRootClusterConnection("test-connection-root-server-encryption");
+        var rootDsl = postgreSQLContextFactory.getDSLContext(rootConnection);
+        var adminConnection = givenNonSuperuserClusterConnection(rootDsl, "test-connection-non-superuser-server-encryption");
+
+        var roleName = "test-non-superuser-role-server-encryption";
+        var password = "server-side-password";
+
+        var secretRef = given.one()
+                .secretRef()
+                .withPassword(password)
+                .returnFirst();
+
+        // when
+        var role = given.one()
+                .role()
+                .withName(roleName)
+                .withClusterConnectionName(adminConnection.getMetadata().getName())
+                .withPasswordSecretRef(secretRef)
+                .withPasswordEncryption(PasswordEncryption.SERVER)
+                .returnFirst();
+
+        // then
+        assertThat(role.getStatus().getPhase()).isEqualTo(CRPhase.READY);
+        assertThat(PostgreSQLPasswordVerifier.passwordMatches(
+                rootDsl,
+                roleName,
+                password
+        )).isTrue();
+    }
+
+    @Test
+    @DisplayName("When the admin is not a superuser and the Secret contains a SCRAM-SHA-256 verifier, it should be stored verbatim")
+    void nonSuperuserAdmin_preHashedPassword_isStoredVerbatim() {
+        // given
+        var rootConnection = givenRootClusterConnection("test-connection-root-prehashed");
+        var rootDsl = postgreSQLContextFactory.getDSLContext(rootConnection);
+        var adminConnection = givenNonSuperuserClusterConnection(rootDsl, "test-connection-non-superuser-prehashed");
+
+        var roleName = "test-non-superuser-role-prehashed";
+
+        // Verifier for the password "abc", generated by PostgreSQL
+        var verifier = "SCRAM-SHA-256$4096:gxUQWxfrRYegSTNiHXFT+g==$lxMC2yO9Lx9gm2dgNPo/1Qar+pjAvxCP2VN4yPWYnzE=:0QzcS9VJHJszBq4vSce3n4M6NZmyWa1GWdkJDi8hRNc=";
+
+        var secretRef = given.one()
+                .secretRef()
+                .withPassword(verifier)
+                .returnFirst();
+
+        // when
+        var role = given.one()
+                .role()
+                .withName(roleName)
+                .withClusterConnectionName(adminConnection.getMetadata().getName())
+                .withPasswordSecretRef(secretRef)
+                .returnFirst();
+
+        // then
+        assertThat(role.getStatus().getPhase()).isEqualTo(CRPhase.READY);
+        assertThat(PostgreSQLPasswordVerifier.storedVerifier(rootDsl, roleName)).isEqualTo(verifier);
+        assertThat(PostgreSQLPasswordVerifier.passwordMatches(
+                rootDsl,
+                roleName,
+                "abc"
+        )).isTrue();
+    }
+
+    private static Stream<Arguments> provideSuperuserOnlyFlags() {
+        return Stream.of(
+                Arguments.of("superuser", (BiConsumer<RoleSpec.Flags, Boolean>) RoleSpec.Flags::setSuperuser),
+                Arguments.of("replication", (BiConsumer<RoleSpec.Flags, Boolean>) RoleSpec.Flags::setReplication),
+                Arguments.of("bypassrls", (BiConsumer<RoleSpec.Flags, Boolean>) RoleSpec.Flags::setBypassrls)
+        );
+    }
+
     private ClusterConnection givenRootClusterConnection(String name) {
         return given.one()
                 .clusterConnection()