Port test modules to Python 3.10+

Author: sinclert-canonical

tests/integration/helpers.py

@@ -8,7 +8,6 @@
 import string
 import subprocess
 import tempfile
-from typing import Dict, List, Optional, Set
 
 import juju.unit
 import yaml
@@ -133,7 +132,7 @@ async def get_primary_unit(
     return primary_unit
 
 
-async def get_server_config_credentials(unit: Unit) -> Dict:
+async def get_server_config_credentials(unit: Unit) -> dict:
     """Helper to run an action to retrieve server config credentials.
 
     Args:
@@ -145,7 +144,7 @@ async def get_server_config_credentials(unit: Unit) -> Dict:
     return await juju_.run_action(unit, "get-password", username=SERVER_CONFIG_USERNAME)
 
 
-async def fetch_credentials(unit: Unit, username: str = None) -> Dict:
+async def fetch_credentials(unit: Unit, username: str = None) -> dict:
     """Helper to run an action to fetch credentials.
 
     Args:
@@ -159,7 +158,7 @@ async def fetch_credentials(unit: Unit, username: str = None) -> Dict:
     return await juju_.run_action(unit, "get-password", username=username)
 
 
-async def rotate_credentials(unit: Unit, username: str = None, password: str = None) -> Dict:
+async def rotate_credentials(unit: Unit, username: str = None, password: str = None) -> dict:
     """Helper to run an action to rotate credentials.
 
     Args:
@@ -176,7 +175,7 @@ async def rotate_credentials(unit: Unit, username: str = None, password: str = N
         return await juju_.run_action(unit, "set-password", username=username, password=password)
 
 
-async def get_legacy_mysql_credentials(unit: Unit) -> Dict:
+async def get_legacy_mysql_credentials(unit: Unit) -> dict:
     """Helper to run an action to retrieve legacy mysql config credentials.
 
     Args:
@@ -189,7 +188,7 @@ async def get_legacy_mysql_credentials(unit: Unit) -> Dict:
 
 
 @retry(stop=stop_after_attempt(20), wait=wait_fixed(5), reraise=True)
-async def get_system_user_password(unit: Unit, user: str) -> Dict:
+async def get_system_user_password(unit: Unit, user: str) -> dict:
     """Helper to run an action to retrieve system user password.
 
     Args:
@@ -206,10 +205,10 @@ async def execute_queries_on_unit(
     unit_address: str,
     username: str,
     password: str,
-    queries: List[str],
+    queries: list[str],
     commit: bool = False,
     raw: bool = False,
-) -> List:
+) -> list:
     """Execute given MySQL queries on a unit.
 
     Args:
@@ -271,7 +270,7 @@ def is_relation_broken(ops_test: OpsTest, endpoint_one: str, endpoint_two: str)
 
 @retry(stop=stop_after_attempt(30), wait=wait_fixed(5), reraise=True)
 def is_connection_possible(
-    credentials: Dict, *, retry_if_not_possible=False, **extra_opts
+    credentials: dict, *, retry_if_not_possible=False, **extra_opts
 ) -> bool:
     """Test a connection to a MySQL server.
 
@@ -544,7 +543,7 @@ async def get_relation_data(
     return relation_data
 
 
-def get_read_only_endpoints(relation_data: list) -> Set[str]:
+def get_read_only_endpoints(relation_data: list) -> set[str]:
     """Returns the read-only-endpoints from the relation data.
 
     Args:
@@ -573,8 +572,8 @@ def get_read_only_endpoints(relation_data: list) -> Set[str]:
 
 
 async def get_leader_unit(
-    ops_test: Optional[OpsTest], app_name: str, model: Optional[Model] = None
-) -> Optional[Unit]:
+    ops_test: OpsTest | None, app_name: str, model: Model | None = None
+) -> Unit | None:
     """Get the leader unit of a given application.
 
     Args:
@@ -593,7 +592,7 @@ async def get_leader_unit(
     return leader_unit
 
 
-def get_read_only_endpoint_ips(relation_data: list) -> List[str]:
+def get_read_only_endpoint_ips(relation_data: list) -> list[str]:
     """Returns the read-only-endpoint hostnames from the relation data.
 
     Args:
@@ -640,7 +639,7 @@ async def remove_leader_unit(ops_test: OpsTest, application_name: str):
         )
 
 
-async def get_units_ip_addresses(ops_test: OpsTest, app_name: str) -> List[str]:
+async def get_units_ip_addresses(ops_test: OpsTest, app_name: str) -> list[str]:
     """Retrieves hostnames of given application units.
 
     Args:
@@ -810,12 +809,12 @@ async def unit_file_md5(ops_test: OpsTest, unit_name: str, file_path: str) -> st
         return None
 
 
-async def get_cluster_status(unit: Unit, cluster_set: Optional[bool] = False) -> Dict:
+async def get_cluster_status(unit: Unit, cluster_set: bool | None = False) -> dict:
     """Get the cluster status by running the get-cluster-status action.
 
     Args:
-        ops_test: The ops test framework
         unit: The unit on which to execute the action on
+        cluster_set: Whether to get the cluster-set instead
 
     Returns:
         A dictionary representing the cluster status
@@ -978,9 +977,7 @@ def get_unit_by_index(app_name: str, units: list, index: int):
             return unit
 
 
-async def get_status_log(
-    ops_test: OpsTest, unit_name: str, num_logs: Optional[int] = None
-) -> list:
+async def get_status_log(ops_test: OpsTest, unit_name: str, num_logs: int | None = None) -> list:
     """Get the status log for a unit.
 
     Args:

tests/integration/high_availability/high_availability_helpers.py

@@ -3,7 +3,6 @@
 
 import logging
 from pathlib import Path
-from typing import List, Optional
 
 import yaml
 from juju.unit import Unit
@@ -76,7 +75,7 @@ def get_application_name(ops_test: OpsTest, application_name_substring: str) ->
 
 
 async def ensure_n_online_mysql_members(
-    ops_test: OpsTest, number_online_members: int, mysql_units: Optional[List[Unit]] = None
+    ops_test: OpsTest, number_online_members: int, mysql_units: list[Unit] | None = None
 ) -> bool:
     """Waits until N mysql cluster members are online.
 
@@ -241,8 +240,8 @@ async def insert_data_into_mysql_and_validate_replication(
     ops_test: OpsTest,
     database_name: str,
     table_name: str,
-    mysql_application_substring: Optional[str] = "mysql",
-    mysql_units: Optional[List[Unit]] = None,
+    mysql_application_substring: str | None = "mysql",
+    mysql_units: list[Unit] | None = None,
 ) -> str:
     """Inserts data into the mysql cluster and validates its replication.
 
@@ -332,7 +331,7 @@ async def clean_up_database_and_table(
 
 
 async def ensure_all_units_continuous_writes_incrementing(
-    ops_test: OpsTest, mysql_units: Optional[List[Unit]] = None
+    ops_test: OpsTest, mysql_units: list[Unit] | None = None
 ) -> None:
     """Ensure that continuous writes is incrementing on all units.
 

tests/integration/high_availability/test_async_replication.py

@@ -8,7 +8,6 @@
 from asyncio import gather
 from pathlib import Path
 from time import sleep
-from typing import Optional
 
 import pytest
 import yaml
@@ -34,7 +33,7 @@
 
 
 @pytest.fixture(scope="module")
-def first_model(ops_test: OpsTest) -> Optional[Model]:
+def first_model(ops_test: OpsTest) -> Model | None:
     """Return the first model."""
     first_model = ops_test.model
     return first_model

tests/integration/high_availability/test_primary_switchover.py

@@ -3,7 +3,6 @@
 
 import logging
 from subprocess import run
-from typing import Optional
 
 import pytest
 from jubilant import Juju, all_active
@@ -75,7 +74,7 @@ def test_cluster_failover_after_majority_loss(juju: Juju, highly_available_clust
     assert get_primary_unit_name(juju, primary_unit) == unit_to_promote, "Failover failed"
 
 
-def get_primary_unit_name(juju: Juju, mysql_unit) -> Optional[str]:
+def get_primary_unit_name(juju: Juju, mysql_unit) -> str | None:
     """Get the current primary node of the cluster."""
     cluster_status_task = juju.run(mysql_unit, "get-cluster-status")
     assert cluster_status_task.status == "completed", "Failed to retrieve cluster status"
@@ -86,7 +85,7 @@ def get_primary_unit_name(juju: Juju, mysql_unit) -> Optional[str]:
             return label.replace("-", "/")
 
 
-def get_app_name(juju: Juju, charm_name: str) -> Optional[str]:
+def get_app_name(juju: Juju, charm_name: str) -> str | None:
     """Get the application name for the given charm."""
     status = juju.status()
     for app, value in status.apps.items():

tests/integration/relations/test_db_router.py

@@ -5,7 +5,6 @@
 import asyncio
 import logging
 from pathlib import Path
-from typing import Dict, List
 
 import pytest
 import yaml
@@ -27,7 +26,7 @@
 
 
 async def check_successful_keystone_migration(
-    ops_test: OpsTest, server_config_credentials: Dict
+    ops_test: OpsTest, server_config_credentials: dict
 ) -> None:
     """Checks that the keystone application is successfully migrated in mysql.
 
@@ -66,9 +65,9 @@ async def check_successful_keystone_migration(
 
 async def check_keystone_users_existence(
     ops_test: OpsTest,
-    server_config_credentials: Dict[str, str],
-    users_that_should_exist: List[str],
-    users_that_should_not_exist: List[str],
+    server_config_credentials: dict[str, str],
+    users_that_should_exist: list[str],
+    users_that_should_not_exist: list[str],
 ) -> None:
     """Checks that keystone users exist in the database.
 

tests/integration/relations/test_shared_db.py

@@ -4,7 +4,6 @@
 
 import logging
 from pathlib import Path
-from typing import Dict, List
 
 import pytest
 import yaml
@@ -66,7 +65,7 @@ async def deploy_and_relate_keystone_with_mysql(
 
 
 async def check_successful_keystone_migration(
-    ops_test: OpsTest, server_config_credentials: Dict
+    ops_test: OpsTest, server_config_credentials: dict
 ) -> None:
     """Checks that the keystone application is successfully migrated in mysql.
 
@@ -106,9 +105,9 @@ async def check_successful_keystone_migration(
 
 async def check_keystone_users_existence(
     ops_test: OpsTest,
-    server_config_credentials: Dict[str, str],
-    users_that_should_exist: List[str],
-    users_that_should_not_exist: List[str],
+    server_config_credentials: dict[str, str],
+    users_that_should_exist: list[str],
+    users_that_should_not_exist: list[str],
 ) -> None:
     """Checks that keystone users exist in the database.