WIP: Use integration test helpers from data-platform-helpers

pyproject.toml

@@ -67,7 +67,7 @@ pyyaml = "^6.0"
 urllib3 = "^1.26.16"
 allure-pytest = "^2.13.2"
 allure-pytest-collection-report = {git = "https://github.com/canonical/data-platform-workflows", tag = "v22.0.0", subdirectory = "python/pytest_plugins/allure_pytest_collection_report"}
-
+data-platform-helpers = { git = "https://github.com/canonical/data-platform-helpers.git", branch = "test/add-network-cut-restore-vm-2" }
 
 [tool.coverage.run]
 branch = true

tests/integration/helpers.py

@@ -12,6 +12,7 @@
 
 import juju.unit
 import yaml
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from juju.model import Model
 from juju.unit import Unit
 from mysql.connector.errors import (
@@ -379,28 +380,6 @@ async def is_unit_in_cluster(unit_name: str, action_unit: juju.unit.Unit) -> boo
     raise TimeoutError
 
 
-def cut_network_from_unit(machine_name: str) -> None:
-    """Cut network from a lxc container.
-
-    Args:
-        machine_name: lxc container hostname
-    """
-    # apply a mask (device type `none`)
-    cut_network_command = f"lxc config device add {machine_name} eth0 none"
-    subprocess.check_call(cut_network_command.split())
-
-
-def restore_network_for_unit(machine_name: str) -> None:
-    """Restore network from a lxc container.
-
-    Args:
-        machine_name: lxc container hostname
-    """
-    # remove mask from eth0
-    restore_network_command = f"lxc config device remove {machine_name} eth0"
-    subprocess.check_call(restore_network_command.split())
-
-
 async def unit_hostname(ops_test: OpsTest, unit_name: str) -> str:
     """Get hostname for a unit.
 
@@ -414,25 +393,6 @@ async def unit_hostname(ops_test: OpsTest, unit_name: str) -> str:
     return raw_hostname.strip()
 
 
-@retry(stop=stop_after_attempt(20), wait=wait_fixed(15))
-async def wait_network_restore(ops_test: OpsTest, unit_name: str) -> None:
-    """Wait until network is restored.
-
-    Args:
-        ops_test: The ops test object passed into every test case
-        unit_name: The name of the unit
-        old_ip: old registered IP address
-    """
-    return_code, stdout, _ = await ops_test.juju("ssh", unit_name, "ip", "a")
-    if return_code != 0:
-        raise Exception
-
-    juju_unit_ip = await get_unit_ip(ops_test, unit_name)
-
-    if juju_unit_ip in stdout:
-        raise Exception
-
-
 async def graceful_stop_server(ops_test: OpsTest, unit_name: str) -> None:
     """Gracefully stop server.
 
@@ -496,22 +456,6 @@ async def get_primary_unit_wrapper(ops_test: OpsTest, app_name: str, unit_exclud
     raise ValueError("Primary unit found cannot be retrieved")
 
 
-async def get_unit_ip(ops_test: OpsTest, unit_name: str) -> str:
-    """Wrapper for getting unit ip.
-
-    Args:
-        ops_test: The ops test object passed into every test case
-        unit_name: The name of the unit to get the address
-    Returns:
-        The (str) ip of the unit
-    """
-    app_name = unit_name.split("/")[0]
-    unit_num = unit_name.split("/")[1]
-    status = await ops_test.model.get_status()  # noqa: F821
-    address = status["applications"][app_name]["units"][f"{app_name}/{unit_num}"]["public-address"]
-    return address
-
-
 async def get_relation_data(
     ops_test: OpsTest,
     application_name: str,
@@ -679,38 +623,6 @@ async def check_read_only_endpoints(ops_test: OpsTest, app_name: str, relation_n
         assert read_endpoint_ip in app_ips
 
 
-async def get_controller_machine(ops_test: OpsTest) -> str:
-    """Return controller machine hostname.
-
-    Args:
-        ops_test: The ops test framework instance
-    Returns:
-        Controller hostname (str)
-    """
-    _, raw_controller, _ = await ops_test.juju("show-controller")
-
-    controller = yaml.safe_load(raw_controller.strip())
-
-    return [
-        machine.get("instance-id")
-        for machine in controller[ops_test.controller_name]["controller-machines"].values()
-    ][0]
-
-
-def is_machine_reachable_from(origin_machine: str, target_machine: str) -> bool:
-    """Test network reachability between hosts.
-
-    Args:
-        origin_machine: hostname of the machine to test connection from
-        target_machine: hostname of the machine to test connection to
-    """
-    try:
-        subprocess.check_call(f"lxc exec {origin_machine} -- ping -c 3 {target_machine}".split())
-        return True
-    except subprocess.CalledProcessError:
-        return False
-
-
 async def write_random_chars_to_test_table(ops_test: OpsTest, primary_unit: Unit) -> str:
     """Writes to common test table.
 

tests/integration/high_availability/high_availability_helpers.py

@@ -6,6 +6,7 @@
 from typing import List, Optional
 
 import yaml
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from juju.unit import Unit
 from pytest_operator.plugin import OpsTest
 from tenacity import RetryError, Retrying, stop_after_delay, wait_fixed
@@ -16,7 +17,6 @@
     get_cluster_status,
     get_primary_unit_wrapper,
     get_server_config_credentials,
-    get_unit_ip,
     is_relation_joined,
     scale_application,
 )

tests/integration/high_availability/test_self_healing.py

@@ -8,27 +8,29 @@
 
 import pytest
 import yaml
+from data_platform_helpers.tests_helpers.vms.ha_helpers import (
+    cut_network_from_unit_with_ip_change,
+    get_controller_machine,
+    get_unit_ip,
+    is_machine_reachable_from,
+    restore_network_for_unit_with_ip_change,
+    wait_network_restore_with_ip_change,
+)
 from pytest_operator.plugin import OpsTest
 from tenacity import RetryError, Retrying, stop_after_attempt, wait_fixed
 
 from constants import CLUSTER_ADMIN_USERNAME, SERVER_CONFIG_USERNAME
 
 from ..helpers import (
-    cut_network_from_unit,
     execute_queries_on_unit,
-    get_controller_machine,
     get_primary_unit_wrapper,
     get_process_pid,
     get_system_user_password,
-    get_unit_ip,
     graceful_stop_server,
     is_connection_possible,
-    is_machine_reachable_from,
     is_unit_in_cluster,
-    restore_network_for_unit,
     start_server,
     unit_hostname,
-    wait_network_restore,
     write_random_chars_to_test_table,
 )
 from .high_availability_helpers import (
@@ -167,7 +169,7 @@ async def test_network_cut(ops_test: OpsTest, highly_available_cluster, continuo
     ), f"❌ Connection to host {primary_unit_ip} is not possible"
 
     logger.info(f"Cutting network for {primary_hostname}")
-    cut_network_from_unit(primary_hostname)
+    cut_network_from_unit_with_ip_change(primary_hostname)
 
     # verify machine is not reachable from peer units
     for unit in set(all_units) - {primary_unit}:
@@ -186,10 +188,10 @@ async def test_network_cut(ops_test: OpsTest, highly_available_cluster, continuo
     assert not is_connection_possible(config), "❌ Connection is possible after network cut"
 
     logger.info(f"Restoring network for {primary_hostname}")
-    restore_network_for_unit(primary_hostname)
+    restore_network_for_unit_with_ip_change(primary_hostname)
 
     # wait until network is reestablished for the unit
-    await wait_network_restore(ops_test, primary_unit.name)
+    await wait_network_restore_with_ip_change(ops_test, primary_unit.name, primary_unit_ip)
 
     # ensure continuous writes still incrementing for all units
     async with ops_test.fast_forward():

tests/integration/test_backup_aws.py

@@ -8,14 +8,14 @@
 
 import boto3
 import pytest
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from pytest_operator.plugin import OpsTest
 
 from . import juju_
 from .helpers import (
     execute_queries_on_unit,
     get_primary_unit_wrapper,
     get_server_config_credentials,
-    get_unit_ip,
     rotate_credentials,
     scale_application,
 )

tests/integration/test_backup_ceph.py

@@ -9,14 +9,14 @@
 import boto3
 import pytest
 import pytest_microceph
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from pytest_operator.plugin import OpsTest
 
 from . import juju_
 from .helpers import (
     execute_queries_on_unit,
     get_primary_unit_wrapper,
     get_server_config_credentials,
-    get_unit_ip,
     rotate_credentials,
     scale_application,
 )

tests/integration/test_backup_gcp.py

@@ -8,14 +8,14 @@
 
 import boto3
 import pytest
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from pytest_operator.plugin import OpsTest
 
 from . import juju_
 from .helpers import (
     execute_queries_on_unit,
     get_primary_unit_wrapper,
     get_server_config_credentials,
-    get_unit_ip,
     rotate_credentials,
     scale_application,
 )

tests/integration/test_saturate_max_connections.py

@@ -4,11 +4,11 @@
 import logging
 
 import pytest
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from mysql.connector.errors import OperationalError
 from pytest_operator.plugin import OpsTest
 
 from .connector import create_db_connections
-from .helpers import get_unit_ip
 from .juju_ import run_action
 
 logger = logging.getLogger(__name__)

tests/integration/test_tls.py

@@ -7,6 +7,7 @@
 
 import pytest
 import yaml
+from data_platform_helpers.tests_helpers.vms.ha_helpers import get_unit_ip
 from pytest_operator.plugin import OpsTest
 
 from constants import CLUSTER_ADMIN_USERNAME, TLS_SSL_CERT_FILE
@@ -16,7 +17,6 @@
     app_name,
     get_system_user_password,
     get_tls_ca,
-    get_unit_ip,
     is_connection_possible,
     scale_application,
     unit_file_md5,