diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 00000000..ec628020 --- /dev/null +++ b/test/conftest.py @@ -0,0 +1,45 @@ +import os +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) + +TAGS = { + "rhel8": "-el8", + "rhel9": "-el9", + "rhel10": "-el10", +} +TEST_DIR = Path(__file__).parent.absolute() +Vars = namedtuple( + "Vars", + [ + "OS", + "VERSION", + "IMAGE_NAME", + "TEST_DIR", + "TAG", + "TEST_APP", + "VERY_LONG_DB_NAME", + "VERY_LONG_USER_NAME", + ], +) +VERSION = os.getenv("VERSION") +OS = os.getenv("TARGET").lower() +TEST_APP = TEST_DIR / "test-app" +VERY_LONG_DB_NAME = "very_long_database_name_" + "x" * 40 +VERY_LONG_USER_NAME = "very_long_user_name_" + "x" * 40 +VARS = Vars( + OS=OS, + VERSION=VERSION, + IMAGE_NAME=os.getenv("IMAGE_NAME"), + TEST_DIR=Path(__file__).parent.absolute(), + TAG=TAGS.get(OS), + TEST_APP=TEST_APP, + VERY_LONG_DB_NAME=VERY_LONG_DB_NAME, + VERY_LONG_USER_NAME=VERY_LONG_USER_NAME, +) diff --git a/test/constants.py b/test/constants.py deleted file mode 100644 index 26eb71ae..00000000 --- a/test/constants.py +++ /dev/null @@ -1,5 +0,0 @@ -TAGS = { - "rhel8": "-el8", - "rhel9": "-el9", - "rhel10": "-el10", -} diff --git a/test/run b/test/run index d233e106..1664318c 100755 --- a/test/run +++ b/test/run @@ -8,7 +8,6 @@ set -o nounset shopt -s nullglob - THISDIR=$(dirname ${BASH_SOURCE[0]}) source ${THISDIR}/test-lib.sh diff --git a/test/run-openshift-pytest b/test/run-openshift-pytest index e47a9a5c..4eb6f293 100755 --- a/test/run-openshift-pytest +++ b/test/run-openshift-pytest @@ -9,5 +9,9 @@ THISDIR=$(dirname ${BASH_SOURCE[0]}) git show -s - -cd "${THISDIR}" && python3.12 -m pytest -s -rA --showlocals -vv test_ocp_*.py +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_ocp_*.py diff --git a/test/run-pytest b/test/run-pytest new file mode 100755 index 00000000..fe298e56 --- /dev/null +++ b/test/run-pytest @@ -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 MariaDB 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 diff --git a/test/test_container_basics.py b/test/test_container_basics.py new file mode 100644 index 00000000..9276d460 --- /dev/null +++ b/test/test_container_basics.py @@ -0,0 +1,114 @@ +import shutil +import tempfile + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.container_lib import ContainerTestLibUtils +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper +from pathlib import Path + +from conftest import VARS + + +def build_s2i_app(app_path: Path) -> ContainerTestLib: + container_lib = ContainerTestLib(VARS.IMAGE_NAME) + 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 TestMySqlBasicsContainer: + """ + Test MySQL container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.app_image = build_s2i_app(app_path=VARS.TEST_DIR / "test-app") + self.app_image.set_new_db_type(db_type="mysql") + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.app_image.cleanup() + + def test_s2i_usage(self): + """ + Test if MySQL container failed in case of invalid combinations. + 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 + """ + cid_config_build = "s2i_usage_build" + self.app_image.assert_container_creation_fails( + cid_file_name=cid_config_build, + command="", + container_args=[ + "-e MYSQL_USER=root", + "-e MYSQL_PASSWORD=pass", + "-e MYSQL_DATABASE=db", + "-e MYSQL_ROOT_PASSWORD=pass", + ], + ) + assert self.app_image.create_container( + cid_file_name=cid_config_build, + container_args=[ + "-e MYSQL_USER=config_test_user", + "-e MYSQL_PASSWORD=config_test_user", + "-e MYSQL_DATABASE=db", + "-e MYSQL_OPERATIONS_USER=operations_user", + "-e MYSQL_OPERATIONS_PASSWORD=operations_user", + ], + ) + cip, cid = self.app_image.get_cip_cid(cid_file_name=cid_config_build) + assert cip, cid + assert self.app_image.test_db_connection( + container_ip=cip, username="operations_user", password="operations_user" + ) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") + + def test_s2i_usage_with_mount(self): + """ + Test if the MySQL container works properly with mounted application directory. + Steps are: + 1. Copy the test-app directory to a temporary directory and set proper permissions + 2. Create a container with the mounted directory + 3. Test if the database connection works with the operations user + """ + data_dir = tempfile.mkdtemp(prefix="/tmp/mysql-test_data") + shutil.copytree(VARS.TEST_DIR / "test-app", f"{data_dir}/test-app") + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"chown -R 27:27 {data_dir}/test-app", + ] + ) + cid_s2i_test_mount = "s2i_test_mount" + self.app_image.create_container( + cid_file_name=cid_s2i_test_mount, + container_args=[ + "-e MYSQL_USER=config_test_user", + "-e MYSQL_PASSWORD=config_test_user", + "-e MYSQL_DATABASE=db", + "-e MYSQL_OPERATIONS_USER=operations_user", + "-e MYSQL_OPERATIONS_PASSWORD=operations_pass", + f"-v {data_dir}/test-app:/opt/app-root/src/:z", + ], + ) + cip, cid = self.app_image.get_cip_cid(cid_file_name=cid_s2i_test_mount) + assert cip, cid + assert self.app_image.test_db_connection( + container_ip=cip, + username="operations_user", + password="operations_pass", + max_attempts=10, + ) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") + shutil.rmtree(data_dir) diff --git a/test/test_container_configuration.py b/test/test_container_configuration.py new file mode 100644 index 00000000..af65513d --- /dev/null +++ b/test/test_container_configuration.py @@ -0,0 +1,282 @@ +import re + +import pytest + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.engines.database import DatabaseWrapper +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper + +from conftest import VARS + + +class TestMySqlConfigurationContainer: + """ + Test MySQL container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.db = ContainerTestLib(image_name=VARS.IMAGE_NAME) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.db.cleanup() + + def test_container_creation_fails(self): + """ + Test container creation fails with no arguments. + """ + cid_config_test = "container_creation_fails" + assert self.db.assert_container_creation_fails( + cid_file_name=cid_config_test, container_args=[], command="" + ) + + @pytest.mark.parametrize( + "container_args", + [ + ["-e MYSQL_USER=user", "-e MYSQL_DATABASE=db"], + ["-e MYSQL_PASSWORD=pass", "-e MYSQL_DATABASE=db"], + ], + ) + def test_try_image_invalid_combinations(self, container_args): + """ + Test container creation fails with invalid combinations of arguments. + """ + cid_file_name = "try_image_invalid_combinations" + assert self.db.assert_container_creation_fails( + cid_file_name=cid_file_name, container_args=container_args, command="" + ) + + @pytest.mark.parametrize( + "mysql_user, mysql_password, mysql_database, mysql_root_password", + [ + ["user", "pass", "", ""], + [ + "$invalid", + "pass", + "db", + "root_pass", + ], + [ + VARS.VERY_LONG_USER_NAME, + "pass", + "db", + "root_pass", + ], + [ + "user", + "", + "db", + "root_pass", + ], + [ + "user", + "pass", + "$invalid", + "root_pass", + ], + [ + "user", + "pass", + VARS.VERY_LONG_DB_NAME, + "root_pass", + ], + [ + "user", + "pass", + "db", + "", + ], + [ + "root", + "pass", + "db", + "pass", + ], + ], + ) + def test_invalid_configuration_tests( + self, mysql_user, mysql_password, mysql_database, mysql_root_password + ): + """ + Test invalid configuration combinations for MySQL container. + """ + cid_config_test = "invalid_configuration_tests" + mysql_user_arg = f"-e MYSQL_USER={mysql_user}" if mysql_user else "" + mysql_password_arg = ( + f"-e MYSQL_PASSWORD={mysql_password}" if mysql_password else "" + ) + mysql_database_arg = ( + f"-e MYSQL_DATABASE={mysql_database}" if mysql_database else "" + ) + mysql_root_password_arg = ( + f"-e MYSQL_ROOT_PASSWORD={mysql_root_password}" + if mysql_root_password + else "" + ) + assert self.db.assert_container_creation_fails( + cid_file_name=cid_config_test, + container_args=[ + mysql_user_arg, + mysql_password_arg, + mysql_database_arg, + mysql_root_password_arg, + ], + command="", + ) + + +class TestMySqlConfigurationTests: + """ + Test MySQL container configuration tests. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.db_config = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.db_api = DatabaseWrapper(image_name=VARS.IMAGE_NAME) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.db_config.cleanup() + + def test_configuration_auto_calculated_settings(self): + """ + Test MySQL container configuration auto-calculated settings. + Steps are: + 1. Create a container with the given arguments + 2. Check if the container is created successfully + 3. Check if the database connection works + 4. Check if the database configurations are correct + 5. Stop the container + """ + cid_config_test = "auto-config_test" + username = "config_test_user" + password = "config_test" + assert self.db_config.create_container( + cid_file_name=cid_config_test, + container_args=[ + "--env MYSQL_COLLATION=latin2_czech_cs", + "--env MYSQL_CHARSET=latin2", + f"--env MYSQL_USER={username}", + f"--env MYSQL_PASSWORD={password}", + "--env MYSQL_DATABASE=db", + ], + docker_args="--memory=512m", + ) + cip, cid = self.db_config.get_cip_cid(cid_file_name=cid_config_test) + assert cip, cid + assert self.db_config.test_db_connection( + container_ip=cip, + username=username, + password=password, + max_attempts=10, + ) + db_configuration = PodmanCLIWrapper.podman_exec_shell_command( + cid_file_name=cid, + cmd="cat /etc/my.cnf /etc/my.cnf.d/*", + ) + expected_values = [ + r"key_buffer_size\s*=\s*51M", + r"read_buffer_size\s*=\s*25M", + r"innodb_buffer_pool_size\s*=\s*256M", + r"innodb_log_file_size\s*=\s*76M", + r"innodb_log_buffer_size\s*=\s*76M", + r"authentication_policy\s*=\s*'caching_sha2_password,,'", + ] + for value in expected_values: + assert re.search(value, db_configuration), ( + f"Expected value {value} not found in {db_configuration}" + ) + # do some real work to test replication in practice + self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=cid, + sql_cmd="CREATE TABLE tbl (col VARCHAR(20));", + podman_run_command="exec", + ) + show_table_output = self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=cid, + sql_cmd="SHOW CREATE TABLE tbl;", + podman_run_command="exec", + ) + assert "CHARSET=latin2" in show_table_output + assert "COLLATE=latin2_czech_cs" in show_table_output + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") + + # FIX + def test_configuration_options_settings(self): + """ + Test MySQL container configuration options. + """ + cid_config_test = "config_test" + username = "config_test_user" + password = "config_test" + assert self.db_config.create_container( + cid_file_name=cid_config_test, + container_args=[ + f"--env MYSQL_USER={username}", + f"--env MYSQL_PASSWORD={password}", + "--env MYSQL_DATABASE=db", + "--env MYSQL_LOWER_CASE_TABLE_NAMES=1", + "--env MYSQL_LOG_QUERIES_ENABLED=1", + "--env MYSQL_MAX_CONNECTIONS=1337", + "--env MYSQL_FT_MIN_WORD_LEN=8", + "--env MYSQL_FT_MAX_WORD_LEN=15", + "--env MYSQL_MAX_ALLOWED_PACKET=10M", + "--env MYSQL_TABLE_OPEN_CACHE=100", + "--env MYSQL_SORT_BUFFER_SIZE=256K", + "--env MYSQL_KEY_BUFFER_SIZE=16M", + "--env MYSQL_READ_BUFFER_SIZE=16M", + "--env MYSQL_INNODB_BUFFER_POOL_SIZE=16M", + "--env MYSQL_INNODB_LOG_FILE_SIZE=4M", + "--env MYSQL_INNODB_LOG_BUFFER_SIZE=4M", + "--env MYSQL_AUTHENTICATION_POLICY=sha256_password,,", + "--env WORKAROUND_DOCKER_BUG_14203=", + ], + ) + cip, cid = self.db_config.get_cip_cid(cid_file_name=cid_config_test) + assert cip, cid + assert self.db_config.test_db_connection( + container_ip=cip, + username=username, + password=password, + max_attempts=10, + ) + db_configuration = PodmanCLIWrapper.podman_exec_shell_command( + cid_file_name=cid, + cmd="cat /etc/my.cnf /etc/my.cnf.d/*", + ) + expected_values = [ + r"lower_case_table_names\s*=\s*1", + r"general_log\s*=\s*1", + r"max_connections\s*=\s*1337", + r"ft_min_word_len\s*=\s*8", + r"ft_max_word_len\s*=\s*15", + r"max_allowed_packet\s*=\s*10M", + r"table_open_cache\s*=\s*100", + r"sort_buffer_size\s*=\s*256K", + r"key_buffer_size\s*=\s*16M", + r"read_buffer_size\s*=\s*16M", + r"innodb_log_file_size\s*=\s*4M", + r"innodb_log_buffer_size\s*=\s*4M", + r"authentication_policy\s*=\s*'sha256_password,,'", + ] + for value in expected_values: + assert re.search(value, db_configuration), ( + f"Expected value {value} not found in {db_configuration}" + ) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") diff --git a/test/test_container_general.py b/test/test_container_general.py new file mode 100644 index 00000000..ee9e8408 --- /dev/null +++ b/test/test_container_general.py @@ -0,0 +1,239 @@ +import re +import pytest +import tempfile + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.container_lib import ContainerTestLibUtils +from container_ci_suite.engines.database import DatabaseWrapper +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper + +from conftest import VARS + + +class TestMySqlGeneralContainer: + """ + Test MySQL container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.db_image = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.db_image.set_new_db_type(db_type="mysql") + self.db_api = DatabaseWrapper(image_name=VARS.IMAGE_NAME) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.db_image.cleanup() + + @pytest.mark.parametrize( + "docker_args, username, password, root_password", + [ + ("", "user", "pass", ""), + ("", "user1", "pass1", "r00t"), + ("--user 12345", "user", "pass", ""), + ("--user 12345", "user1", "pass1", "r00t"), + ], + ) + def test_run(self, docker_args, username, password, root_password): + """ + Test if the MySQL container works properly with the different arguments + like docker_args, username, password, and root_password. + Steps are: + 1. Create a container with the given arguments + 2. Check if the container is created successfully + 3. Check if the database connection works + 4. Check if mariadb version is correct + 5. Check if the login access works + 6. Check if the local access works + 7. Test the database creation + """ + root_password_arg = ( + f"-e MYSQL_ROOT_PASSWORD={root_password}" if root_password else "" + ) + cid_file_name = f"test_{username}_{password}_{root_password}" + assert self.db_image.create_container( + cid_file_name=cid_file_name, + container_args=[ + f"-e MYSQL_USER={username}", + f"-e MYSQL_PASSWORD={password}", + "-e MYSQL_DATABASE=db", + f"{root_password_arg}", + f"{docker_args}", + ], + command="run-mysqld --innodb_buffer_pool_size=5242880", + ) + cip, cid = self.db_image.get_cip_cid(cid_file_name=cid_file_name) + assert cip, cid + assert self.db_image.test_db_connection( + container_ip=cip, username=username, password=password + ) + output = PodmanCLIWrapper.podman_exec_shell_command( + cid_file_name=cid, + cmd="mysql --version", + ) + assert VARS.VERSION in output + login_access = True + for user, pwd, ret_value in [ + (username, password, True), + (username, f"{password}_foo", False), + ("root", "foo", False), + ("root", "", False), + ]: + test_assert = self.db_image.db_lib.assert_login_access( + container_ip=cip, + username=user, + password=pwd, + expected_success=ret_value, + ) + if not test_assert: + print( + f"Login access failed for {user}:{pwd} with expected success {ret_value}" + ) + login_access = False + assert login_access + if root_password: + root_login_access = True + for user, pwd, ret_value in [ + ("root", root_password, True), + ("root", f"{root_password}_foo", False), + ]: + test_assert = self.db_image.db_lib.assert_login_access( + container_ip=cip, + username=user, + password=pwd, + expected_success=ret_value, + ) + if not test_assert: + print( + f"Root login access failed for {user}:{pwd} with expected success {ret_value}" + ) + root_login_access = False + continue + assert root_login_access + assert self.db_image.db_lib.assert_local_access(container_id=cid) + self.database_test(cip, username, password) + + def database_test(self, cip, username, password): + """ + Test MariaDB database table creation and data insertion is valid. + """ + self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=VARS.IMAGE_NAME, + sql_cmd=[ + "CREATE TABLE tbl (a integer, b integer);", + ], + ) + self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=VARS.IMAGE_NAME, + sql_cmd=[ + "INSERT INTO tbl VALUES (1, 2);", + "INSERT INTO tbl VALUES (3, 4);", + "INSERT INTO tbl VALUES (5, 6);", + ], + ) + output = self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=VARS.IMAGE_NAME, + sql_cmd="SELECT * FROM tbl;", + ) + expected_db_output = [ + r"1\s*\t*2", + r"3\s*\t*4", + r"5\s*\t*6", + ] + for row in expected_db_output: + assert re.search(row, output), f"Row {row} not found in {output}" + self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=VARS.IMAGE_NAME, + sql_cmd="DROP TABLE tbl;", + ) + + def test_datadir_actions(self): + """ + Test container creation fails with invalid combinations of arguments. + """ + cid_testupg1 = "testupg1" + datadir = tempfile.mkdtemp(prefix="/tmp/mysql-datadir-actions") + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"mkdir -p {datadir}/data", + f"chmod -R a+rwx {datadir}", + ] + ) + mysql_user = "user" + mysql_password = "foo" + mysql_database = "db" + assert self.db_image.create_container( + cid_file_name=cid_testupg1, + container_args=[ + f"-e MYSQL_USER={mysql_user}", + f"-e MYSQL_PASSWORD={mysql_password}", + f"-e MYSQL_DATABASE={mysql_database}", + f"-v {datadir}/data:/var/lib/mysql/data:Z", + ], + ) + cip, cid = self.db_image.get_cip_cid(cid_file_name=cid_testupg1) + assert cip, cid + assert self.db_image.test_db_connection( + container_ip=cip, username="user", password="foo" + ) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") + + cid_testupg5 = "testupg5" + assert self.db_image.create_container( + cid_file_name=cid_testupg5, + container_args=[ + f"-e MYSQL_USER={mysql_user}", + f"-e MYSQL_PASSWORD={mysql_password}", + f"-e MYSQL_DATABASE={mysql_database}", + f"-v {datadir}/data:/var/lib/mysql/data:Z", + "-e MYSQL_DATADIR_ACTION=analyze", + ], + ) + cip, cid = self.db_image.get_cip_cid(cid_file_name=cid_testupg5) + assert cip, cid + assert self.db_image.test_db_connection( + container_ip=cip, username=mysql_user, password=mysql_password + ) + output = PodmanCLIWrapper.podman_logs( + container_id=cid, + ) + assert re.search(r"--analyze --all-databases", output) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") + + cid_testupg6 = "testupg6" + assert self.db_image.create_container( + cid_file_name=cid_testupg6, + container_args=[ + f"-e MYSQL_USER={mysql_user}", + f"-e MYSQL_PASSWORD={mysql_password}", + f"-e MYSQL_DATABASE={mysql_database}", + f"-v {datadir}/data:/var/lib/mysql/data:Z", + "-e MYSQL_DATADIR_ACTION=optimize", + ], + ) + cip, cid = self.db_image.get_cip_cid(cid_file_name=cid_testupg6) + assert cip, cid + assert self.db_image.test_db_connection( + container_ip=cip, username=mysql_user, password=mysql_password + ) + output = PodmanCLIWrapper.podman_logs( + container_id=cid, + ) + assert re.search(r"--optimize --all-databases", output) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") diff --git a/test/test_container_password.py b/test/test_container_password.py new file mode 100644 index 00000000..a6e2a165 --- /dev/null +++ b/test/test_container_password.py @@ -0,0 +1,134 @@ +import tempfile +import pytest + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.container_lib import ContainerTestLibUtils +from container_ci_suite.engines.database import DatabaseWrapper +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper + +from conftest import VARS + +pwd_dir_change = tempfile.mkdtemp(prefix="/tmp/mysql-pwd") +assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"chmod -R a+rwx {pwd_dir_change}", + ] +) + +user_dir_change = tempfile.mkdtemp(prefix="/tmp/mysql-user") +assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"chmod -R a+rwx {user_dir_change}", + ] +) + + +class TestMySqlPasswordContainer: + """ + Test MySQL container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.pwd_change = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.pwd_change.set_new_db_type(db_type="mysql") + self.dw_api = DatabaseWrapper(image_name=VARS.IMAGE_NAME) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.pwd_change.cleanup() + + @pytest.mark.parametrize( + "username, password, pwd_change, user_change, test_dir", + [ + ("user", "foo", False, False, pwd_dir_change), + ("user", "bar", True, False, pwd_dir_change), + ("user", "foo", False, False, user_dir_change), + ("user2", "bar", False, True, user_dir_change), + ], + ) + def test_password_change( + self, username, password, pwd_change, user_change, test_dir + ): + """ + Test password change. + """ + self.password_change_test( + username=username, + password=password, + pwd_dir=test_dir, + user_change=user_change, + pwd_change=pwd_change, + ) + + def password_change_test( + self, username, password, pwd_dir, user_change=False, pwd_change=False + ): + """ + Test password change. + Steps are: + 1. Create a container with the given arguments + 2. Check if the container is created successfully + 3. Check if the database connection works + 4. If user_change is True, then 'user' and 'foo' are used for testing connection + 4. Check if the userchange, then user2 does exist in the database logs + 5. Check if the userchange, then sql command should work with the 'user' and 'bar' should + not work and should return an error message + 6. If pwd_change is True, then 'user' and 'pwdfoo' should not work and should return an error message + """ + cid_file_name = f"test_{username}_{password}_{user_change}" + + container_args = [ + f"-e MYSQL_USER={username}", + f"-e MYSQL_PASSWORD={password}", + "-e MYSQL_DATABASE=db", + f"-v {pwd_dir}:/var/lib/mysql/data:Z", + ] + assert self.pwd_change.create_container( + cid_file_name=cid_file_name, + container_args=container_args, + ) + cip, cid = self.pwd_change.get_cip_cid(cid_file_name=cid_file_name) + assert cip, cid + if user_change: + username = "user" + password = "foo" + # Test if the database connection works with the old connection parameters + assert self.pwd_change.test_db_connection( + container_ip=cip, + username=username, + password=password, + ) + if user_change: + mariadb_logs = PodmanCLIWrapper.podman_logs( + container_id=cid, + ) + assert "User user2 does not exist in database" in mariadb_logs + username = "user" + password = "bar" + output = self.dw_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=VARS.IMAGE_NAME, + ignore_error=True, + ) + assert f"Access denied for user '{username}'@" in output, ( + f"The new password {password} should not work, but it does" + ) + if pwd_change: + output = self.dw_api.run_sql_command( + container_ip=cip, + username=username, + password="pwdfoo", + container_id=VARS.IMAGE_NAME, + ignore_error=True, + ) + assert f"Access denied for user '{username}'@" in output, ( + f"The old password {password} should not work, but it does" + ) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") diff --git a/test/test_container_replication.py b/test/test_container_replication.py new file mode 100644 index 00000000..6e97b2b2 --- /dev/null +++ b/test/test_container_replication.py @@ -0,0 +1,106 @@ +import re + +from time import sleep + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.engines.database import DatabaseWrapper + +from conftest import VARS + + +class TestMySqlReplicationContainer: + """ + Test MySQL container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.replication_db = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.replication_db.set_new_db_type(db_type="mysql") + self.db_wrapper_api = DatabaseWrapper(image_name=VARS.IMAGE_NAME) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.replication_db.cleanup() + + def test_replication(self): + """ + Test replication. + """ + cluster_args = "-e MYSQL_SOURCE_USER=source -e MYSQL_SOURCE_PASSWORD=source -e MYSQL_DATABASE=db" + source_cid = "source.cid" + username = "user" + password = "foo" + # Run the MySQL source + assert self.replication_db.create_container( + cid_file_name=source_cid, + container_args=[ + f"-e MYSQL_USER={username}", + f"-e MYSQL_PASSWORD={password}", + "-e MYSQL_ROOT_PASSWORD=root", + "-e MYSQL_INNODB_BUFFER_POOL_SIZE=5M", + ], + docker_args=cluster_args, + command="run-mysqld-source", + ) + source_cip = self.replication_db.get_cip(cid_file_name=source_cid) + source_cid = self.replication_db.get_cid(cid_file_name=source_cid) + assert source_cid + # Run the MySQL replica + replica_cid = "replica.cid" + assert self.replication_db.create_container( + cid_file_name=replica_cid, + container_args=[ + f"-e MYSQL_SOURCE_SERVICE_NAME={source_cip}", + "-e MYSQL_INNODB_BUFFER_POOL_SIZE=5M", + ], + docker_args=cluster_args, + command="run-mysqld-replica", + ) + replica_cip = self.replication_db.get_cip(cid_file_name=replica_cid) + assert replica_cip + replica_cid = self.replication_db.get_cid(cid_file_name=replica_cid) + assert replica_cid + # Now wait till the SOURCE will see the REPLICA + result = self.replication_db.test_db_connection( + container_ip=replica_cip, + username="root", + password="root", + ) + result = self.db_wrapper_api.run_sql_command( + container_ip=source_cip, + username="root", + password="root", + container_id=source_cid, + sql_cmd="SHOW REPLICAS;", + podman_run_command="exec", + ) + assert replica_cip in result, ( + f"Replica {replica_cip} not found in SOURCE {source_cip}" + ) + # do some real work to test replication in practice + table_output = self.db_wrapper_api.run_sql_command( + container_ip=source_cip, + username="root", + password="root", + container_id=source_cid, + sql_cmd="CREATE TABLE t1 (a INT); INSERT INTO t1 VALUES (24);", + podman_run_command="exec", + ) + # let's wait for the table to be created and available for replication + sleep(3) + + table_output = self.db_wrapper_api.run_sql_command( + container_ip=replica_cip, + username="root", + password="root", + container_id=VARS.IMAGE_NAME, + sql_cmd="select * from t1;", + ) + assert re.search(r"^a\n^24", table_output.strip(), re.MULTILINE), ( + f"Replica {replica_cip} did not get value from SOURCE {source_cip}" + ) diff --git a/test/test_container_ssl.py b/test/test_container_ssl.py new file mode 100644 index 00000000..1e77b3e0 --- /dev/null +++ b/test/test_container_ssl.py @@ -0,0 +1,93 @@ +import tempfile +import re + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.container_lib import ContainerTestLibUtils +from container_ci_suite.engines.database import DatabaseWrapper +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper + +from conftest import VARS + + +class TestMySqlGeneralContainer: + """ + Test MySQL container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.ssl_db = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.ssl_db.set_new_db_type(db_type="mysql") + self.db_api = DatabaseWrapper(image_name=VARS.IMAGE_NAME) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.ssl_db.cleanup() + + def test_ssl(self): + """ + Test SSL. + """ + ssl_dir = tempfile.mkdtemp(prefix="/tmp/mysql-ssl_data") + username = "ssl_test_user" + password = "ssl_test" + with open(f"{ssl_dir}/ssl.cnf", mode="wt+") as f: + lines = [ + "[mysqld]", + "ssl-key=${APP_DATA}/mysql-certs/server-key.pem", + "ssl-cert=${APP_DATA}/mysql-certs/server-cert-selfsigned.pem", + ] + f.write("\n".join(lines)) + srv_key_pem = f"{ssl_dir}/server-key.pem" + srv_req_pem = f"{ssl_dir}/server-req.pem" + srv_self_pem = f"{ssl_dir}/server-cert-selfsigned.pem" + openssl_cmd = "openssl req -newkey rsa:2048 -nodes" + openssl_cmd_new = "openssl req -new -x509 -nodes" + subj = "/C=GB/ST=Berkshire/L=Newbury/O=My Server Company" + ContainerTestLibUtils.run_command( + cmd=f"{openssl_cmd} -keyout {srv_key_pem} -subj '{subj}' > {srv_req_pem}" + ) + ContainerTestLibUtils.run_command( + cmd=f"{openssl_cmd_new} -key {srv_key_pem} -batch > {srv_self_pem}" + ) + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"mkdir -p {ssl_dir}/mysql-certs {ssl_dir}/mysql-cfg", + f"cp {ssl_dir}/server-cert-selfsigned.pem {ssl_dir}/mysql-certs/server-cert-selfsigned.pem", + f"cp {ssl_dir}/server-key.pem {ssl_dir}/mysql-certs/server-key.pem", + f"cp {ssl_dir}/ssl.cnf {ssl_dir}/mysql-cfg/ssl.cnf", + f"chown -R 27:27 {ssl_dir}", + ] + ) + + ca_cert_path = "/opt/app-root/src/mysql-certs/server-cert-selfsigned.pem" + cid_file_name = "s2i_test_ssl" + assert self.ssl_db.create_container( + cid_file_name=cid_file_name, + container_args=[ + f"-e MYSQL_USER={username}", + f"-e MYSQL_PASSWORD={password}", + "-e MYSQL_DATABASE=db", + f"-v {ssl_dir}:/opt/app-root/src/:z", + ], + ) + cip = self.ssl_db.get_cip(cid_file_name=cid_file_name) + assert cip + assert self.ssl_db.test_db_connection( + container_ip=cip, username=username, password=password + ) + cid = self.ssl_db.get_cid(cid_file_name=cid_file_name) + assert cid + + mysql_cmd = ( + f"mysql --host {cip} -u{username} -p{password} --ssl-ca={ca_cert_path}" + + " -e 'show status like \"Ssl_cipher\" \\G' db" + ) + ssl_output = PodmanCLIWrapper.podman_run_command( + cmd=f"--rm -v {ssl_dir}:/opt/app-root/src/:z {VARS.IMAGE_NAME} {mysql_cmd}", + ) + assert re.search(r"Value: [A-Z][A-Z0-9-]*", ssl_output) diff --git a/test/test_ocp_mysql_imagestream.py b/test/test_ocp_mysql_imagestream.py index fed19dd6..b4b4c836 100644 --- a/test/test_ocp_mysql_imagestream.py +++ b/test/test_ocp_mysql_imagestream.py @@ -1,46 +1,28 @@ -import os -import sys - import pytest from container_ci_suite.openshift import OpenShiftAPI -from container_ci_suite.utils import check_variables - -from constants import TAGS -if not check_variables(): - print("At least one variable from IMAGE_NAME, OS, VERSION is missing.") - sys.exit(1) - -VERSION = os.getenv("VERSION") -IMAGE_NAME = os.getenv("IMAGE_NAME") -OS = os.getenv("TARGET") -TAG = TAGS.get(OS) +from conftest import VARS class TestMySQLImagestreamTemplate: - def setup_method(self): - self.oc_api = OpenShiftAPI(pod_name_prefix="mysql", version=VERSION, shared_cluster=True) + self.oc_api = OpenShiftAPI( + pod_name_prefix="mysql", version=VARS.VERSION, shared_cluster=True + ) def teardown_method(self): self.oc_api.delete_project() @pytest.mark.parametrize( - "template", - [ - "mysql-ephemeral-template.json", - "mysql-persistent-template.json" - ] + "template", ["mysql-ephemeral-template.json", "mysql-persistent-template.json"] ) def test_imagestream_template(self, template): - os_name = ''.join(i for i in OS if not i.isdigit()) + os_name = "".join(i for i in VARS.OS if not i.isdigit()) assert self.oc_api.deploy_image_stream_template( imagestream_file=f"imagestreams/mysql-{os_name}.json", template_file=f"examples/{template}", app_name=self.oc_api.pod_name_prefix, - openshift_args=[ - f"MYSQL_VERSION={VERSION}{TAG}" - ] + openshift_args=[f"MYSQL_VERSION={VARS.VERSION}{VARS.TAG}"], ) assert self.oc_api.is_pod_running(pod_name_prefix=self.oc_api.pod_name_prefix) diff --git a/test/test_ocp_mysql_imagestream_template.py b/test/test_ocp_mysql_imagestream_template.py index 9442953c..75429805 100644 --- a/test/test_ocp_mysql_imagestream_template.py +++ b/test/test_ocp_mysql_imagestream_template.py @@ -1,43 +1,27 @@ -import os -import sys - import pytest from container_ci_suite.openshift import OpenShiftAPI -from container_ci_suite.utils import check_variables - -from constants import TAGS -if not check_variables(): - print("At least one variable from IMAGE_NAME, OS, VERSION is missing.") - sys.exit(1) - -VERSION = os.getenv("VERSION") -IMAGE_NAME = os.getenv("IMAGE_NAME") -OS = os.getenv("TARGET") -TAG = TAGS.get(OS) +from conftest import VARS class TestMySQLImagestreamTemplate: - def setup_method(self): - self.oc_api = OpenShiftAPI(pod_name_prefix="mysql", version=VERSION, shared_cluster=True) + self.oc_api = OpenShiftAPI( + pod_name_prefix="mysql", version=VARS.VERSION, shared_cluster=True + ) def teardown_method(self): self.oc_api.delete_project() @pytest.mark.parametrize( - "template", - [ - "mysql-ephemeral-template.json", - "mysql-persistent-template.json" - ] + "template", ["mysql-ephemeral-template.json", "mysql-persistent-template.json"] ) def test_imagestream_template(self, template): - os_name = ''.join(i for i in OS if not i.isdigit()) + os_name = "".join(i for i in VARS.OS if not i.isdigit()) assert self.oc_api.deploy_image_stream_template( imagestream_file=f"imagestreams/mysql-{os_name}.json", template_file=f"examples/{template}", - app_name=self.oc_api.pod_name_prefix + app_name=self.oc_api.pod_name_prefix, ) assert self.oc_api.is_pod_running(pod_name_prefix=self.oc_api.pod_name_prefix) diff --git a/test/test_ocp_mysql_latest_imagestreams.py b/test/test_ocp_mysql_latest_imagestreams.py index b30c384b..b357060f 100644 --- a/test/test_ocp_mysql_latest_imagestreams.py +++ b/test/test_ocp_mysql_latest_imagestreams.py @@ -1,25 +1,13 @@ -import os -import sys -import pytest - -from pathlib import Path - from container_ci_suite.imagestreams import ImageStreamChecker -from container_ci_suite.utils import check_variables -TEST_DIR = Path(os.path.abspath(os.path.dirname(__file__))) - -if not check_variables(): - print("At least one variable from IMAGE_NAME, OS, VERSION is missing.") - sys.exit(1) +from conftest import VARS class TestLatestImagestreams: - def setup_method(self): - self.isc = ImageStreamChecker(working_dir=TEST_DIR.parent) + self.isc = ImageStreamChecker(working_dir=VARS.TEST_DIR.parent) def test_latest_imagestream(self): self.latest_version = self.isc.get_latest_version() - assert self.latest_version != "" + assert self.latest_version self.isc.check_imagestreams(self.latest_version) diff --git a/test/test_ocp_mysql_local_template.py b/test/test_ocp_mysql_local_template.py index fee611ec..3bb3a15b 100644 --- a/test/test_ocp_mysql_local_template.py +++ b/test/test_ocp_mysql_local_template.py @@ -1,60 +1,42 @@ -import os -import sys - import pytest from container_ci_suite.openshift import OpenShiftAPI -from container_ci_suite.utils import check_variables - -from constants import TAGS -if not check_variables(): - print("At least one variable from IMAGE_NAME, OS, VERSION is missing.") - sys.exit(1) - - -VERSION = os.getenv("VERSION") -IMAGE_NAME = os.getenv("IMAGE_NAME") -OS = os.getenv("TARGET") - -TAG = TAGS.get(OS) +from conftest import VARS class TestMySQLDeployTemplate: - def setup_method(self): - self.oc_api = OpenShiftAPI(pod_name_prefix="mysql-testing", version=VERSION, shared_cluster=True) + self.oc_api = OpenShiftAPI( + pod_name_prefix="mysql-testing", version=VARS.VERSION, shared_cluster=True + ) self.oc_api.import_is("imagestreams/mysql-rhel.json", "", skip_check=True) def teardown_method(self): self.oc_api.delete_project() @pytest.mark.parametrize( - "template", - [ - "mysql-ephemeral-template.json", - "mysql-persistent-template.json" - ] + "template", ["mysql-ephemeral-template.json", "mysql-persistent-template.json"] ) def test_template_inside_cluster(self, template): - short_version = VERSION.replace(".", "") + short_version = VARS.VERSION.replace(".", "") assert self.oc_api.deploy_template_with_image( - image_name=IMAGE_NAME, + image_name=VARS.IMAGE_NAME, template=template, name_in_template="mysql", openshift_args=[ - f"MYSQL_VERSION={VERSION}{TAG}", + f"MYSQL_VERSION={VARS.VERSION}{VARS.TAG}", f"DATABASE_SERVICE_NAME={self.oc_api.pod_name_prefix}", - f"MYSQL_USER=testu", - f"MYSQL_PASSWORD=testp", - f"MYSQL_DATABASE=testdb" - ] + "MYSQL_USER=testu", + "MYSQL_PASSWORD=testp", + "MYSQL_DATABASE=testdb", + ], ) assert self.oc_api.is_pod_running(pod_name_prefix=self.oc_api.pod_name_prefix) assert self.oc_api.check_command_internal( - image_name=f"registry.redhat.io/{OS}/mysql-{short_version}", + image_name=f"registry.redhat.io/{VARS.OS}/mysql-{short_version}", service_name=self.oc_api.pod_name_prefix, cmd="echo 'SELECT 42 as testval\\g' | mysql --connect-timeout=15 -h testdb -utestu -ptestp", - expected_output="42" + expected_output="42", ) diff --git a/test/test_ocp_mysql_shared_helm_imagestreams.py b/test/test_ocp_mysql_shared_helm_imagestreams.py index 4eaa61a3..d71e71bc 100644 --- a/test/test_ocp_mysql_shared_helm_imagestreams.py +++ b/test/test_ocp_mysql_shared_helm_imagestreams.py @@ -1,22 +1,23 @@ -import os - import pytest -from pathlib import Path from container_ci_suite.helm import HelmChartsAPI -test_dir = Path(os.path.abspath(os.path.dirname(__file__))) +from conftest import VARS class TestHelmRHELMySQLImageStreams: - def setup_method(self): package_name = "redhat-mysql-imagestreams" - path = test_dir - self.hc_api = HelmChartsAPI(path=path, package_name=package_name, tarball_dir=test_dir, shared_cluster=True) + self.hc_api = HelmChartsAPI( + path=VARS.TEST_DIR, + package_name=package_name, + tarball_dir=VARS.TEST_DIR, + shared_cluster=True, + ) self.hc_api.clone_helm_chart_repo( - repo_url="https://github.com/sclorg/helm-charts", repo_name="helm-charts", - subdir="charts/redhat" + repo_url="https://github.com/sclorg/helm-charts", + repo_name="helm-charts", + subdir="charts/redhat", ) def teardown_method(self): diff --git a/test/test_ocp_mysql_shared_helm_template.py b/test/test_ocp_mysql_shared_helm_template.py index c80d0e1e..31c5046a 100644 --- a/test/test_ocp_mysql_shared_helm_template.py +++ b/test/test_ocp_mysql_shared_helm_template.py @@ -1,29 +1,21 @@ -import os - -import pytest -from pathlib import Path - from container_ci_suite.helm import HelmChartsAPI -from constants import TAGS -test_dir = Path(os.path.abspath(os.path.dirname(__file__))) - -VERSION = os.getenv("VERSION") -IMAGE_NAME = os.getenv("IMAGE_NAME") -OS = os.getenv("TARGET") +from conftest import VARS -TAG = TAGS.get(OS) - class TestHelmMySQLDBPersistent: - def setup_method(self): package_name = "redhat-mysql-persistent" - path = test_dir - self.hc_api = HelmChartsAPI(path=path, package_name=package_name, tarball_dir=test_dir, shared_cluster=True) + self.hc_api = HelmChartsAPI( + path=VARS.TEST_DIR, + package_name=package_name, + tarball_dir=VARS.TEST_DIR, + shared_cluster=True, + ) self.hc_api.clone_helm_chart_repo( - repo_url="https://github.com/sclorg/helm-charts", repo_name="helm-charts", - subdir="charts/redhat" + repo_url="https://github.com/sclorg/helm-charts", + repo_name="helm-charts", + subdir="charts/redhat", ) def teardown_method(self): @@ -37,9 +29,9 @@ def test_package_persistent(self): assert self.hc_api.helm_package() assert self.hc_api.helm_installation( values={ - "mysql_version": f"{VERSION}{TAG}", + "mysql_version": f"{VARS.VERSION}{VARS.TAG}", "namespace": self.hc_api.namespace, - "database_service_name": "mysql" + "database_service_name": "mysql", } ) assert self.hc_api.is_pod_running(pod_name_prefix="mysql")