Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 41 additions & 3 deletions test/conftest.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import os
import sys
import re

from pathlib import Path
from collections import namedtuple
Expand Down Expand Up @@ -38,6 +39,9 @@


def get_previous_major_version():
"""
Get the previous major version of the PostgreSQL container.
"""
version_dict = {
"13": "12",
"15": "13",
Expand All @@ -47,11 +51,14 @@ def get_previous_major_version():


def get_upgrade_path():
"""
Get the upgrade path of the PostgreSQL container.
"""
upgrade_path = {
"rhel8": "none 12 13 15 16 none",
"rhel9": "none 13 15 16 none",
"rhel10": "none 13 15 16 none",
"fedora": "none 12 13 14 15 16 none",
"rhel9": "none 13 15 16 18 none",
"rhel10": "none 16 18 none",
"fedora": "none 15 16 18 none",
}
for version in upgrade_path.keys():
if version == VARS.VERSION:
Expand All @@ -63,6 +70,9 @@ def get_upgrade_path():


def get_image_id(version):
"""
Get the image ID of the PostgreSQL container.
"""
ns = {
"rhel8": f"registry.redhat.io/rhel8/postgresql-{version}",
"rhel9": f"registry.redhat.io/rhel9/postgresql-{version}",
Expand All @@ -71,3 +81,31 @@ def get_image_id(version):
"c10s": f"quay.io/sclorg/postgresql-{version}-c10s",
}
return ns[VARS.OS]


def check_db_output(
dw_api,
cip,
username,
password,
database,
) -> bool:
"""
Check the database output if the data is inserted correctly
by running a SELECT statement.
"""
output = dw_api.run_sql_command(
container_ip=cip,
username=username,
password=password,
container_id=VARS.IMAGE_NAME,
database=database,
sql_cmd='-At -c "SELECT * FROM tbl;"',
)
expected_db_output = [
"1|2",
"3|4",
"5|6",
]
for row in expected_db_output:
assert re.search(row, output), f"Row {row} not found in {output}"
18 changes: 18 additions & 0 deletions test/run-pytest
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
#!/bin/bash
#
# IMAGE_NAME specifies a name of the candidate image used for testing.
# The image has to be available before this script is executed.
# VERSION specifies the major version of the PostgreSQL in format of X.Y
# OS specifies RHEL version (e.g. OS=rhel10)
#

THISDIR=$(dirname ${BASH_SOURCE[0]})

git show -s

if python3 -c 'import sys; sys.exit(0 if sys.version_info < (3,13) else 1)'; then
PYTHON_VERSION="3.12"
else
PYTHON_VERSION="3"
fi
cd "${THISDIR}" && "python${PYTHON_VERSION}" -m pytest -s -rA --showlocals -vv test_container_*.py
122 changes: 122 additions & 0 deletions test/test_container_basics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
import shutil
import tempfile

from pathlib import Path

from container_ci_suite.container_lib import ContainerTestLib
from container_ci_suite.utils import ContainerTestLibUtils

from conftest import VARS


def build_s2i_app(app_path: Path) -> ContainerTestLib:
container_lib = ContainerTestLib(image_name=VARS.IMAGE_NAME, db_type="postgresql")
app_name = app_path.name
s2i_app = container_lib.build_as_df(
app_path=app_path,
s2i_args="--pull-policy=never",
src_image=VARS.IMAGE_NAME,
dst_image=f"{VARS.IMAGE_NAME}-{app_name}",
)
return s2i_app


class TestPostgreSQLBasicsContainer:
"""
Test PostgreSQL container configuration.
"""

def setup_method(self):
"""
Setup the test environment.
"""
self.app_image = build_s2i_app(app_path=VARS.TEST_APP)
self.app_image.db_lib.db_type = "postgresql"

def teardown_method(self):
"""
Teardown the test environment.
"""
self.app_image.cleanup()

def test_backup_functionality(self):
"""
Test backup functionality of the PostgreSQL container.
Steps are:
1. Test if the container creation fails with invalid combinations of arguments
2. Test if the container creation succeeds with valid combinations of arguments
3. Test if the database connection works
4. Test if the backup functionality works
"""
with tempfile.NamedTemporaryFile(prefix="/tmp/psql-temp-file") as temp_file:
cid_create = "conf_backup"
psql_password = "password"
psql_database = "db"
psql_user = "user"
psql_admin_password = psql_password
psql_backup_user = "backuser"
psql_backup_password = "pass"
self.app_image.assert_container_creation_fails(
cid_file_name=cid_create,
command="",
container_args=[
f"-e POSTGRESQL_PASSWORD={psql_password}",
f"-e POSTGRESQL_DATABASE={psql_database}",
],
)
assert self.app_image.create_container(
cid_file_name=cid_create,
docker_args=[
f"-e POSTGRESQL_USER={psql_user}",
f"-e POSTGRESQL_PASSWORD={psql_password}",
f"-e POSTGRESQL_DATABASE={psql_database}",
f"-e POSTGRESQL_BACKUP_USER={psql_backup_user}",
f"-e POSTGRESQL_BACKUP_PASSWORD={psql_backup_password}",
f"-e POSTGRESQL_ADMIN_PASSWORD={psql_admin_password}",
],
)
cip, cid = self.app_image.get_cip_cid(cid_file_name=cid_create)
assert cip and cid
self.check_psql_connection(cip, psql_user, psql_password)
backup_user_script = (
VARS.TEST_DIR / "test-app/postgresql-init/backup_user.sh"
)
shutil.copy(backup_user_script, temp_file.name)
ContainerTestLibUtils.commands_to_run(
commands_to_run=[
f"setfacl -m u:26:rw- {temp_file.name}",
]
)

cid_backup = "cid_backup"
mount_point = "/opt/app-root/src/postgresql-init/add_backup_user.sh"
assert self.app_image.create_container(
cid_file_name=cid_backup,
docker_args=[
f"-e POSTGRESQL_USER={psql_user}",
f"-e POSTGRESQL_PASSWORD={psql_password}",
f"-e POSTGRESQL_DATABASE={psql_database}",
f"-e POSTGRESQL_BACKUP_USER={psql_backup_user}",
f"-e POSTGRESQL_BACKUP_PASSWORD={psql_backup_password}",
f"-e POSTGRESQL_ADMIN_PASSWORD={psql_admin_password}",
f"-v {temp_file.name}:{mount_point}:z,ro",
],
)
cip, cid = self.app_image.get_cip_cid(cid_file_name=cid_backup)
assert cip and cid
self.check_psql_connection(cip, psql_backup_user, psql_backup_password)

def check_psql_connection(self, cip, psql_user, psql_password):
"""
Check the PostgreSQL connection.
Check also connection to the backup database.
"""
assert self.app_image.test_db_connection(
container_ip=cip, username=psql_user, password=psql_password
)
assert self.app_image.test_db_connection(
container_ip=cip,
username=psql_user,
password=psql_password,
database="backup",
)
Loading
Loading