handle error: storage belongs to different cluster

Author: BalabaDmitri

tests/integration/ha_tests/test_restore_cluster.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
+import asyncio
 import logging
 
 import pytest
@@ -37,24 +38,23 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
     charm = await ops_test.build_charm(".")
     async with ops_test.fast_forward():
         # Deploy the first cluster with reusable storage
-        await ops_test.model.deploy(
-            charm,
-            application_name=FIRST_APPLICATION,
-            num_units=3,
-            series=CHARM_SERIES,
-            storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
-            config={"profile": "testing"},
+        await asyncio.gather(
+            ops_test.model.deploy(
+                charm,
+                application_name=FIRST_APPLICATION,
+                num_units=3,
+                series=CHARM_SERIES,
+                storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
+                config={"profile": "testing"},
+            ),
+            ops_test.model.deploy(
+                charm,
+                application_name=SECOND_APPLICATION,
+                num_units=1,
+                series=CHARM_SERIES,
+                config={"profile": "testing"},
+            )
         )
-
-        # Deploy the second cluster
-        await ops_test.model.deploy(
-            charm,
-            application_name=SECOND_APPLICATION,
-            num_units=1,
-            series=CHARM_SERIES,
-            config={"profile": "testing"},
-        )
-
         await ops_test.model.wait_for_idle(status="active", timeout=1500)
 
         # TODO have a better way to bootstrap clusters with existing storage

tests/integration/ha_tests/test_self_healing.py

@@ -85,6 +85,14 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
                 storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
                 config={"profile": "testing"},
             )
+            await ops_test.model.deploy(
+                charm,
+                num_units=1,
+                application_name=SECOND_APPLICATION,
+                series=CHARM_SERIES,
+                storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
+                config={"profile": "testing"},
+            )
     # Deploy the continuous writes application charm if it wasn't already deployed.
     if not await app_name(ops_test, APPLICATION_NAME):
         wait_for_apps = True
@@ -553,19 +561,8 @@ async def test_network_cut_without_ip_change(
 
 
 @pytest.mark.group(1)
-async def test_deploy_zero_units(ops_test: OpsTest):
+async def test_deploy_zero_units(ops_test: OpsTest, charm):
     """Scale the database to zero units and scale up again."""
-    async with ops_test.fast_forward():
-        await ops_test.model.deploy(
-            APP_NAME,
-            num_units=1,
-            application_name=SECOND_APPLICATION,
-            series=CHARM_SERIES,
-            storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
-            config={"profile": "testing"},
-        )
-        await ops_test.model.wait_for_idle(status="active", timeout=1500)
-
     app = await app_name(ops_test)
     dbname = f"{APPLICATION_NAME.replace('-', '_')}_first_database"
     connection_string, _ = await get_db_connection(ops_test, dbname=dbname)