Container PyTest suite for Postgresql-container

test/conftest.py

@@ -1,11 +1,13 @@
 import os
+import re
 import sys
 
 from pathlib import Path
 from collections import namedtuple
 
 from container_ci_suite.utils import check_variables
 
+
 if not check_variables():
     sys.exit(1)
 
@@ -20,12 +22,40 @@
         "TAG",
         "TEST_APP",
         "VERY_LONG_IDENTIFIER",
+        "PREVIOUS_VERSION",
+        "PAGILA",
+        "MIGRATION_PATHS",
+        "UPGRADE_PATH_DICT",
+        "SQL_CMDS",
     ],
 )
 VERSION = os.getenv("VERSION")
 OS = os.getenv("TARGET").lower()
 TEST_APP = TEST_DIR / "test-app"
 VERY_LONG_IDENTIFIER = "very_long_identifier_" + "x" * 40
+MIGRATION_PATHS = ["12", "13", "15", "16", "18"]
+PREVIOUS_VERSION_DICT = {
+    "13": "12",
+    "15": "13",
+    "16": "15",
+    "18": "16",
+}
+PAGILA = ["pagila-schema.sql", "pagila-data.sql", "pagila-insert-data.sql"]
+SQL_CMDS = {
+    "select count(*) from information_schema.triggers;": "16",
+    "select count(*) from staff;": "2",
+    "select * from information_schema.tables;": "28",
+}
+RHEL9 = [13, 15, 16, 18]
+RHEL10 = [16, 18]
+UPGRADE_PATH_DICT = {
+    "rhel8": [12, 13, 15, 16],
+    "rhel9": RHEL9,
+    "c9s": RHEL9,
+    "rhel10": RHEL10,
+    "c10s": RHEL10,
+    "fedora": [15, 16, 18],
+}
 VARS = Vars(
     OS=OS,
     VERSION=VERSION,
@@ -34,40 +64,58 @@
     TAG=OS.replace("rh", "-", 1),
     TEST_APP=TEST_APP,
     VERY_LONG_IDENTIFIER=VERY_LONG_IDENTIFIER,
+    PREVIOUS_VERSION=PREVIOUS_VERSION_DICT.get(VERSION),
+    PAGILA=PAGILA,
+    MIGRATION_PATHS=MIGRATION_PATHS,
+    UPGRADE_PATH_DICT=UPGRADE_PATH_DICT,
+    SQL_CMDS=SQL_CMDS,
 )
 
 
-def get_previous_major_version():
-    version_dict = {
-        "13": "12",
-        "15": "13",
-        "16": "15",
-    }
-    return version_dict.get(VARS.VERSION)
-
-
 def get_upgrade_path():
-    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",
-    }
-    for version in upgrade_path.keys():
-        if version == VARS.VERSION:
-            break
-        prev = version
-    if prev == "none":
+    """
+    Get the upgrade path of the PostgreSQL container.
+    """
+    if VARS.VERSION not in UPGRADE_PATH_DICT[VARS.OS]:
         return None
-    return prev
+    current_index = UPGRADE_PATH_DICT[VARS.OS].index(VARS.VERSION)
+    if current_index - 1 == -1:
+        return None
+    return UPGRADE_PATH_DICT[VARS.OS][current_index - 1]
 
 
 def get_image_id(version):
-    ns = {
-        "rhel8": f"registry.redhat.io/rhel8/postgresql-{version}",
-        "rhel9": f"registry.redhat.io/rhel9/postgresql-{version}",
-        "rhel10": f"registry.redhat.io/rhel10/postgresql-{version}",
-        "c9s": f"quay.io/sclorg/postgresql-{version}-c9s",
-        "c10s": f"quay.io/sclorg/postgresql-{version}-c10s",
-    }
-    return ns[VARS.OS]
+    """
+    Get the image ID of the PostgreSQL container.
+    """
+    if VARS.OS.startswith("rhel"):
+        return f"registry.redhat.io/{VARS.OS}/postgresql-{version}"
+    return f"quay.io/sclorg/postgresql-{version}-{VARS.OS}"
+
+
+def check_db_output(
+    dw_api,
+    cip,
+    username,
+    password,
+    database,
+):
+    """
+    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}"