[DPE-7521] Fix HBA rules for Landscape related through PgBouncer (#946)

* Fix HBA rules for Landscape related through PgBouncer

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* Update comment

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* Order users and databases

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* Add unit test for relations_user_databases_map property

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* Fix typo

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

---------

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

Author: marceloneppel

src/charm.py

@@ -2157,16 +2157,29 @@ def relations_user_databases_map(self) -> dict:
         if not self.is_cluster_initialised or not self._patroni.member_started:
             return {USER: "all", REPLICATION_USER: "all", REWIND_USER: "all"}
         user_database_map = {}
-        for user in self.postgresql.list_users_from_relation(
-            current_host=self.is_connectivity_enabled
+        for user in sorted(
+            self.postgresql.list_users_from_relation(current_host=self.is_connectivity_enabled)
         ):
             user_database_map[user] = ",".join(
-                self.postgresql.list_accessible_databases_for_user(
-                    user, current_host=self.is_connectivity_enabled
+                sorted(
+                    self.postgresql.list_accessible_databases_for_user(
+                        user, current_host=self.is_connectivity_enabled
+                    )
                 )
             )
-            # Add "landscape" superuser by default to the list when the "db-admin" relation is present.
-            if any(True for relation in self.client_relations if relation.name == "db-admin"):
+            # Add "landscape" superuser by default to the list when the "db-admin" relation is present
+            # or when the "database" relation has "extra-user-roles" set to "SUPERUSER" (which may mean
+            # that PgBouncer is related to the database and there is the possibility that Landscape
+            # is related to it).
+            if any(
+                True
+                for relation in self.client_relations
+                if relation.name == "db-admin"  # Possibly Landscape relation.
+                or (
+                    relation.name == "database"
+                    and relation.data[relation.app].get("extra-user-roles") == "SUPERUSER"
+                )  # PgBouncer (which may be related to Landscape).
+            ):
                 user_database_map["landscape"] = "all"
         if self.postgresql.list_access_groups(current_host=self.is_connectivity_enabled) != set(
             ACCESS_GROUPS

tests/unit/test_charm.py

@@ -2916,3 +2916,59 @@ def test_get_ldap_parameters(harness):
         harness.charm.get_ldap_parameters()
         _get_relation_data.assert_called_once()
         _get_relation_data.reset_mock()
+
+
+def test_relations_user_databases_map(harness):
+    with (
+        patch("charm.PostgresqlOperatorCharm.postgresql") as _postgresql,
+        patch("charm.Patroni.member_started", new_callable=PropertyMock) as _member_started,
+        patch(
+            "charm.PostgresqlOperatorCharm.is_cluster_initialised", new_callable=PropertyMock
+        ) as _is_cluster_initialised,
+    ):
+        # Initial empty results from the functions used in the property that's being tested.
+        _postgresql.list_users_from_relation.return_value = set()
+        _postgresql.list_accessible_databases_for_user.return_value = set()
+        _postgresql.list_access_groups.return_value = {
+            "identity_access",
+            "internal_access",
+            "relation_access",
+        }
+
+        # Test when the cluster isn't initialised yet.
+        _is_cluster_initialised.return_value = False
+        _member_started.return_value = True
+        assert harness.charm.relations_user_databases_map == {
+            "operator": "all",
+            "replication": "all",
+            "rewind": "all",
+        }
+
+        # Test when the cluster is initialised but the cluster member hasn't started yet.
+        _is_cluster_initialised.return_value = True
+        _member_started.return_value = False
+        assert harness.charm.relations_user_databases_map == {
+            "operator": "all",
+            "replication": "all",
+            "rewind": "all",
+        }
+
+        # Test when there are no relation users in the database.
+        _member_started.return_value = True
+        assert harness.charm.relations_user_databases_map == {}
+
+        # Test when there are relation users in the database.
+        _postgresql.list_users_from_relation.return_value = {"user1", "user2"}
+        _postgresql.list_accessible_databases_for_user.side_effect = [{"db1", "db2"}, {"db3"}]
+        assert harness.charm.relations_user_databases_map == {"user1": "db1,db2", "user2": "db3"}
+
+        # Test when the access groups where not created yet.
+        _postgresql.list_accessible_databases_for_user.side_effect = [{"db1", "db2"}, {"db3"}]
+        _postgresql.list_access_groups.return_value = set()
+        assert harness.charm.relations_user_databases_map == {
+            "user1": "db1,db2",
+            "user2": "db3",
+            "operator": "all",
+            "replication": "all",
+            "rewind": "all",
+        }