[DPE-8314] Upgrade Jubilant tests

pyproject.toml

@@ -65,7 +65,7 @@ pyyaml = "^6.0"
 urllib3 = "^2.0.0"
 allure-pytest = "^2.13.2"
 allure-pytest-default-results = "^0.1.2"
-jubilant = "^1.0.1"
+jubilant-backports = "^1.0.0a1"
 
 [tool.coverage.run]
 branch = true

tests/integration/conftest.py

@@ -3,7 +3,7 @@
 import os
 import uuid
 
-import jubilant
+import jubilant_backports
 import pytest
 
 from . import architecture
@@ -57,11 +57,11 @@ def juju(request: pytest.FixtureRequest):
     keep_models = bool(request.config.getoption("--keep-models"))
 
     if model:
-        juju = jubilant.Juju(model=model)  # type: ignore
+        juju = jubilant_backports.Juju(model=model)  # type: ignore
         yield juju
         log = juju.debug_log(limit=1000)
     else:
-        with jubilant.temp_model(keep=keep_models) as juju:
+        with jubilant_backports.temp_model(keep=keep_models) as juju:
             yield juju
             log = juju.debug_log(limit=1000)
 

tests/integration/high_availability/high_availability_helpers_new.py

@@ -2,11 +2,13 @@
 # Copyright 2025 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import json
+import subprocess
 from collections.abc import Callable
 
-import jubilant
-from jubilant import Juju
-from jubilant.statustypes import Status, UnitStatus
+import jubilant_backports
+from jubilant_backports import Juju
+from jubilant_backports.statustypes import Status, UnitStatus
 from tenacity import Retrying, stop_after_delay, wait_fixed
 
 from constants import SERVER_CONFIG_USERNAME
@@ -19,19 +21,22 @@
 JujuAppsStatusFn = Callable[[Status, str], bool]
 
 
-async def check_mysql_units_writes_increment(juju: Juju, app_name: str) -> None:
+async def check_mysql_units_writes_increment(
+    juju: Juju, app_name: str, app_units: list[str] | None = None
+) -> None:
     """Ensure that continuous writes is incrementing on all units.
 
     Also, ensure that all continuous writes up to the max written value is available
     on all units (ensure that no committed data is lost).
     """
-    mysql_app_units = get_app_units(juju, app_name)
-    mysql_app_primary = get_mysql_primary_unit(juju, app_name)
+    if not app_units:
+        app_units = get_app_units(juju, app_name)
 
-    app_max_value = await get_mysql_max_written_value(juju, app_name, mysql_app_primary)
+    app_primary = get_mysql_primary_unit(juju, app_name)
+    app_max_value = await get_mysql_max_written_value(juju, app_name, app_primary)
 
     juju.model_config({"update-status-hook-interval": "15s"})
-    for unit_name in mysql_app_units:
+    for unit_name in app_units:
         for attempt in Retrying(
             reraise=True,
             stop=stop_after_delay(5 * MINUTE_SECS),
@@ -72,6 +77,17 @@ def get_app_units(juju: Juju, app_name: str) -> dict[str, UnitStatus]:
     return app_status.units
 
 
+def get_unit_by_number(juju: Juju, app_name: str, unit_number: int) -> str:
+    """Get unit by number."""
+    model_status = juju.status()
+    app_status = model_status.apps[app_name]
+    for name in app_status.units:
+        if name == f"{app_name}/{unit_number}":
+            return name
+
+    raise Exception("No application unit found")
+
+
 def get_unit_ip(juju: Juju, app_name: str, unit_name: str) -> str:
     """Get the application unit IP."""
     model_status = juju.status()
@@ -83,21 +99,75 @@ def get_unit_ip(juju: Juju, app_name: str, unit_name: str) -> str:
     raise Exception("No application unit found")
 
 
-def get_mysql_cluster_status(juju: Juju, unit: str, cluster_set: bool | None = False) -> dict:
+def get_unit_info(juju: Juju, unit_name: str) -> dict:
+    """Return a dictionary with the show-unit data."""
+    output = subprocess.check_output(
+        ["juju", "show-unit", f"--model={juju.model}", "--format=json", unit_name],
+        text=True,
+    )
+
+    return json.loads(output)
+
+
+def get_unit_status_log(juju: Juju, unit_name: str, log_lines: int = 0) -> list[dict]:
+    """Get the status log for a unit.
+
+    Args:
+        juju: The juju instance to use.
+        unit_name: The name of the unit to retrieve the status log for
+        log_lines: The number of status logs to retrieve (optional)
+    """
+    # fmt: off
+    output = subprocess.check_output(
+        ["juju", "show-status-log", f"--model={juju.model}", "--format=json", unit_name, "-n", f"{log_lines}"],
+        text=True,
+    )
+
+    return json.loads(output)
+
+
+def get_relation_data(juju: Juju, app_name: str, rel_name: str) -> list[dict]:
+    """Returns a list that contains the relation-data.
+
+    Args:
+        juju: The juju instance to use.
+        app_name: The name of the application
+        rel_name: name of the relation to get connection data from
+
+    Returns:
+        A list that contains the relation-data
+    """
+    app_leader = get_app_leader(juju, app_name)
+    app_leader_info = get_unit_info(juju, app_leader)
+    if not app_leader_info:
+        raise ValueError(f"No unit info could be grabbed for unit {app_leader}")
+
+    relation_data = [
+        value
+        for value in app_leader_info[app_leader]["relation-info"]
+        if value["endpoint"] == rel_name
+    ]
+    if not relation_data:
+        raise ValueError(f"No relation data could be grabbed for relation {rel_name}")
+
+    return relation_data
+
+
+def get_mysql_cluster_status(juju: Juju, unit: str, cluster_set: bool = False) -> dict:
     """Get the cluster status by running the get-cluster-status action.
 
     Args:
         juju: The juju instance to use.
         unit: The unit on which to execute the action on
-        cluster_set: Whether to get the cluster-set instead
+        cluster_set: Whether to get the cluster-set instead (optional)
 
     Returns:
         A dictionary representing the cluster status
     """
     task = juju.run(
         unit=unit,
         action="get-cluster-status",
-        params={"cluster-set": bool(cluster_set)},
+        params={"cluster-set": cluster_set},
         wait=5 * MINUTE_SECS,
     )
     task.raise_on_failure()
@@ -185,6 +255,20 @@ def wait_for_apps_status(jubilant_status_func: JujuAppsStatusFn, *apps: str) ->
         Juju model status function.
     """
     return lambda status: all((
-        jubilant.all_agents_idle(status, *apps),
+        jubilant_backports.all_agents_idle(status, *apps),
         jubilant_status_func(status, *apps),
     ))
+
+
+def wait_for_unit_status(app_name: str, unit_name: str, unit_status: str) -> JujuModelStatusFn:
+    """Returns whether a Juju unit to have a specific status."""
+    return lambda status: (
+        status.apps[app_name].units[unit_name].workload_status.current == unit_status
+    )
+
+
+def wait_for_unit_message(app_name: str, unit_name: str, unit_message: str) -> JujuModelStatusFn:
+    """Returns whether a Juju unit to have a specific message."""
+    return lambda status: (
+        status.apps[app_name].units[unit_name].workload_status.message == unit_message
+    )

tests/integration/high_availability/test_async_replication.py

@@ -6,9 +6,9 @@
 import time
 from collections.abc import Generator
 
-import jubilant
+import jubilant_backports
 import pytest
-from jubilant import Juju
+from jubilant_backports import Juju
 
 from .. import architecture
 from ..markers import juju3
@@ -98,11 +98,11 @@ def test_build_and_deploy(first_model: str, second_model: str, charm: str) -> No
 
     logging.info("Waiting for the applications to settle")
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_1),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_1),
         timeout=10 * MINUTE_SECS,
     )
     model_2.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_2),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_2),
         timeout=10 * MINUTE_SECS,
     )
 
@@ -127,11 +127,11 @@ def test_async_relate(first_model: str, second_model: str) -> None:
 
     logging.info("Waiting for the applications to settle")
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.any_blocked, MYSQL_APP_1),
+        ready=wait_for_apps_status(jubilant_backports.any_blocked, MYSQL_APP_1),
         timeout=5 * MINUTE_SECS,
     )
     model_2.wait(
-        ready=wait_for_apps_status(jubilant.any_waiting, MYSQL_APP_2),
+        ready=wait_for_apps_status(jubilant_backports.any_waiting, MYSQL_APP_2),
         timeout=5 * MINUTE_SECS,
     )
 
@@ -170,7 +170,7 @@ def test_deploy_router_and_app(first_model: str) -> None:
     )
 
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_TEST_APP_NAME),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_TEST_APP_NAME),
         timeout=10 * MINUTE_SECS,
     )
 
@@ -192,11 +192,11 @@ def test_create_replication(first_model: str, second_model: str) -> None:
 
     logging.info("Waiting for the applications to settle")
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_1),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_1),
         timeout=5 * MINUTE_SECS,
     )
     model_2.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_2),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_2),
         timeout=5 * MINUTE_SECS,
     )
 
@@ -326,11 +326,11 @@ async def test_unrelate_and_relate(first_model: str, second_model: str, continuo
 
     logging.info("Waiting for the applications to settle")
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_1),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_1),
         timeout=10 * MINUTE_SECS,
     )
     model_2.wait(
-        ready=wait_for_apps_status(jubilant.all_blocked, MYSQL_APP_2),
+        ready=wait_for_apps_status(jubilant_backports.all_blocked, MYSQL_APP_2),
         timeout=10 * MINUTE_SECS,
     )
 
@@ -340,7 +340,7 @@ async def test_unrelate_and_relate(first_model: str, second_model: str, continuo
         f"{MYSQL_APP_2}:replication",
     )
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.any_blocked, MYSQL_APP_1),
+        ready=wait_for_apps_status(jubilant_backports.any_blocked, MYSQL_APP_1),
         timeout=5 * MINUTE_SECS,
     )
 
@@ -354,11 +354,11 @@ async def test_unrelate_and_relate(first_model: str, second_model: str, continuo
 
     logging.info("Waiting for the applications to settle")
     model_1.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_1),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_1),
         timeout=10 * MINUTE_SECS,
     )
     model_2.wait(
-        ready=wait_for_apps_status(jubilant.all_active, MYSQL_APP_2),
+        ready=wait_for_apps_status(jubilant_backports.all_active, MYSQL_APP_2),
         timeout=10 * MINUTE_SECS,
     )