[MISC] Define charm constants (#774)

Author: sinclert-canonical

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

@@ -71,6 +71,7 @@
 from constants import (
     APP_SCOPE,
     BACKUP_USER,
+    DATABASE_DEFAULT_NAME,
     METRICS_PORT,
     MONITORING_PASSWORD_KEY,
     MONITORING_SNAP_SERVICE,
@@ -373,7 +374,7 @@ def postgresql(self) -> PostgreSQL:
             current_host=self._unit_ip,
             user=USER,
             password=self.get_secret(APP_SCOPE, f"{USER}-password"),
-            database="postgres",
+            database=DATABASE_DEFAULT_NAME,
             system_users=SYSTEM_USERS,
         )
 

src/constants.py

@@ -6,6 +6,7 @@
 BACKUP_ID_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
 PGBACKREST_BACKUP_ID_FORMAT = "%Y%m%d-%H%M%S"
 DATABASE = "database"
+DATABASE_DEFAULT_NAME = "postgres"
 DATABASE_PORT = "5432"
 LEGACY_DB = "db"
 LEGACY_DB_ADMIN = "db-admin"

tests/integration/helpers.py

@@ -30,6 +30,8 @@
     wait_fixed,
 )
 
+from constants import DATABASE_DEFAULT_NAME
+
 CHARM_BASE = "ubuntu@22.04"
 METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
 DATABASE_APP_NAME = METADATA["name"]
@@ -497,7 +499,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,
     assert_sync_standbys,
@@ -277,7 +279,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)
@@ -487,7 +492,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",
     ]:
@@ -511,11 +516,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, (
@@ -533,7 +538,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
@@ -542,8 +547,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
 from .helpers import build_connection_string
 from .test_new_relations_1 import DATA_INTEGRATOR_APP_NAME
@@ -125,14 +127,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) as connection,
@@ -142,7 +144,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: