diff --git a/container_ci_suite/engines/database.py b/container_ci_suite/engines/database.py index 7236027..75a4df1 100644 --- a/container_ci_suite/engines/database.py +++ b/container_ci_suite/engines/database.py @@ -38,8 +38,9 @@ import re import subprocess import time -from typing import Optional, Literal, Union +from typing import Optional, Literal, Union, Dict from enum import Enum +from urllib.parse import quote, urlencode from container_ci_suite.engines.podman_wrapper import PodmanCLIWrapper @@ -74,11 +75,11 @@ class DatabaseWrapper: Example: >>> # MySQL >>> db = DatabaseWrapper(image_name="mysql:8.0", db_type="mysql") - >>> assert db.assert_login_success("172.17.0.2", "user", "pass") + >>> assert db.assert_login_access("172.17.0.2", "user", "pass", expected_success=True) >>> # PostgreSQL >>> db = DatabaseWrapper(image_name="postgres:13", db_type="postgresql") - >>> assert db.assert_login_success("172.17.0.2", "user", "pass") + >>> assert db.assert_login_access("172.17.0.2", "user", "pass", expected_success=True) """ def __init__( @@ -145,48 +146,6 @@ def wait_for_database( logger.error("Database not ready after %s attempts", max_attempts) return False - def assert_login_success( - self, - container_ip: str, - username: str, - password: str, - database: str = "db", - port: int = None, - ) -> bool: - """ - Assert that login succeeds for the given credentials. - - This is a convenience function that calls assert_login_access with - expected_success=True. It works for both MySQL and PostgreSQL. - - Args: - container_ip: IP address of the container - username: Username to test login with - password: Password to test login with - database: Database name to connect to (default: "db") - port: Port number (default: 3306 for MySQL, 5432 for PostgreSQL) - - Returns: - True if login succeeds, False otherwise - - Example: - >>> # MySQL - >>> db = DatabaseWrapper(image_name="mysql:8.0", db_type="mysql") - >>> assert db.assert_login_success("172.17.0.2", "user", "pass") - - >>> # PostgreSQL - >>> db = DatabaseWrapper(image_name="postgres:13", db_type="postgresql") - >>> assert db.assert_login_success("172.17.0.2", "user", "pass") - """ - return self.assert_login_access( - container_ip=container_ip, - username=username, - password=password, - expected_success=True, - database=database, - port=port, - ) - def assert_login_access( self, container_ip: str, @@ -309,18 +268,15 @@ def _test_postgresql_login( "postgresql://$PGUSER@$CONTAINER_IP:5432/${DB-db}" -At -c 'SELECT 1;' """ try: - str_port = str(port) if port else "5432" - connection_string = ( - f"postgresql://{username}@{container_ip}:{str_port}/{database}" - ) - - cmd = ( - f"run --rm -e PGPASSWORD={password} {self.image_name} " - f"psql -v ON_ERROR_STOP=1 '{connection_string}' -At -c 'SELECT 1;'" - ) - - output = PodmanCLIWrapper.call_podman_command(cmd=cmd, return_output=True) - + output = self.postgresql_cmd(container_ip, + username, + password, + container_id=self.image_name, + database=database, + port=port if port else 5432, + extra_args="-At", + sql_command="-c 'SELECT 1;'" + ) # PostgreSQL returns "1" on successful query return "1" in output.strip() @@ -361,8 +317,8 @@ def mysql_cmd( port: Port number (default: 3306) extra_args: Additional arguments to pass to mysql command sql_command: SQL command to execute (e.g., "-e 'SELECT 1;'") + container_id: The container ID (usually the image name) podman_run_command: Podman run command to use (default: "run --rm") - ignore_error: Ignore error and return output (default: False) Returns: Command output as string @@ -410,9 +366,10 @@ def postgresql_cmd( password: str, container_id: Optional[str] = None, database: str = "db", + uri_params: Dict[str, str] = None, port: int = 5432, extra_args: str = "", - sql_command: Optional[str] = None, + sql_command: Optional[str] = "", podman_run_command: Optional[str] = "run --rm", docker_args: str = "", ) -> str: @@ -431,7 +388,9 @@ def postgresql_cmd( container_ip: IP address of the PostgreSQL container username: PostgreSQL username password: PostgreSQL password + container_id: Container ID database: Database name (default: "db") + uri_params: Extra parameters in the postgreSQL connection URI port: Port number (default: 5432) extra_args: Additional arguments to pass to psql command sql_command: SQL command to execute (e.g., "-c 'SELECT 1;'") @@ -448,25 +407,24 @@ def postgresql_cmd( >>> output = db.postgresql_cmd("172.17.0.2", "user", "pass", ... sql_command="-c 'SELECT 1;'") """ - connection_string = f"postgresql://{username}@{container_ip}:{port}/{database}" + encoded_username = quote(username) + encoded_database = quote(database) + encoded_params = f"?{urlencode(uri_params)}" if uri_params else "" + connection_string = f"postgresql://{encoded_username}@{container_ip}:{port}/{encoded_database}{encoded_params}" if not container_id: container_id = self.image_name cmd_parts = [ podman_run_command, docker_args, - f"-e PGPASSWORD={password}", + f'-e PGPASSWORD="{password}"', container_id, "psql", "-v ON_ERROR_STOP=1", connection_string, + extra_args, + sql_command ] - if extra_args: - cmd_parts.append(extra_args) - - if sql_command: - cmd_parts.append(sql_command) - cmd = " ".join(cmd_parts) return PodmanCLIWrapper.call_podman_command(cmd=cmd, return_output=True) @@ -477,6 +435,7 @@ def test_connection( username: str, password: str, database: str = "db", + uri_params: Dict[str, str] = None, max_attempts: int = 60, sleep_time: int = 3, sql_cmd: Optional[str] = None, @@ -493,7 +452,6 @@ def test_connection( username: Database username password: Database password database: Database name (default: "db") - port: Port number (default: 3306 for MySQL, 5432 for PostgreSQL) max_attempts: Maximum number of connection attempts (default: 60) sleep_time: Seconds to wait between attempts (default: 3) sql_cmd: SQL command to execute (e.g., "SELECT 1;") @@ -516,6 +474,7 @@ def test_connection( username=username, password=password, database=database, + uri_params=uri_params, sql_command=sql_cmd, ) else: # mysql or mariadb @@ -625,9 +584,11 @@ def run_sql_command( max_attempts: Maximum number of attempts (default: 60) sleep_time: Time to sleep between attempts (default: 3) container_id: Container ID or name + docker_args: Additional arguments for the `podman run` command podman_run_command: Podman run command to use (default: "run --rm") ignore_error: Ignore error and return output (default: False) expected_output: Expected output of the command (default: None) + use_bash: Set to true to execute each command in `sql_cmd` in bash Returns: Command output as string or False if command failed """ diff --git a/setup.py b/setup.py index 011f44f..2eccea9 100644 --- a/setup.py +++ b/setup.py @@ -43,7 +43,7 @@ def get_requirements(): description="A python3 container CI tool for testing images.", long_description=long_description, long_description_content_type="text/markdown", - version="0.11.2", + version="0.11.3", keywords="tool,containers,images,tests", packages=find_packages(exclude=["tests"]), url="https://github.com/sclorg/container-ci-suite", diff --git a/tests/test_database_wrapper.py b/tests/test_database_wrapper.py index 9076589..55dc103 100644 --- a/tests/test_database_wrapper.py +++ b/tests/test_database_wrapper.py @@ -54,27 +54,6 @@ def test_init_mysql(self): assert self.db.image_name == self.image_name assert self.db.db_type == "mysql" - @patch( - "container_ci_suite.engines.podman_wrapper.PodmanCLIWrapper.call_podman_command" - ) - def test_assert_login_success_mysql(self, mock_podman): - """Test assert_login_success for MySQL when login succeeds.""" - mock_podman.return_value = "1\n" - - result = self.db.assert_login_success( - container_ip=self.container_ip, - username=self.username, - password=self.password, - ) - - assert result is True - mock_podman.assert_called_once() - call_args = mock_podman.call_args[1]["cmd"] - assert "mysql" in call_args - assert f"--host {self.container_ip}" in call_args - assert f"-u{self.username}" in call_args - assert f"-p{self.password}" in call_args - @patch( "container_ci_suite.engines.podman_wrapper.PodmanCLIWrapper.call_podman_command" ) @@ -144,26 +123,6 @@ def test_init_postgresql(self): assert self.db.image_name == self.image_name assert self.db.db_type == "postgresql" - @patch( - "container_ci_suite.engines.podman_wrapper.PodmanCLIWrapper.call_podman_command" - ) - def test_assert_login_success_postgresql(self, mock_podman): - """Test assert_login_success for PostgreSQL when login succeeds.""" - mock_podman.return_value = "1" - - result = self.db.assert_login_success( - container_ip=self.container_ip, - username=self.username, - password=self.password, - ) - - assert result is True - mock_podman.assert_called_once() - call_args = mock_podman.call_args[1]["cmd"] - assert "psql" in call_args - assert f"PGPASSWORD={self.password}" in call_args - assert f"postgresql://{self.username}@{self.container_ip}" in call_args - @patch( "container_ci_suite.engines.podman_wrapper.PodmanCLIWrapper.call_podman_command" ) @@ -369,8 +328,8 @@ def test_special_characters_in_password_mysql(self, mock_podman): db = DatabaseWrapper(image_name="mysql:8.0", db_type="mysql") special_password = "p@ss!w0rd#123" - result = db.assert_login_success( - container_ip="172.17.0.2", username="user", password=special_password + result = db.assert_login_access( + container_ip="172.17.0.2", username="user", password=special_password, expected_success=True ) assert result is True @@ -384,8 +343,8 @@ def test_special_characters_in_password_postgresql(self, mock_podman): db = DatabaseWrapper(image_name="postgres:13", db_type="postgresql") special_password = "p@ss!w0rd#123" - result = db.assert_login_success( - container_ip="172.17.0.2", username="user", password=special_password + result = db.assert_login_access( + container_ip="172.17.0.2", username="user", password=special_password, expected_success=True ) assert result is True @@ -399,11 +358,12 @@ def test_custom_port_mysql(self, mock_podman): db = DatabaseWrapper(image_name="mysql:8.0", db_type="mysql") custom_port = 3307 - result = db.assert_login_success( + result = db.assert_login_access( container_ip="172.17.0.2", username="user", password="pass", port=custom_port, + expected_success=True ) assert result is True @@ -419,11 +379,12 @@ def test_custom_port_postgresql(self, mock_podman): db = DatabaseWrapper(image_name="postgres:13", db_type="postgresql") custom_port = 5433 - result = db.assert_login_success( + result = db.assert_login_access( container_ip="172.17.0.2", username="user", password="pass", port=custom_port, + expected_success=True ) assert result is True @@ -446,7 +407,7 @@ def test_real_mysql_connection(self): password = "rootpass" assert db.test_connection(container_ip, username, password, max_attempts=10) - assert db.assert_login_success(container_ip, username, password) + assert db.assert_login_access(container_ip, username, password, True) @pytest.mark.integration @pytest.mark.skip(reason="Requires actual PostgreSQL container running") @@ -460,7 +421,7 @@ def test_real_postgresql_connection(self): password = "postgres" assert db.test_connection(container_ip, username, password, max_attempts=10) - assert db.assert_login_success(container_ip, username, password) + assert db.assert_login_access(container_ip, username, password, True) class TestDatabaseWrapperDocumentation: @@ -474,7 +435,7 @@ def test_docstring_example_mysql(self, mock_podman): mock_podman.return_value = "1\n" db = DatabaseWrapper(image_name="mysql:8.0", db_type="mysql") - assert db.assert_login_success("172.17.0.2", "user", "pass") + assert db.assert_login_access("172.17.0.2", "user", "pass", True) @patch( "container_ci_suite.engines.podman_wrapper.PodmanCLIWrapper.call_podman_command" @@ -484,7 +445,7 @@ def test_docstring_example_postgresql(self, mock_podman): mock_podman.return_value = "1" db = DatabaseWrapper(image_name="postgres:13", db_type="postgresql") - assert db.assert_login_success("172.17.0.2", "user", "pass") + assert db.assert_login_access("172.17.0.2", "user", "pass", True) # Run tests with: pytest tests/test_database_wrapper.py -v