[DPE-8473] Add 'databases' to the invalid database names list

pyproject.toml

@@ -4,7 +4,7 @@
 [project]
 name = "postgresql-charms-single-kernel"
 description = "Shared and reusable code for PostgreSQL-related charms"
-version = "16.0.2"
+version = "16.1.0"
 readme = "README.md"
 license = "Apache-2.0"
 authors = [

single_kernel_postgresql/abstract_charm.py

@@ -4,7 +4,7 @@
 
 from ops.charm import CharmBase
 
-from .config.literals import SYSTEM_USERS, USER
+from .config.literals import SYSTEM_USERS, USER, Substrates
 from .utils.postgresql import PostgreSQL
 
 
@@ -15,6 +15,7 @@ def __init__(self, *args):
         super().__init__(*args)
 
         self.postgresql = PostgreSQL(
+            substrate=Substrates.VM,
             primary_host="localhost",
             current_host="localhost",
             user=USER,

single_kernel_postgresql/config/literals.py

@@ -5,6 +5,8 @@
 This module should contain the literals used in the charms (paths, enums, etc).
 """
 
+from enum import Enum
+
 # Permissions.
 POSTGRESQL_STORAGE_PERMISSIONS = 0o700
 
@@ -19,3 +21,10 @@
 SNAP_USER = "_daemon_"
 USER = "operator"
 SYSTEM_USERS = [MONITORING_USER, REPLICATION_USER, REWIND_USER, USER]
+
+
+class Substrates(str, Enum):
+    """Possible substrates."""
+
+    K8S = "k8s"
+    VM = "vm"

single_kernel_postgresql/utils/postgresql.py

@@ -30,7 +30,13 @@
 from ops import ConfigData
 from psycopg2.sql import SQL, Identifier, Literal
 
-from ..config.literals import BACKUP_USER, POSTGRESQL_STORAGE_PERMISSIONS, SNAP_USER, SYSTEM_USERS
+from ..config.literals import (
+    BACKUP_USER,
+    POSTGRESQL_STORAGE_PERMISSIONS,
+    SNAP_USER,
+    SYSTEM_USERS,
+    Substrates,
+)
 from .filesystem import change_owner
 
 # Groups to distinguish HBA access
@@ -60,6 +66,7 @@
 }
 
 INVALID_DATABASE_NAME_BLOCKING_MESSAGE = "invalid database name"
+INVALID_DATABASE_NAMES = ["databases", "postgres", "template0", "template1"]
 INVALID_EXTRA_USER_ROLE_BLOCKING_MESSAGE = "invalid role(s) for extra user roles"
 
 REQUIRED_PLUGINS = {
@@ -216,13 +223,26 @@ class PostgreSQL:
 
     def __init__(
         self,
+        substrate: Substrates,
         primary_host: Optional[str],
         current_host: Optional[str],
         user: str,
         password: Optional[str],
         database: str,
         system_users: Optional[List[str]] = None,
     ):
+        """Create a PostgreSQL helper.
+
+        Args:
+            substrate: substrate where the charm is running (Substrates.K8S or Substrates.VM).
+            primary_host: hostname or address for primary database host.
+            current_host: hostname or address for the current database host.
+            user: username to connect as.
+            password: password for the user.
+            database: default database name.
+            system_users: list of system users.
+        """
+        self.substrate = substrate
         self.primary_host = primary_host
         self.current_host = current_host
         self.user = user
@@ -322,7 +342,7 @@ def create_database(
         if len(database) > 49:
             logger.error(f"Invalid database name (it must not exceed 49 characters): {database}.")
             raise PostgreSQLCreateDatabaseError(INVALID_DATABASE_NAME_BLOCKING_MESSAGE)
-        if database in ["postgres", "template0", "template1"]:
+        if database in INVALID_DATABASE_NAMES:
             logger.error(f"Invalid database name: {database}.")
             raise PostgreSQLCreateDatabaseError(INVALID_DATABASE_NAME_BLOCKING_MESSAGE)
         plugins = plugins if plugins else []
@@ -1077,7 +1097,7 @@ def set_up_database(self, temp_location: Optional[str] = None) -> None:
             if temp_location is not None:
                 # Fix permissions on the temporary tablespace location when a reboot happens and tmpfs is being used.
                 temp_location_stats = os.stat(temp_location)
-                if (
+                if self.substrate == Substrates.VM and (
                     pwd.getpwuid(temp_location_stats.st_uid).pw_name != SNAP_USER
                     or int(temp_location_stats.st_mode & 0o777) != POSTGRESQL_STORAGE_PERMISSIONS
                 ):

tests/unit/test_filesystem.py

@@ -4,6 +4,7 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from single_kernel_postgresql.config.literals import SNAP_USER
 from single_kernel_postgresql.utils.filesystem import change_owner
 
 
@@ -21,7 +22,8 @@ def test_change_owner_calls_pwd_and_os_chown_with_daemon_user():
 
         change_owner(tmp.name)
 
-        getpwnam.assert_called_once_with("_daemon_")
+        # Ensure getpwnam was called for SNAP_USER and ended up using snap user
+        getpwnam.assert_called_once_with(SNAP_USER)
         chown.assert_called_once_with(tmp.name, uid=1234, gid=4321)
 
 

tests/unit/test_postgresql.py

@@ -28,6 +28,7 @@
     PostgreSQLUndefinedPasswordError,
     ROLE_DATABASES_OWNER,
 )
+from single_kernel_postgresql.config.literals import Substrates
 
 
 @pytest.fixture(autouse=True)
@@ -341,9 +342,11 @@ def test_set_up_database_with_temp_tablespace_and_missing_owner_role(harness):
         patch("single_kernel_postgresql.utils.postgresql.pwd.getpwuid") as _getpwuid,
     ):
         # Simulate a temp location owned by wrong user/permissions to trigger fixup (33188 means 0o644)
-        stat_result = type("stat_result", (), {"st_uid": 0, "st_mode": 33188})
+        stat_result = type("stat_result", (), {"st_uid": 0, "st_gid": 0, "st_mode": 33188})
         _stat.return_value = stat_result
         _getpwuid.return_value.pw_name = "root"
+        _getpwuid.return_value.pw_uid = 0
+        _getpwuid.return_value.pw_gid = 0
 
         # First connection (non-context) for temp tablespace
         execute_direct = _connect_to_database.return_value.cursor.return_value.execute
@@ -403,9 +406,13 @@ def test_set_up_database_owner_mismatch_triggers_rename_and_fix(harness):
         patch("single_kernel_postgresql.utils.postgresql.datetime") as _dt,
     ):
         # Owner differs, permissions are correct (16832 means 0o700)
-        stat_result = type("stat_result", (), {"st_uid": 0, "st_mode": 16832})
+        # Simulate directory owned by uid 1000 while expected owner has uid 0 to force mismatch
+        stat_result = type("stat_result", (), {"st_uid": 1000, "st_gid": 1000, "st_mode": 16832})
         _stat.return_value = stat_result
+        # The expected owner (SNAP_USER) resolves to uid 0/gid 0 for the test
         _getpwuid.return_value.pw_name = "root"
+        _getpwuid.return_value.pw_uid = 0
+        _getpwuid.return_value.pw_gid = 0
 
         # Mock datetime.now(timezone.utc) to a fixed timestamp
         _dt.now.return_value = datetime(2025, 1, 1, 1, 2, 3, tzinfo=UTC)
@@ -439,9 +446,11 @@ def test_set_up_database_permissions_mismatch_triggers_rename_and_fix(harness):
         patch("single_kernel_postgresql.utils.postgresql.datetime") as _dt,
     ):
         # Owner matches SNAP_USER, permissions differ (33188 means 0o644)
-        stat_result = type("stat_result", (), {"st_uid": 0, "st_mode": 33188})
+        stat_result = type("stat_result", (), {"st_uid": 0, "st_gid": 0, "st_mode": 33188})
         _stat.return_value = stat_result
         _getpwuid.return_value.pw_name = SNAP_USER
+        _getpwuid.return_value.pw_uid = 0
+        _getpwuid.return_value.pw_gid = 0
 
         # Mock datetime.now(timezone.utc) to a fixed timestamp
         _dt.now.return_value = datetime(2025, 1, 1, 1, 2, 3, tzinfo=UTC)
@@ -497,7 +506,13 @@ def test_set_up_database_raises_wrapped_error(harness):
         ) as _connect_to_database,
         patch("single_kernel_postgresql.utils.postgresql.change_owner"),
         patch("single_kernel_postgresql.utils.postgresql.os.chmod"),
+        patch("single_kernel_postgresql.utils.postgresql.pwd.getpwuid") as _getpwuid,
     ):
+        # Provide a dummy passwd entry so has_correct_ownership_and_permissions does not raise
+        _getpwuid.return_value.pw_name = SNAP_USER
+        _getpwuid.return_value.pw_uid = 0
+        _getpwuid.return_value.pw_gid = 0
+
         execute_direct = _connect_to_database.return_value.cursor.return_value.execute
         execute_direct.side_effect = psycopg2.Error
         with pytest.raises(PostgreSQLDatabasesSetupError):
@@ -506,17 +521,17 @@ def test_set_up_database_raises_wrapped_error(harness):
 
 def test_connect_to_database():
     # Error on no host
-    pg = PostgreSQL(None, None, "operator", None, "postgres", None)
+    pg = PostgreSQL(Substrates.VM, None, None, "operator", None, "postgres", None)
     with pytest.raises(PostgreSQLUndefinedHostError):
         pg._connect_to_database()
 
     # Error on no password
-    pg = PostgreSQL("primary", "current", "operator", None, "postgres", None)
+    pg = PostgreSQL(Substrates.VM, "primary", "current", "operator", None, "postgres", None)
     with pytest.raises(PostgreSQLUndefinedPasswordError):
         pg._connect_to_database()
 
     # Returns connection
-    pg = PostgreSQL("primary", "current", "operator", "password", "postgres", None)
+    pg = PostgreSQL(Substrates.VM, "primary", "current", "operator", "password", "postgres", None)
     with patch(
         "single_kernel_postgresql.utils.postgresql.psycopg2.connect",
         return_value=sentinel.connection,
@@ -531,7 +546,9 @@ def test_is_user_in_hba():
     with patch(
         "single_kernel_postgresql.utils.postgresql.PostgreSQL._connect_to_database",
     ) as _connect_to_database:
-        pg = PostgreSQL("primary", "current", "operator", "password", "postgres", None)
+        pg = PostgreSQL(
+            Substrates.VM, "primary", "current", "operator", "password", "postgres", None
+        )
         _cursor = _connect_to_database().__enter__().cursor().__enter__()
 
         # No result
@@ -562,7 +579,9 @@ def test_drop_hba_triggers():
         ) as _connect_to_database,
         patch("single_kernel_postgresql.utils.postgresql.logger") as _logger,
     ):
-        pg = PostgreSQL("primary", "current", "operator", "password", "postgres", None)
+        pg = PostgreSQL(
+            Substrates.VM, "primary", "current", "operator", "password", "postgres", None
+        )
         _cursor = _connect_to_database().__enter__().cursor().__enter__()
         _cursor.fetchall.return_value = (("db1",), ("db2",))
 
@@ -614,7 +633,9 @@ def test_create_user():
             "single_kernel_postgresql.utils.postgresql.PostgreSQL._process_extra_user_roles",
         ) as _process_extra_user_roles,
     ):
-        pg = PostgreSQL("primary", "current", "operator", "password", "postgres", None)
+        pg = PostgreSQL(
+            Substrates.VM, "primary", "current", "operator", "password", "postgres", None
+        )
         _cursor = _connect_to_database().__enter__().cursor().__enter__()
         _process_extra_user_roles.return_value = (["role1", "role2"], ["priv1", "priv2"])
 
@@ -731,9 +752,11 @@ def test_set_up_database_owner_and_permissions_match_no_rename_or_fix(harness):
         patch("single_kernel_postgresql.utils.postgresql.pwd.getpwuid") as _getpwuid,
     ):
         # Owner matches SNAP_USER and permissions are correct (16832 means 0o700)
-        stat_result = type("stat_result", (), {"st_uid": 0, "st_mode": 16832})
+        stat_result = type("stat_result", (), {"st_uid": 0, "st_gid": 0, "st_mode": 16832})
         _stat.return_value = stat_result
         _getpwuid.return_value.pw_name = SNAP_USER
+        _getpwuid.return_value.pw_uid = 0
+        _getpwuid.return_value.pw_gid = 0
 
         execute_direct = _connect_to_database.return_value.cursor.return_value.execute
         fetchone_direct = _connect_to_database.return_value.cursor.return_value.fetchone
@@ -753,3 +776,40 @@ def test_set_up_database_owner_and_permissions_match_no_rename_or_fix(harness):
         for c in execute_direct.call_args_list:
             if c.args:
                 assert "ALTER TABLESPACE temp RENAME TO" not in c.args[0]
+
+
+def test_set_up_database_k8s_skips_change_owner_and_chmod(harness):
+    """When running on the K8S substrate, filesystem ownership/permission fixes should be skipped.
+
+    Even if the on-disk owner/permissions appear incorrect, change_owner and os.chmod must
+    not be called for Substrates.K8S.
+    """
+    with (
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL._connect_to_database"
+        ) as _connect_to_database,
+        patch("single_kernel_postgresql.utils.postgresql.PostgreSQL.set_up_login_hook_function"),
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL.set_up_predefined_catalog_roles_function"
+        ),
+        patch("single_kernel_postgresql.utils.postgresql.change_owner") as _change_owner,
+        patch("single_kernel_postgresql.utils.postgresql.os.chmod") as _chmod,
+        patch("single_kernel_postgresql.utils.postgresql.os.stat") as _stat,
+        patch("single_kernel_postgresql.utils.postgresql.pwd.getpwuid") as _getpwuid,
+    ):
+        # Simulate a temp location owned by wrong user/permissions which would normally
+        # trigger a fixup when running on VM substrate.
+        stat_result = type("stat_result", (), {"st_uid": 0, "st_gid": 0, "st_mode": 33188})
+        _stat.return_value = stat_result
+        _getpwuid.return_value.pw_name = "root"
+        _getpwuid.return_value.pw_uid = 0
+        _getpwuid.return_value.pw_gid = 0
+
+        # Force the charm's PostgreSQL instance to think it's running on K8S.
+        harness.charm.postgresql.substrate = Substrates.K8S
+
+        harness.charm.postgresql.set_up_database(temp_location="/var/lib/postgresql/tmp")
+
+        # On K8S substrate we must not attempt to change ownership or chmod the path.
+        _change_owner.assert_not_called()
+        _chmod.assert_not_called()