DPE-5210 Common credentials fixture and exec timeout workaround (#493)

* refactor for common credentials fixture in tests

* timeout issue workaround

* add fixture

* lib bump

Author: paulomach

lib/charms/mysql/v0/mysql.py

@@ -134,7 +134,7 @@ def wait_until_mysql_connection(self) -> None:
 # Increment this major API version when introducing breaking changes
 LIBAPI = 0
 
-LIBPATCH = 69
+LIBPATCH = 70
 
 UNIT_TEARDOWN_LOCKNAME = "unit-teardown"
 UNIT_ADD_LOCKNAME = "unit-add"
@@ -1536,7 +1536,9 @@ def cluster_metadata_exists(self, from_instance: str) -> bool:
         )
 
         try:
-            output = self._run_mysqlsh_script("\n".join(check_cluster_metadata_commands))
+            output = self._run_mysqlsh_script(
+                "\n".join(check_cluster_metadata_commands), timeout=10
+            )
         except MySQLClientError:
             logger.warning(f"Failed to check if cluster metadata exists {from_instance=}")
             return False

src/mysql_k8s_helpers.py

@@ -581,7 +581,7 @@ def _execute_commands(
             raise MySQLExecError from None
 
     def _run_mysqlsh_script(
-        self, script: str, verbose: int = 1, timeout: Optional[int] = None
+        self, script: str, timeout: Optional[int] = None, verbose: int = 0
     ) -> str:
         """Execute a MySQL shell script.
 
@@ -611,8 +611,14 @@ def _run_mysqlsh_script(
             MYSQLSH_SCRIPT_FILE,
         ]
 
+        # workaround for timeout not working on pebble exec
+        # https://github.com/canonical/operator/issues/1329
+        if timeout:
+            cmd.insert(0, str(timeout))
+            cmd.insert(0, "timeout")
+
         try:
-            process = self.container.exec(cmd, timeout=timeout)
+            process = self.container.exec(cmd)
             stdout, _ = process.wait_output()
             return stdout
         except ExecError:

tests/integration/conftest.py

@@ -0,0 +1,25 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import logging
+
+import pytest
+from pytest_operator.plugin import OpsTest
+
+from constants import SERVER_CONFIG_USERNAME
+
+from . import juju_
+from .high_availability.high_availability_helpers import get_application_name
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.fixture(scope="function")
+async def credentials(ops_test: OpsTest):
+    """Return the credentials for the MySQL cluster."""
+    logger.info("Getting credentials for the MySQL cluster")
+    mysql_app_name = get_application_name(ops_test, "mysql-k8s")
+    unit = ops_test.model.applications[mysql_app_name].units[0]
+    credentials = await juju_.run_action(unit, "get-password", username=SERVER_CONFIG_USERNAME)
+
+    yield credentials

tests/integration/helpers.py

@@ -159,7 +159,7 @@ async def get_primary_unit(
     return None
 
 
-async def execute_queries_on_unit(
+def execute_queries_on_unit(
     unit_address: str,
     username: str,
     password: str,
@@ -474,7 +474,7 @@ async def retrieve_database_variable_value(
     server_config_creds = await get_server_config_credentials(unit)
     queries = [f"SELECT @@{variable_name};"]
 
-    output = await execute_queries_on_unit(
+    output = execute_queries_on_unit(
         unit_ip, server_config_creds["username"], server_config_creds["password"], queries
     )
 

tests/integration/high_availability/conftest.py

@@ -58,8 +58,8 @@ def built_charm(ops_test: OpsTest) -> pathlib.Path:
     return packed_charm[0].resolve(strict=True)
 
 
-@pytest.fixture()
-async def highly_available_cluster(ops_test: OpsTest) -> None:
+@pytest.fixture(scope="module")
+async def highly_available_cluster(ops_test: OpsTest):
     """Run the set up for high availability tests.
 
     Args:

tests/integration/high_availability/high_availability_helpers.py

@@ -26,7 +26,6 @@
     generate_random_string,
     get_cluster_status,
     get_primary_unit,
-    get_server_config_credentials,
     get_unit_address,
     is_relation_joined,
     scale_application,
@@ -47,30 +46,32 @@
 logger = logging.getLogger(__name__)
 
 
-async def get_max_written_value_in_database(ops_test: OpsTest, unit: Unit) -> int:
+async def get_max_written_value_in_database(
+    ops_test: OpsTest, unit: Unit, credentials: dict
+) -> int:
     """Retrieve the max written value in the MySQL database.
 
     Args:
         ops_test: The ops test framework
         unit: The MySQL unit on which to execute queries on
+        credentials: The credentials to use to connect to the MySQL database
     """
-    server_config_credentials = await get_server_config_credentials(unit)
     unit_address = await get_unit_address(ops_test, unit.name)
 
     select_max_written_value_sql = [f"SELECT MAX(number) FROM `{DATABASE_NAME}`.`{TABLE_NAME}`;"]
 
-    output = await execute_queries_on_unit(
-        unit_address,
-        server_config_credentials["username"],
-        server_config_credentials["password"],
-        select_max_written_value_sql,
+    output = execute_queries_on_unit(
+        unit_address=unit_address,
+        username=credentials["username"],
+        password=credentials["password"],
+        queries=select_max_written_value_sql,
     )
 
     return output[0]
 
 
 def get_application_name(ops_test: OpsTest, application_name_substring: str) -> Optional[str]:
-    """Returns the name of the application witt the provided application name.
+    """Returns the name of the application with the provided application name.
 
     This enables us to retrieve the name of the deployed application in an existing model.
 
@@ -354,6 +355,7 @@ async def insert_data_into_mysql_and_validate_replication(
     ops_test: OpsTest,
     database_name: str,
     table_name: str,
+    credentials: dict,
     mysql_units: Optional[List[Unit]] = None,
     mysql_application_substring: Optional[str] = "mysql",
 ) -> str:
@@ -370,7 +372,6 @@ async def insert_data_into_mysql_and_validate_replication(
     primary = await get_primary_unit(ops_test, mysql_units[0], mysql_application_name)
 
     # insert some data into the new primary and ensure that the writes get replicated
-    server_config_credentials = await get_server_config_credentials(primary)
     primary_address = await get_unit_address(ops_test, primary.name)
 
     value = generate_random_string(255)
@@ -380,10 +381,10 @@ async def insert_data_into_mysql_and_validate_replication(
         f"INSERT INTO `{database_name}`.`{table_name}` (id) VALUES ('{value}')",
     ]
 
-    await execute_queries_on_unit(
+    execute_queries_on_unit(
         primary_address,
-        server_config_credentials["username"],
-        server_config_credentials["password"],
+        credentials["username"],
+        credentials["password"],
         insert_value_sql,
         commit=True,
     )
@@ -398,10 +399,10 @@ async def insert_data_into_mysql_and_validate_replication(
                 for unit in mysql_units:
                     unit_address = await get_unit_address(ops_test, unit.name)
 
-                    output = await execute_queries_on_unit(
+                    output = execute_queries_on_unit(
                         unit_address,
-                        server_config_credentials["username"],
-                        server_config_credentials["password"],
+                        credentials["username"],
+                        credentials["password"],
                         select_value_sql,
                     )
                     assert output[0] == value
@@ -412,20 +413,21 @@ async def insert_data_into_mysql_and_validate_replication(
 
 
 async def clean_up_database_and_table(
-    ops_test: OpsTest, database_name: str, table_name: str
+    ops_test: OpsTest, database_name: str, table_name: str, credentials: dict
 ) -> None:
     """Cleans the database and table created by insert_data_into_mysql_and_validate_replication.
 
     Args:
         ops_test: The ops test framework
         database_name: The name of the database to drop
         table_name: The name of the table to drop
+        credentials: The credentials to use to connect to the MySQL database
     """
     mysql_application_name = get_application_name(ops_test, "mysql")
 
-    mysql_unit = ops_test.model.applications[mysql_application_name].units[0]
+    assert mysql_application_name, "MySQL application not found"
 
-    server_config_credentials = await get_server_config_credentials(mysql_unit)
+    mysql_unit = ops_test.model.applications[mysql_application_name].units[0]
 
     primary = await get_primary_unit(ops_test, mysql_unit, mysql_application_name)
     primary_address = await get_unit_address(ops_test, primary.name)
@@ -435,17 +437,18 @@ async def clean_up_database_and_table(
         f"DROP DATABASE IF EXISTS `{database_name}`",
     ]
 
-    await execute_queries_on_unit(
+    execute_queries_on_unit(
         primary_address,
-        server_config_credentials["username"],
-        server_config_credentials["password"],
+        credentials["username"],
+        credentials["password"],
         clean_up_database_and_table_sql,
         commit=True,
     )
 
 
 async def ensure_all_units_continuous_writes_incrementing(
     ops_test: OpsTest,
+    credentials: dict,
     mysql_units: Optional[List[Unit]] = None,
     mysql_application_name: Optional[str] = None,
 ) -> None:
@@ -464,36 +467,24 @@ async def ensure_all_units_continuous_writes_incrementing(
 
     assert primary, "Primary unit not found"
 
-    last_max_written_value = await get_max_written_value_in_database(ops_test, primary)
-
-    select_all_continuous_writes_sql = [f"SELECT * FROM `{DATABASE_NAME}`.`{TABLE_NAME}`"]
-    server_config_credentials = await get_server_config_credentials(mysql_units[0])
+    last_max_written_value = await get_max_written_value_in_database(
+        ops_test, primary, credentials
+    )
 
-    async with ops_test.fast_forward():
+    async with ops_test.fast_forward(fast_interval="15s"):
         for attempt in Retrying(stop=stop_after_delay(15 * 60), wait=wait_fixed(10)):
             with attempt:
                 # ensure that all units are up to date (including the previous primary)
                 for unit in mysql_units:
-                    unit_address = await get_unit_address(ops_test, unit.name)
-
                     # ensure the max written value is incrementing (continuous writes is active)
-                    max_written_value = await get_max_written_value_in_database(ops_test, unit)
+                    max_written_value = await get_max_written_value_in_database(
+                        ops_test, unit, credentials
+                    )
+                    logger.info(f"{max_written_value=} on unit {unit.name}")
                     assert (
                         max_written_value > last_max_written_value
                     ), "Continuous writes not incrementing"
 
-                    # ensure that the unit contains all values up to the max written value
-                    all_written_values = await execute_queries_on_unit(
-                        unit_address,
-                        server_config_credentials["username"],
-                        server_config_credentials["password"],
-                        select_all_continuous_writes_sql,
-                    )
-                    for number in range(1, max_written_value):
-                        assert (
-                            number in all_written_values
-                        ), f"Missing {number} in database for unit {unit.name}"
-
                     last_max_written_value = max_written_value
 
 

tests/integration/high_availability/test_async_replication.py

@@ -431,7 +431,7 @@ async def get_max_written_value(first_model: Model, second_model: Model) -> list
     logger.info("Querying max value on all units")
     for unit in first_model_units + second_model_units:
         address = await get_unit_address(None, unit.name, unit.model)
-        values = await execute_queries_on_unit(
+        values = execute_queries_on_unit(
             address, credentials["username"], credentials["password"], select_max_written_value_sql
         )
         results.append(values[0])

tests/integration/high_availability/test_node_drain.py

@@ -30,7 +30,7 @@
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
 async def test_pod_eviction_and_pvc_deletion(
-    ops_test: OpsTest, highly_available_cluster, continuous_writes
+    ops_test: OpsTest, highly_available_cluster, continuous_writes, credentials
 ) -> None:
     """Test behavior when node drains - pod is evicted and pvs are rotated."""
     mysql_application_name = get_application_name(ops_test, "mysql")
@@ -42,7 +42,7 @@ async def test_pod_eviction_and_pvc_deletion(
     ), "The deployed mysql application is not fully online"
 
     logger.info("Ensuring all units have continuous writes incrementing")
-    await ensure_all_units_continuous_writes_incrementing(ops_test)
+    await ensure_all_units_continuous_writes_incrementing(ops_test, credentials=credentials)
 
     mysql_unit = ops_test.model.applications[mysql_application_name].units[0]
     primary = await get_primary_unit(ops_test, mysql_unit, mysql_application_name)
@@ -71,4 +71,4 @@ async def test_pod_eviction_and_pvc_deletion(
     ), "The deployed mysql application is not fully online after primary pod eviction"
 
     logger.info("Ensuring all units have continuous writes incrementing")
-    await ensure_all_units_continuous_writes_incrementing(ops_test)
+    await ensure_all_units_continuous_writes_incrementing(ops_test, credentials=credentials)