[MISC] Define charm constants

lib/charms/postgresql_k8s/v0/postgresql.py

@@ -35,7 +35,10 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 42
+LIBPATCH = 43
+
+# Groups to distinguish database permissions
+PERMISSIONS_GROUP_ADMIN = "admin"
 
 INVALID_EXTRA_USER_ROLE_BLOCKING_MESSAGE = "invalid role(s) for extra user roles"
 
@@ -187,7 +190,7 @@ def create_database(
                     Identifier(database)
                 )
             )
-            for user_to_grant_access in [user, "admin", *self.system_users]:
+            for user_to_grant_access in [user, PERMISSIONS_GROUP_ADMIN, *self.system_users]:
                 cursor.execute(
                     SQL("GRANT ALL PRIVILEGES ON DATABASE {} TO {};").format(
                         Identifier(database), Identifier(user_to_grant_access)
@@ -236,15 +239,17 @@ def create_user(
             roles = privileges = None
             if extra_user_roles:
                 extra_user_roles = tuple(extra_user_roles.lower().split(","))
-                admin_role = "admin" in extra_user_roles
+                admin_role = PERMISSIONS_GROUP_ADMIN in extra_user_roles
                 valid_privileges, valid_roles = self.list_valid_privileges_and_roles()
                 roles = [
-                    role for role in extra_user_roles if role in valid_roles and role != "admin"
+                    role
+                    for role in extra_user_roles
+                    if role in valid_roles and role != PERMISSIONS_GROUP_ADMIN
                 ]
                 privileges = {
                     extra_user_role
                     for extra_user_role in extra_user_roles
-                    if extra_user_role not in roles and extra_user_role != "admin"
+                    if extra_user_role not in roles and extra_user_role != PERMISSIONS_GROUP_ADMIN
                 }
                 invalid_privileges = [
                     privilege for privilege in privileges if privilege not in valid_privileges
@@ -566,7 +571,7 @@ def set_up_database(self) -> None:
                         )
                     )
                 self.create_user(
-                    "admin",
+                    PERMISSIONS_GROUP_ADMIN,
                     extra_user_roles="pg_read_all_data,pg_write_all_data",
                 )
                 cursor.execute("GRANT CONNECT ON DATABASE postgres TO admin;")

src/charm.py

@@ -86,6 +86,7 @@
 from constants import (
     APP_SCOPE,
     BACKUP_USER,
+    DATABASE_DEFAULT_NAME,
     METRICS_PORT,
     MONITORING_PASSWORD_KEY,
     MONITORING_USER,
@@ -416,7 +417,7 @@ def postgresql(self) -> PostgreSQL:
             current_host=self.endpoint,
             user=USER,
             password=self.get_secret(APP_SCOPE, f"{USER}-password"),
-            database="postgres",
+            database=DATABASE_DEFAULT_NAME,
             system_users=SYSTEM_USERS,
         )
 

src/constants.py

@@ -3,6 +3,7 @@
 
 """File containing constants to be used in the charm."""
 
+DATABASE_DEFAULT_NAME = "postgres"
 DATABASE_PORT = "5432"
 PEER = "database-peers"
 BACKUP_USER = "backup"

tests/integration/helpers.py

@@ -32,6 +32,8 @@
     wait_fixed,
 )
 
+from constants import DATABASE_DEFAULT_NAME
+
 CHARM_BASE = "ubuntu@22.04"
 CHARM_SERIES = "jammy"
 METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
@@ -325,7 +327,7 @@ async def execute_query_on_unit(
     unit_address: str,
     password: str,
     query: str,
-    database: str = "postgres",
+    database: str = DATABASE_DEFAULT_NAME,
     sslmode: str | None = None,
 ):
     """Execute given PostgreSQL query on a unit.

tests/integration/new_relations/test_new_relations_1.py

@@ -12,6 +12,8 @@
 from pytest_operator.plugin import OpsTest
 from tenacity import Retrying, stop_after_attempt, wait_fixed
 
+from constants import DATABASE_DEFAULT_NAME
+
 from ..helpers import (
     CHARM_BASE,
     check_database_users_existence,
@@ -218,7 +220,10 @@ async def test_two_applications_doesnt_share_the_same_relation_data(ops_test: Op
         (another_application_app_name, f"{APPLICATION_APP_NAME.replace('-', '_')}_database"),
     ]:
         connection_string = await build_connection_string(
-            ops_test, application, FIRST_DATABASE_RELATION_NAME, database="postgres"
+            ops_test,
+            application,
+            FIRST_DATABASE_RELATION_NAME,
+            database=DATABASE_DEFAULT_NAME,
         )
         with pytest.raises(psycopg2.Error):
             psycopg2.connect(connection_string)
@@ -448,7 +453,7 @@ async def test_admin_role(ops_test: OpsTest):
 
     # Check that the user can access all the databases.
     for database in [
-        "postgres",
+        DATABASE_DEFAULT_NAME,
         f"{APPLICATION_APP_NAME.replace('-', '_')}_database",
         "another_application_database",
     ]:
@@ -472,11 +477,11 @@ async def test_admin_role(ops_test: OpsTest):
                 )
                 assert version == data
 
-                # Write some data (it should fail in the "postgres" database).
+                # Write some data (it should fail in the default database name).
                 random_name = (
                     f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}"
                 )
-                should_fail = database == "postgres"
+                should_fail = database == DATABASE_DEFAULT_NAME
                 cursor.execute(f"CREATE TABLE {random_name}(data TEXT);")
                 if should_fail:
                     assert False, (
@@ -494,7 +499,7 @@ async def test_admin_role(ops_test: OpsTest):
 
     # Test the creation and deletion of databases.
     connection_string = await build_connection_string(
-        ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", database="postgres"
+        ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", database=DATABASE_DEFAULT_NAME
     )
     connection = psycopg2.connect(connection_string)
     connection.autocommit = True
@@ -503,8 +508,10 @@ async def test_admin_role(ops_test: OpsTest):
     cursor.execute(f"CREATE DATABASE {random_name};")
     cursor.execute(f"DROP DATABASE {random_name};")
     try:
-        cursor.execute("DROP DATABASE postgres;")
-        assert False, "the admin extra user role was able to drop the `postgres` system database"
+        cursor.execute(f"DROP DATABASE {DATABASE_DEFAULT_NAME};")
+        assert False, (
+            f"the admin extra user role was able to drop the `{DATABASE_DEFAULT_NAME}` system database"
+        )
     except psycopg2.errors.InsufficientPrivilege:
         # Ignore the error, as the admin extra user role mustn't be able to drop
         # the "postgres" system database.

tests/integration/new_relations/test_relations_coherence.py

@@ -9,6 +9,8 @@
 import pytest
 from pytest_operator.plugin import OpsTest
 
+from constants import DATABASE_DEFAULT_NAME
+
 from ..helpers import CHARM_BASE, DATABASE_APP_NAME, build_and_deploy
 from .helpers import build_connection_string
 from .test_new_relations_1 import DATA_INTEGRATOR_APP_NAME
@@ -120,14 +122,14 @@ async def test_relations(ops_test: OpsTest, charm):
 
         for database in [
             DATA_INTEGRATOR_APP_NAME.replace("-", "_"),
-            "postgres",
+            DATABASE_DEFAULT_NAME,
         ]:
             logger.info(f"connecting to the following database: {database}")
             connection_string = await build_connection_string(
                 ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", database=database
             )
             connection = None
-            should_fail = database == "postgres"
+            should_fail = database == DATABASE_DEFAULT_NAME
             try:
                 with psycopg2.connect(
                     connection_string
@@ -136,7 +138,7 @@ async def test_relations(ops_test: OpsTest, charm):
                     data = cursor.fetchone()
                     assert data[0] == "some data"
 
-                    # Write some data (it should fail in the "postgres" database).
+                    # Write some data (it should fail in the default database name).
                     random_name = f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}"
                     cursor.execute(f"CREATE TABLE {random_name}(data TEXT);")
                     if should_fail:

tests/unit/test_postgresql.py

@@ -5,14 +5,22 @@
 import psycopg2
 import pytest
 from charms.postgresql_k8s.v0.postgresql import (
+    PERMISSIONS_GROUP_ADMIN,
     PostgreSQLCreateDatabaseError,
     PostgreSQLGetLastArchivedWALError,
 )
 from ops.testing import Harness
 from psycopg2.sql import SQL, Composed, Identifier, Literal
 
 from charm import PostgresqlOperatorCharm
-from constants import PEER
+from constants import (
+    BACKUP_USER,
+    MONITORING_USER,
+    PEER,
+    REPLICATION_USER,
+    REWIND_USER,
+    USER,
+)
 
 
 @pytest.fixture(autouse=True)
@@ -75,7 +83,7 @@ def test_create_database(harness):
                     SQL("GRANT ALL PRIVILEGES ON DATABASE "),
                     Identifier(database),
                     SQL(" TO "),
-                    Identifier("admin"),
+                    Identifier(PERMISSIONS_GROUP_ADMIN),
                     SQL(";"),
                 ])
             ),
@@ -84,7 +92,7 @@ def test_create_database(harness):
                     SQL("GRANT ALL PRIVILEGES ON DATABASE "),
                     Identifier(database),
                     SQL(" TO "),
-                    Identifier("backup"),
+                    Identifier(BACKUP_USER),
                     SQL(";"),
                 ])
             ),
@@ -93,7 +101,7 @@ def test_create_database(harness):
                     SQL("GRANT ALL PRIVILEGES ON DATABASE "),
                     Identifier(database),
                     SQL(" TO "),
-                    Identifier("replication"),
+                    Identifier(REPLICATION_USER),
                     SQL(";"),
                 ])
             ),
@@ -102,7 +110,7 @@ def test_create_database(harness):
                     SQL("GRANT ALL PRIVILEGES ON DATABASE "),
                     Identifier(database),
                     SQL(" TO "),
-                    Identifier("rewind"),
+                    Identifier(REWIND_USER),
                     SQL(";"),
                 ])
             ),
@@ -111,7 +119,7 @@ def test_create_database(harness):
                     SQL("GRANT ALL PRIVILEGES ON DATABASE "),
                     Identifier(database),
                     SQL(" TO "),
-                    Identifier("operator"),
+                    Identifier(USER),
                     SQL(";"),
                 ])
             ),
@@ -120,7 +128,7 @@ def test_create_database(harness):
                     SQL("GRANT ALL PRIVILEGES ON DATABASE "),
                     Identifier(database),
                     SQL(" TO "),
-                    Identifier("monitoring"),
+                    Identifier(MONITORING_USER),
                     SQL(";"),
                 ])
             ),