podman-quadlet-cookbook/cookbooks/postgresql/tests/test_01_install_upgrade_backup.py

import pytest
import textwrap

# Add the current cookbook's tests directory to the path so we can import helpers.py.
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent))
import helpers  # noqa: E402

# Major version of PostgreSQL to install by default on a fresh VM boot.
PG_MAJOR_DEFAULT = 14

# Extra files to inject into the FCOS image for the tests in this file.
# The config.env is used to configure the PostgreSQL Quadlet, and the init.d/test.sql file is an init hook that creates a test database and user on the first boot.
PYTEST_FCOS_EXTRA_FILES = {
    "/etc/quadlets/postgresql/config.env": (
        textwrap.dedent(f"""
            # This file is generated by conftest.py for testing purposes.
            POSTGRES_USER=postgres
            POSTGRES_PASSWORD=postgres
            POSTGRES_DB=postgres
            POSTGRES_HOST_AUTH_METHOD=scram-sha-256
            POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256
            POSTGRES_ARGS=-h 127.0.0.1
            PGPORT=5432
            PG_MAJOR={PG_MAJOR_DEFAULT}
            POSTGRES_BACKUP_RETENTION=7
        """),
        0,
        0,
        0o600,
    ),
    "/etc/quadlets/postgresql/init.d/test.sql": (
        textwrap.dedent("""
            -- This file is generated by conftest.py for testing purposes.
            CREATE USER test WITH PASSWORD 'test';
            CREATE DATABASE testdb OWNER test;
            GRANT ALL PRIVILEGES ON DATABASE testdb TO test;
            ALTER ROLE test SET client_encoding TO 'utf8';
        """),
        10004,
        10000,
        0o600,
    ),
}

"""
Verify that the postgresql Quadlet is correctly installed and configured on a fresh VM boot.
"""
class TestPostgresqlQuadletInstallUpgradeBackup(helpers.TestPostgresqlQuadlet):

    expected_pg_major = PG_MAJOR_DEFAULT

    def test_can_create_database(self, fcos_host):
        """Should be possible to create a new database."""
        self._run_sql(fcos_host, "CREATE DATABASE upgrade_path_db")
        output = self._run_sql(fcos_host, "SELECT datname FROM pg_database WHERE datname = 'upgrade_path_db'")
        assert output == "upgrade_path_db", f"Unexpected output from SQL query: {output}"
        output = self._run_sql(fcos_host, "CREATE TABLE upgrade_path (version VARCHAR);", database="upgrade_path_db")
        output = self._run_sql(fcos_host, "INSERT INTO upgrade_path (version) SELECT version();", database="upgrade_path_db")

    def test_init_hook_has_created_database(self, fcos_host):
        """The injected init hook has created the test database and user."""
        output = self._run_sql(fcos_host, "SELECT datname FROM pg_database WHERE datname = 'testdb'")
        assert output == "testdb", f"Unexpected output from SQL query: {output}"
        output = self._run_sql(fcos_host, "SELECT 1 FROM pg_roles WHERE rolname = 'test'")
        assert output == "1", f"Unexpected output from SQL query: {output}"

    def test_created_database_and_user_is_working(self, fcos_host):
        """Should be able to connect to the test database with the test user."""
        result = fcos_host.run(
            "podman exec postgresql-server psql -U test -d testdb --csv -t -c %s", "SELECT 1 AS probe"
        )
        assert result.exit_status == 0, f"SQL query failed with exit code {result.exit_status}: {result.stderr}"
        output = result.stdout.strip()
        assert output == "1", f"Unexpected output from SQL query: {output}"

    def test_upgrade_postgresql(self, fcos_host, pg_upgrade_major):
        """Should be able to upgrade PostgreSQL by changing PG_MAJOR and rebooting."""
        # Stop the server to release the data directory
        result = fcos_host.run("systemctl stop postgresql.target")
        assert result.exit_status == 0, f"Failed to stop postgresql.target with exit code {result.exit_status}: {result.stderr}"
        self.check_expected_services(fcos_host, expected_services=[
            { "name": "postgresql-server.service", "state": "inactive", "exists": True },
        ])

        # Change PG_MAJOR in the config.env
        fcos_host.run(f"sed -i 's/^PG_MAJOR=.*/PG_MAJOR={pg_upgrade_major}/' /etc/quadlets/postgresql/config.env")

        # Start the server after changing the data directory
        result = fcos_host.run("systemctl start postgresql.target")
        assert result.exit_status == 0, f"Failed to start postgresql.target with exit code {result.exit_status}: {result.stderr}"
        self.check_expected_services(fcos_host, expected_services=[
            { "name": "postgresql-server.service", "state": "active", "exists": True },
            { "name": "postgresql-init.service", "state": "inactive", "exists": True },
            { "name": "postgresql-upgrade.service", "state": "inactive", "exists": True },
        ])

        # The server_version must reflect the new major version after the upgrade
        output = self._run_sql(fcos_host, "SHOW server_version")
        assert output.startswith(f"{pg_upgrade_major}."), f"Expected PostgreSQL server version to start with {pg_upgrade_major}, but got {output}"

    def test_data_is_still_there_after_upgrade(self, fcos_host, pg_upgrade_major):
        """Data created before the upgrade must still be there after the upgrade."""
        # Check that the old data is still there after the upgrade
        output = self._run_sql(fcos_host, "SELECT datname FROM pg_database WHERE datname = 'upgrade_path_db'")
        assert output == "upgrade_path_db", f"Unexpected output from SQL query: {output}"
        output = self._run_sql(fcos_host, "SELECT datname FROM pg_database WHERE datname = 'testdb'")
        assert output == "testdb", f"Unexpected output from SQL query: {output}"
        result = fcos_host.run(
            "podman exec postgresql-server psql -U test -d testdb --csv -t -c %s", "SELECT 1 AS probe"
        )
        assert result.exit_status == 0, f"SQL query failed with exit code {result.exit_status}: {result.stderr}"

    def test_insert_version(self, fcos_host, pg_upgrade_major):
        """Should be able to insert data into the database after the upgrade."""
        output = self._run_sql(fcos_host, "INSERT INTO upgrade_path (version) SELECT version();", database="upgrade_path_db")

    def test_upgraded_postgresql_version_is_correct(self, fcos_host, pg_upgrade_major):
        """The running PostgreSQL server must report the updated version."""

        # The server_version must reflect the new major version after the upgrade
        output = self._run_sql(fcos_host, "SHOW server_version")
        assert output.startswith(f"{pg_upgrade_major}."), f"Expected PostgreSQL server version to start with {pg_upgrade_major}, but got {output}"

        # The new PostgreSQL major version's image must be pulled and present in Podman after the upgrade
        self.check_expected_podman_images(fcos_host, expected_podman_images=[
            { "name": "docker.io/library/postgres", "tag": f"{pg_upgrade_major}-alpine", "state": "present" },
        ])

    def test_latest_symlink_has_expected_target(self, fcos_host, pg_upgrade_major):
        """The 'latest' symlink must point to the active major-version directory."""
        link = fcos_host.file("/var/lib/quadlets/postgresql/latest")
        assert link.exists
        assert link.is_symlink
        assert link.linked_to == f"/var/lib/quadlets/postgresql/{pg_upgrade_major}"

    def test_create_backup(self, fcos_host):
        """Should be able to create a backup using the backup service."""
        result = fcos_host.run("systemctl start postgresql-backup.service")
        assert result.exit_status == 0, f"Failed to start postgresql-backup.service with exit code {result.exit_status}: {result.stderr}"
        # Check that a backup file has been created in the backup directory
        backup_dir = fcos_host.file("/var/lib/virtiofs/data/postgresql/backup")
        assert backup_dir.exists
        assert backup_dir.is_directory
        backup_list = backup_dir.listdir()
        assert len(backup_list) > 0, "No backup files found in the backup directory after running the backup service!"
        latest_backup = max(backup_list)
        latest_backup_content = fcos_host.file(f"/var/lib/virtiofs/data/postgresql/backup/{latest_backup}").listdir()
        assert len(latest_backup_content) > 0, "No files found in the latest backup directory after running the backup service!"
        assert "backup_manifest" in latest_backup_content, f"Expected 'backup_manifest' file in the backup, but got: {latest_backup_content}"
        assert "base.tar" in latest_backup_content, f"Expected 'base.tar' file in the backup, but got: {latest_backup_content}"
        assert "pg_wal.tar" in latest_backup_content, f"Expected 'pg_wal.tar' file in the backup, but got: {latest_backup_content}"
        assert "dump-upgrade_path_db.sql.gz" in latest_backup_content, f"Expected 'dump-upgrade_path_db.sql.gz' file in the backup, but got: {latest_backup_content}"
        assert "dump-testdb.sql.gz" in latest_backup_content, f"Expected 'dump-testdb.sql.gz' file in the backup, but got: {latest_backup_content}"