diff --git a/test/conftest.py b/test/conftest.py index 26eb71a..9ee489b 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,5 +1,60 @@ +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", } + +MARIADB_PREVIOUS_VERSIONS = { + "10.3": "10.2", + "10.5": "10.3", + "10.11": "10.5", + "11.8": "10.11", +} +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", + "SSL_OPTION", + "PREVIOUS_VERSION", + ], +) +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 +# MariaDB 11.3+ enforces stricter SSL/TLS verification (certificate & hostname checks), +# so tests may require this option for compatibility. +# https://mariadb.org/mission-impossible-zero-configuration-ssl/ +SSL_OPTION = "--disable-ssl-verify-server-cert" if VERSION == "11.8" else "" +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, + SSL_OPTION=SSL_OPTION, # used for tests that require SSL verification to be disabled + PREVIOUS_VERSION=MARIADB_PREVIOUS_VERSIONS.get(VERSION), +) diff --git a/test/run-pytest b/test/run-pytest new file mode 100755 index 0000000..cc76ee7 --- /dev/null +++ b/test/run-pytest @@ -0,0 +1,17 @@ +#!/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 + +PYTHON_VERSION="3.12" +if [[ ! -f "/usr/bin/python$PYTHON_VERSION" ]]; then + PYTHON_VERSION="3.13" +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 0000000..dda9846 --- /dev/null +++ b/test/test_container_basics.py @@ -0,0 +1,117 @@ +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 TestMariaDBBasicsContainer: + """ + Test MariaDB 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="mariadb") + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.app_image.cleanup() + + def test_s2i_usage(self): + """ + Test if the MariaDB container works properly + 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 MariaDB container works properly with mounted application directory. + Steps are: + 1. Copy the test-app directory to a temporary directory + 2. Create a container with the mounted application directory + 3. Test if the container creation succeeds with valid combinations of arguments + 4. Test if the database connection works with the operations user + 5. Stop the container + 6. Remove the temporary directory + """ + data_dir = tempfile.mkdtemp(prefix="/tmp/mariadb-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" + assert 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 0000000..8ce3b56 --- /dev/null +++ b/test/test_container_configuration.py @@ -0,0 +1,277 @@ +import re + +import pytest + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper +from container_ci_suite.container_lib import DatabaseWrapper + +from conftest import VARS + + +class TestMariaDBConfigurationContainer: + """ + Test MariaDB 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( + "mysql_user, mysql_password, mysql_database, mysql_root_password", + [ + ["user", "", "db", ""], + ["", "pass", "db", ""], + ["user", "", "db", "pass"], + ["", "pass", "db", "pass"], + ], + ) + def test_try_image_invalid_combinations( + self, mysql_user, mysql_password, mysql_database, mysql_root_password + ): + """ + Test container creation fails with invalid combinations of arguments. + """ + cid_file_name = "try_image_invalid_combinations" + 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_file_name, + container_args=[ + mysql_user_arg, + mysql_password_arg, + mysql_database_arg, + mysql_root_password_arg, + ], + 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 TestMariaDBConfigurationTests: + """ + Test MariaDB container configuration tests. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.db_config = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.db_config.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_config.cleanup() + + def test_configuration_auto_calculated_settings(self): + """ + Test MariaDB 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=256m", + ) + 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, + database=f"db {VARS.SSL_OPTION}", + ) + 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*25M", + r"read_buffer_size\s*=\s*12M", + r"innodb_log_file_size\s*=\s*38M", + r"innodb_log_buffer_size\s*=\s*38M", + ] + for value in expected_values: + assert re.search(value, db_configuration), ( + f"Word {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, + database=f"db {VARS.SSL_OPTION}", + 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, + database=f"db {VARS.SSL_OPTION}", + 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}") + + def test_configuration_options_settings(self): + """ + Test MariaDB container configuration options. + """ + cid_config_test = "config_test" + assert self.db_config.create_container( + cid_file_name=cid_config_test, + container_args=[ + "--env MYSQL_USER=config_test_user", + "--env MYSQL_PASSWORD=config_test", + "--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 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="config_test_user", + password="config_test", + database=f"db {VARS.SSL_OPTION}", + ) + assert self.db_config.test_db_connection( + container_ip=cip, + username="config_test_user", + password="config_test", + max_attempts=10, + database=f"db {VARS.SSL_OPTION}", + ) + 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", + ] + for value in expected_values: + assert re.search(value, db_configuration), ( + f"Word {value} not found in {db_configuration}" + ) diff --git a/test/test_container_general.py b/test/test_container_general.py new file mode 100644 index 0000000..0a065d7 --- /dev/null +++ b/test/test_container_general.py @@ -0,0 +1,224 @@ +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.podman_wrapper import PodmanCLIWrapper +from container_ci_suite.container_lib import DatabaseWrapper + +from conftest import VARS + + +class TestMariaDBGeneralContainer: + """ + Test MariaDB container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.db_image = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.db_api = DatabaseWrapper(image_name=VARS.IMAGE_NAME) + self.datadir = tempfile.mkdtemp(prefix="/tmp/mariadb-datadir-actions") + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"mkdir -p {self.datadir}/data", + f"chmod -R a+rwx {self.datadir}", + ] + ) + + 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 MariaDB 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", + ) + 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 + # Define variable for login_access function + # Let's check if the login access works for all the users and passwords + # then assert at the end + # If any of the login access fails, return False + # If all the login access works, return True + 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 + continue + assert login_access + # If root password is provided, test the root login access + if root_password: + # Define variable for root_login_access function + # Let's check if the login access works for all the root users and passwords + # then assert at the end + # If any of the login access fails, return False + # If all the login access works, return True + 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 creation. + """ + self.db_api.run_sql_command( + container_ip=cip, + username=username, + password=password, + container_id=VARS.IMAGE_NAME, + database=f"db {VARS.SSL_OPTION}", + 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, + database=f"db {VARS.SSL_OPTION}", + 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, + database=f"db {VARS.SSL_OPTION}", + 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, + database=f"db {VARS.SSL_OPTION}", + sql_cmd="DROP TABLE tbl;", + ) + + @pytest.mark.parametrize( + "action", + [ + "", + "analyze", + "optimize", + ], + ) + def test_datadir_actions(self, action): + """ + Test if the datadir works properly with the different actions. + 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 + """ + mysql_user = "user" + mysql_password = "foo" + mysql_database = "db" + cid_test = f"test_upgrade_{action}" + assert self.db_image.create_container( + cid_file_name=cid_test, + container_args=[ + f"-e MYSQL_USER={mysql_user}", + f"-e MYSQL_PASSWORD={mysql_password}", + f"-e MYSQL_DATABASE={mysql_database}", + f"-v {self.datadir}/data:/var/lib/mysql/data:Z", + ], + ) + cip, cid = self.db_image.get_cip_cid(cid_file_name=cid_test) + assert cip, cid + assert self.db_image.test_db_connection( + container_ip=cip, username=mysql_user, password=mysql_password + ) + 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 0000000..005d393 --- /dev/null +++ b/test/test_container_password.py @@ -0,0 +1,154 @@ +import tempfile + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.container_lib import ContainerTestLibUtils +from container_ci_suite.container_lib import DatabaseWrapper +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper +import pytest + +from conftest import VARS + +user_dir_change = tempfile.mkdtemp(prefix="/tmp/mariadb-user") +ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"chmod -R a+rwx {user_dir_change}", + ] +) +pwd_dir_change = tempfile.mkdtemp(prefix="/tmp/mariadb-pwd") +ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"chmod -R a+rwx {pwd_dir_change}", + ] +) + + +class TestMariaDBPasswordContainer: + """ + Test MariaDB 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="mariadb") + 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", "foo", False), + ("user", "bar", True), + ], + ) + def test_password_change(self, username, password, pwd_change): + """ + Test password change. + """ + + self.password_change_test( + username=username, + password=password, + pwd_dir=pwd_dir_change, + pwd_change=pwd_change, + ) + + @pytest.mark.parametrize( + "username, password, user_change", + [ + ("user", "foo", False), + ("user2", "bar", True), + ], + ) + def test_password_change_new_user_test(self, username, password, user_change): + """ + Test user change. + """ + self.password_change_test( + username=username, + password=password, + pwd_dir=user_dir_change, + user_change=user_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. Check if the userchange, then user2 does exist in the database + 5. Check if the password works + 6. Check if the password does not work + """ + 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, + database=f"db {VARS.SSL_OPTION}", + ) + 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, + database=f"db {VARS.SSL_OPTION}", + 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, + database=f"db {VARS.SSL_OPTION}", + 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_plugin.py b/test/test_container_plugin.py new file mode 100644 index 0000000..d8ab8ee --- /dev/null +++ b/test/test_container_plugin.py @@ -0,0 +1,87 @@ +import re + +from container_ci_suite.container_lib import ContainerTestLib +from container_ci_suite.container_lib import DatabaseWrapper +from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper + +from conftest import VARS + + +class TestMariaDBPluginContainer: + """ + Test MariaDB container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.s2i_db = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.s2i_db.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.s2i_db.cleanup() + + def get_cip_cid(self, cid_file_name): + """ + Get the IP and container ID from the cid file name. + """ + cip = self.s2i_db.get_cip(cid_file_name=cid_file_name) + assert cip + cid = self.s2i_db.get_cid(cid_file_name=cid_file_name) + assert cid + return cip, cid + + def test_plugin_installation(self): + """ + Test plugin installation. + """ + cid_file_name = "plugin_install" + container_args = [ + "-e MYSQL_USER=user", + "-e MYSQL_PASSWORD=foo", + "-e MYSQL_DATABASE=db", + "-e MYSQL_ROOT_PASSWORD=rootpass", + ] + assert self.s2i_db.create_container( + cid_file_name=cid_file_name, + container_args=container_args, + ) + cip, cid = self.get_cip_cid(cid_file_name=cid_file_name) + assert self.s2i_db.test_db_connection( + container_ip=cip, username="root", password="rootpass" + ) + output = self.s2i_db.test_db_connection( + container_ip=cip, + username="root", + password="rootpass", + ) + assert output, "Database connection should not fail" + self.dw_api.run_sql_command( + container_ip=cip, + username="root", + password="rootpass", + container_id=VARS.IMAGE_NAME, + database=f"db {VARS.SSL_OPTION}", + sql_cmd='INSTALL PLUGIN SQL_ERROR_LOG SONAME "sql_errlog" \\G', + ) + # should fail, deliberately not checking return status + sql_cmd = "select * from mysql.IdonotExist;" + self.dw_api.run_sql_command( + container_ip=cip, + username="root", + password="rootpass", + container_id=VARS.IMAGE_NAME, + database=f"db {VARS.SSL_OPTION}", + sql_cmd=sql_cmd, + ignore_error=True, + ) + output = PodmanCLIWrapper.podman_get_file_content( + cid_file_name=cid, + filename="/var/lib/mysql/data/sql_errors.log", + ) + assert re.search("IdonotExist", output) diff --git a/test/test_container_replication.py b/test/test_container_replication.py new file mode 100644 index 0000000..b3bf981 --- /dev/null +++ b/test/test_container_replication.py @@ -0,0 +1,150 @@ +import re +from time import sleep + +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 TestMariaDBReplicationContainer: + """ + Test MariaDB container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.replication_db = ContainerTestLib( + image_name=VARS.IMAGE_NAME, db_type="mariadb" + ) + self.db_wrapper_api = DatabaseWrapper( + image_name=VARS.IMAGE_NAME, db_type="mariadb" + ) + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.replication_db.cleanup() + + def test_replication(self): + """ + Test replication. + """ + cluster_args = "-e MYSQL_MASTER_USER=master -e MYSQL_MASTER_PASSWORD=master -e MYSQL_DATABASE=db" + master_cid_name = "master.cid" + username = "user" + password = "foo" + container_args = [ + f"-e MYSQL_USER={username}", + f"-e MYSQL_PASSWORD={password}", + "-e MYSQL_ROOT_PASSWORD=root", + ] + # Run the MySQL source + assert self.replication_db.create_container( + cid_file_name=master_cid_name, + container_args=container_args, + docker_args=cluster_args, + command="mysqld-master", + ) + master_cip, master_cid = self.replication_db.get_cip_cid( + cid_file_name=master_cid_name + ) + assert master_cip, master_cid + # Run the MySQL replica + slave_cid_name = "slave.cid" + assert self.replication_db.create_container( + cid_file_name=slave_cid_name, + container_args=[ + f"-e MYSQL_MASTER_SERVICE_NAME={master_cip}", + ], + docker_args=cluster_args, + command="mysqld-slave", + ) + slave_cip, slave_cid = self.replication_db.get_cip_cid( + cid_file_name=slave_cid_name + ) + assert slave_cip, slave_cid + # Now wait till the SOURCE will see the REPLICA + assert self.replication_db.test_db_connection( + container_ip=master_cip, + username="root", + password="root", + ) + slave_found = False + for _ in range(3): + result = self.db_wrapper_api.run_sql_command( + container_ip=master_cip, + username="root", + password="root", + container_id=master_cid, + database=f"db {VARS.SSL_OPTION}", + sql_cmd="SHOW SLAVE HOSTS;", + podman_run_command="exec", + ) + if not result: + sleep(3) + continue + if slave_cip in result: + slave_found = True + break + sleep(3) + assert slave_found + assert self.replication_db.test_db_connection( + container_ip=slave_cip, + username="root", + password="root", + ) + sql_cmd = "show slave status\\G;" + mysql_cmd = f'mysql -uroot <<< "{sql_cmd}"' + slave_status = PodmanCLIWrapper.call_podman_command( + cmd=f"exec {slave_cid} bash -c '{mysql_cmd}'", + ) + slave_statuses = [ + "Slave_IO_Running:\\s*Yes", + "Slave_SQL_Running:\\s*Yes", + ] + for status in slave_statuses: + assert re.search(status, slave_status), ( + f"Status {status} not found in {slave_status}" + ) + + self.db_wrapper_api.run_sql_command( + container_ip=master_cip, + username="root", + password="root", + database=f"db {VARS.SSL_OPTION}", + container_id=VARS.IMAGE_NAME, + sql_cmd="CREATE TABLE t1 (a INT);", + max_attempts=3, + ) + self.db_wrapper_api.run_sql_command( + container_ip=master_cip, + username="root", + password="root", + database=f"db {VARS.SSL_OPTION}", + container_id=VARS.IMAGE_NAME, + sql_cmd="INSERT INTO t1 VALUES (24);", + max_attempts=3, + ) + # 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=slave_cip, + username="root", + password="root", + database=f"db {VARS.SSL_OPTION}", + container_id=VARS.IMAGE_NAME, + sql_cmd="select * from t1;", + ) + # sql_cmd = "select * from t1;" + # mysql_cmd = f'mysql -uroot <<< "{sql_cmd}"' + # table_output = PodmanCLIWrapper.call_podman_command( + # cmd=f"exec {slave_cid} bash -c '{mysql_cmd}'", + # ) + assert re.search(r"^a\n^24", table_output, re.MULTILINE), ( + f"Replica {slave_cip} did not get value from MASTER {master_cip}" + ) diff --git a/test/test_container_ssl.py b/test/test_container_ssl.py new file mode 100644 index 0000000..0d61abe --- /dev/null +++ b/test/test_container_ssl.py @@ -0,0 +1,91 @@ +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 TestMariaDBGeneralContainer: + """ + Test MariaDB 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" + 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", + ] + assert self.ssl_db.create_container( + cid_file_name=cid_file_name, + container_args=container_args, + ) + cip, cid = self.ssl_db.get_cip_cid(cid_file_name=cid_file_name) + assert cip, cid + assert self.ssl_db.test_db_connection( + container_ip=cip, username=username, password=password + ) + mysql_cmd = ( + f"mysql --host {cip} -u{username} -p{password} --ssl-ca={ca_cert_path}" + + f" -e 'show status like \"Ssl_cipher\" \\G' db {VARS.SSL_OPTION}" + ) + 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_container_upgrade.py b/test/test_container_upgrade.py new file mode 100644 index 0000000..399864c --- /dev/null +++ b/test/test_container_upgrade.py @@ -0,0 +1,151 @@ +import re +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 +import pytest + +from conftest import VARS + + +class TestMariaDBUpgradeContainer: + """ + Test MariaDB container configuration. + """ + + def setup_method(self): + """ + Setup the test environment. + """ + self.s2i_db = ContainerTestLib(image_name=VARS.IMAGE_NAME) + self.s2i_db.set_new_db_type(db_type="mariadb") + self.tmpdir = tempfile.mkdtemp(prefix="/tmp/mariadb-upgrade") + self.datadir = f"{self.tmpdir}/data" + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"mkdir -p {self.datadir}", + f"chmod -R a+rwx {self.tmpdir}", + ] + ) + self.run_mysqld_cmd = "run-mysqld" + + def teardown_method(self): + """ + Teardown the test environment. + """ + self.s2i_db.cleanup() + + @pytest.mark.parametrize( + "action", + [ + "", + "analyze", + "optimize", + ], + ) + def test_upgrade_test(self, action): + """ + Test container creation fails with invalid combinations of arguments. + """ + mysql_user = "user" + mysql_password = "foo" + mysql_database = "db" + self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action=action + ) + output = self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action=action + ) + assert output, "Version of the data could not be determined" + assert not re.search("Running mysql_upgrade", output), ( + "mysql_upgrade did not run" + ) + + # Create version file that is too old + # Testing upgrade from too old data" + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"echo '5.0.12' > {self.datadir}/mysql_upgrade_info", + ] + ) + container_args = [ + f"-e MYSQL_USER={mysql_user}", + f"-e MYSQL_PASSWORD={mysql_password}", + f"-e MYSQL_DATABASE={mysql_database}", + f"-v {self.datadir}:/var/lib/mysql/data:Z", + "-e MYSQL_DATADIR_ACTION=upgrade-auto", + ] + assert self.s2i_db.assert_container_creation_fails( + cid_file_name="create_container_fails", + container_args=container_args, + command=self.run_mysqld_cmd, + ) + + # Testing upgrade from previous version + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"echo '{VARS.PREVIOUS_VERSION}.12' > {self.datadir}/mysql_upgrade_info", + ] + ) + output = self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action="upgrade-auto" + ) + assert re.search("Running mysql_upgrade", output), "mysql_upgrade did not run" + + # Testing upgrade from the same version + assert ContainerTestLibUtils.commands_to_run( + commands_to_run=[ + f"echo '{VARS.VERSION}.12' > {self.datadir}/mysql_upgrade_info", + ] + ) + output = self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action="upgrade-auto" + ) + assert not re.search("Running mysql_upgrade", output), ( + "mysql_upgrade did not run" + ) + output = self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action="upgrade-auto" + ) + assert not re.search("Running mysql_upgrade", output), ( + "Upgrade happened but it should not when upgrading from current version" + ) + output = self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action="analyze" + ) + assert re.search(r"--analyze --all-databases", output) + + output = self.upgrade_db( + mysql_user=mysql_user, mysql_password=mysql_password, action="optimize" + ) + assert re.search(r"--optimize --all-databases", output) + + def upgrade_db(self, mysql_user, mysql_password, action: str = "") -> str: + """ + Test MariaDB upgrade. + """ + cid_testupg = f"testupg-{mysql_user}-{mysql_password}" + container_args = [ + f"-e MYSQL_USER={mysql_user}", + f"-e MYSQL_PASSWORD={mysql_password}", + "-e MYSQL_DATABASE=db", + f"-v {self.datadir}:/var/lib/mysql/data:Z", + ] + if action: + container_args.append(f"-e MYSQL_DATADIR_ACTION={action}") + assert self.s2i_db.create_container( + cid_file_name=cid_testupg, + container_args=container_args, + command=self.run_mysqld_cmd, + ) + cip, cid = self.s2i_db.get_cip_cid(cid_file_name=cid_testupg) + assert cip, cid + assert self.s2i_db.test_db_connection( + container_ip=cip, username=mysql_user, password=mysql_password + ) + output = PodmanCLIWrapper.podman_logs( + container_id=cid, + ) + PodmanCLIWrapper.call_podman_command(cmd=f"stop {cid}") + return output diff --git a/test/test_ocp_latest_imagestreams.py b/test/test_ocp_latest_imagestreams.py index b30c384..fa46034 100644 --- a/test/test_ocp_latest_imagestreams.py +++ b/test/test_ocp_latest_imagestreams.py @@ -1,6 +1,5 @@ import os import sys -import pytest from pathlib import Path @@ -15,11 +14,10 @@ class TestLatestImagestreams: - def setup_method(self): self.isc = ImageStreamChecker(working_dir=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_local_template.py b/test/test_ocp_local_template.py index c5ba039..1323456 100644 --- a/test/test_ocp_local_template.py +++ b/test/test_ocp_local_template.py @@ -21,9 +21,10 @@ class TestMariaDBDeployTemplate: - def setup_method(self): - self.oc_api = OpenShiftAPI(pod_name_prefix="mariadb-testing", version=VERSION, shared_cluster=True) + self.oc_api = OpenShiftAPI( + pod_name_prefix="mariadb-testing", version=VERSION, shared_cluster=True + ) self.oc_api.import_is("imagestreams/mariadb-rhel.json", "", skip_check=True) def teardown_method(self): @@ -31,10 +32,7 @@ def teardown_method(self): @pytest.mark.parametrize( "template", - [ - "mariadb-ephemeral-template.json", - "mariadb-persistent-template.json" - ] + ["mariadb-ephemeral-template.json", "mariadb-persistent-template.json"], ) def test_python_template_inside_cluster(self, template): short_version = VERSION.replace(".", "") @@ -45,10 +43,10 @@ def test_python_template_inside_cluster(self, template): openshift_args=[ f"MARIADB_VERSION={VERSION}{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) @@ -56,5 +54,5 @@ def test_python_template_inside_cluster(self, template): image_name=f"registry.redhat.io/{OS}/mariadb-{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_shared_helm_template.py b/test/test_ocp_shared_helm_template.py index 676b125..5c5025d 100644 --- a/test/test_ocp_shared_helm_template.py +++ b/test/test_ocp_shared_helm_template.py @@ -1,31 +1,23 @@ -import os - -import pytest -from pathlib import Path - from container_ci_suite.helm import HelmChartsAPI -from conftest 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") - -TAG = TAGS.get(OS) +from conftest import VARS class TestHelmMariaDBPersistent: - def setup_method(self): package_name = "redhat-mariadb-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): self.hc_api.delete_project() @@ -37,8 +29,8 @@ def test_package_persistent(self): assert self.hc_api.helm_package() assert self.hc_api.helm_installation( values={ - "mariadb_version": f"{VERSION}{TAG}", - "namespace": self.hc_api.namespace + "mariadb_version": f"{VARS.VERSION}{VARS.TAG}", + "namespace": self.hc_api.namespace, } ) assert self.hc_api.is_pod_running(pod_name_prefix="mariadb")