Smoke testing. Garbage ignorance. No resources conflicts.

Author: Vladyslav Yurchenko

tests/integration/ha_tests/helpers.py

@@ -875,7 +875,6 @@ async def is_storage_exists(ops_test: OpsTest, storage_id: str) -> bool:
         storage_id,
         "--format=json",
     ]
-    print(f"command: {complete_command}")
     return_code, stdout, _ = await ops_test.juju(*complete_command)
     if return_code != 0:
         if return_code == 1:
@@ -904,7 +903,7 @@ async def create_db(ops_test: OpsTest, app: str, db: str) -> None:
 
 
 async def check_db(ops_test: OpsTest, app: str, db: str) -> bool:
-    """Returns True if database with specified name is already exists."""
+    """Returns True if database with specified name already exists."""
     unit = ops_test.model.applications[app].units[0]
     unit_address = await unit.get_public_address()
     password = await get_password(ops_test, app)
@@ -934,14 +933,31 @@ async def get_any_deatached_storage(ops_test: OpsTest) -> str:
         if (str(storage["status"]["current"]) == "detached") and (str(storage["life"] == "alive")):
             return storage_name
 
-    return None
+    raise Exception("failed to get deatached storage")
 
 
-async def check_password_auth(ops_test: OpsTest, unit_name) -> bool:
+async def check_password_auth(ops_test: OpsTest, unit_name: str) -> bool:
     """Checks if "operator" password is valid for current postgresql db."""
     stdout = await run_command_on_unit(
         ops_test,
         unit_name,
         """grep -E 'password authentication failed for user' /var/snap/charmed-postgresql/common/var/log/postgresql/postgresql*""",
     )
     return 'password authentication failed for user "operator"' not in stdout
+
+
+async def remove_unit_force(ops_test: OpsTest, unit_name: str):
+    """Removes unit with --force --no-wait."""
+    app_name = unit_name.split("/")[0]
+    complete_command = ["remove-unit", f"{unit_name}", "--force", "--no-wait", "--no-prompt"]
+    return_code, stdout, _ = await ops_test.juju(*complete_command)
+    if return_code != 0:
+        raise Exception(
+            "Expected command %s to succeed instead it failed: %s with code: ",
+            complete_command,
+            stdout,
+            return_code,
+        )
+
+    for unit in ops_test.model.applications[app_name].units:
+        assert unit != unit_name

tests/integration/ha_tests/test_smoke.py

@@ -15,67 +15,33 @@
     CHARM_SERIES,
     get_primary,
 )
+from ..juju_ import juju_major_version
 from .helpers import (
+    add_unit_with_storage,
     check_db,
     check_password_auth,
     create_db,
     get_any_deatached_storage,
     is_postgresql_ready,
     is_storage_exists,
+    remove_unit_force,
     storage_id,
 )
 
-TEST_GARBADGE_STORAGE_NAME = "test_pgdata"
+TEST_garbage_STORAGE_NAME = "test_pgdata"
 TEST_DATABASE_RELATION_NAME = "test_database"
 DUP_APPLICATION_NAME = "postgres-test-dup"
 
 logger = logging.getLogger(__name__)
 
 
-@pytest.mark.group(1)
-@pytest.mark.abort_on_fail
-async def test_app_removal(ops_test: OpsTest, charm: str):
-    """Test all recoureces is removed after application removal."""
-    # Deploy the charm.
-    async with ops_test.fast_forward():
-        await ops_test.model.deploy(
-            charm,
-            application_name=APPLICATION_NAME,
-            num_units=1,
-            series=CHARM_SERIES,
-            config={"profile": "testing"},
-        )
-
-        # Reducing the update status frequency to speed up the triggering of deferred events.
-        await ops_test.model.set_config({"update-status-hook-interval": "10s"})
-
-        await ops_test.model.wait_for_idle(apps=[APPLICATION_NAME], status="active", timeout=1500)
-        assert ops_test.model.applications[APPLICATION_NAME].units[0].workload_status == "active"
-
-        primary_name = await get_primary(
-            ops_test, ops_test.model.applications[APPLICATION_NAME].units[0].name
-        )
-        assert await is_postgresql_ready(ops_test, primary_name)
-
-        storage_id_str = storage_id(ops_test, primary_name)
-
-        # Check if storage exists after application deployed
-        assert await is_storage_exists(ops_test, storage_id_str)
-
-        await ops_test.model.remove_application(
-            APPLICATION_NAME, block_until_done=True, destroy_storage=True
-        )
-
-        # Check if storage removed after application removal
-        assert not await is_storage_exists(ops_test, storage_id_str)
-
-
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
 async def test_app_force_removal(ops_test: OpsTest, charm: str):
     """Remove unit with force while storage is alive."""
     async with ops_test.fast_forward():
         # Deploy the charm.
+        logger.info("deploying charm")
         await ops_test.model.deploy(
             charm,
             application_name=APPLICATION_NAME,
@@ -88,101 +54,181 @@ async def test_app_force_removal(ops_test: OpsTest, charm: str):
         # Reducing the update status frequency to speed up the triggering of deferred events.
         await ops_test.model.set_config({"update-status-hook-interval": "10s"})
 
+        logger.info("waiting for idle")
         await ops_test.model.wait_for_idle(apps=[APPLICATION_NAME], status="active", timeout=1500)
         assert ops_test.model.applications[APPLICATION_NAME].units[0].workload_status == "active"
 
+        logger.info("getting primary")
         primary_name = await get_primary(
             ops_test, ops_test.model.applications[APPLICATION_NAME].units[0].name
         )
-        assert await is_postgresql_ready(ops_test, primary_name)
 
+        logger.info("waiting for postgresql")
+        for attempt in Retrying(stop=stop_after_delay(15 * 3), wait=wait_fixed(3)):
+            with attempt:
+                assert await is_postgresql_ready(ops_test, primary_name)
+
+        logger.info("getting storage id")
         storage_id_str = storage_id(ops_test, primary_name)
 
         # Check if storage exists after application deployed
-        assert await is_storage_exists(ops_test, storage_id_str)
+        logger.info("werifing is storage exists")
+        for attempt in Retrying(stop=stop_after_delay(15 * 3), wait=wait_fixed(3)):
+            with attempt:
+                assert await is_storage_exists(ops_test, storage_id_str)
 
         # Create test database to check there is no resources conflicts
+        logger.info("creating db")
         await create_db(ops_test, APPLICATION_NAME, TEST_DATABASE_RELATION_NAME)
 
         # Destroy charm
-        await ops_test.model.destroy_unit(
-            primary_name, force=True, destroy_storage=False, max_wait=1500
-        )
+        logger.info("force removing charm")
+        if juju_major_version == 2:
+            await remove_unit_force(ops_test, primary_name)
+        else:
+            await ops_test.model.destroy_unit(
+                primary_name, force=True, destroy_storage=False, max_wait=1500
+            )
 
         # Storage should remain
-        assert await is_storage_exists(ops_test, storage_id_str)
+        logger.info("werifing is storage exists")
+        for attempt in Retrying(stop=stop_after_delay(15 * 3), wait=wait_fixed(3)):
+            with attempt:
+                assert await is_storage_exists(ops_test, storage_id_str)
 
 
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
 async def test_charm_garbage_ignorance(ops_test: OpsTest, charm: str):
     """Test charm deploy in dirty environment with garbage storage."""
     async with ops_test.fast_forward():
-        garbadge_storage = None
+        logger.info("checking garbage storage")
+        garbage_storage = None
         for attempt in Retrying(stop=stop_after_delay(30 * 3), wait=wait_fixed(3)):
             with attempt:
-                garbadge_storage = await get_any_deatached_storage(ops_test)
-                assert garbadge_storage is not None
-
-        assert garbadge_storage is not None
-
-        await ops_test.model.applications[APPLICATION_NAME].add_unit(
-            1, attach_storage=[tag.storage(garbadge_storage)]
-        )
+                garbage_storage = await get_any_deatached_storage(ops_test)
 
         # Reducing the update status frequency to speed up the triggering of deferred events.
         await ops_test.model.set_config({"update-status-hook-interval": "10s"})
 
-        await ops_test.model.wait_for_idle(apps=[APPLICATION_NAME], status="active", timeout=1500)
-        assert ops_test.model.applications[APPLICATION_NAME].units[0].workload_status == "active"
+        logger.info("add unit with attached storage")
+        await add_unit_with_storage(ops_test, APPLICATION_NAME, garbage_storage)
 
+        logger.info("getting primary")
         primary_name = await get_primary(
             ops_test, ops_test.model.applications[APPLICATION_NAME].units[0].name
         )
-        assert await is_postgresql_ready(ops_test, primary_name)
 
+        logger.info("waiting for postgresql")
+        for attempt in Retrying(stop=stop_after_delay(15 * 3), wait=wait_fixed(3)):
+            with attempt:
+                assert await is_postgresql_ready(ops_test, primary_name)
+
+        logger.info("getting storage id")
         storage_id_str = storage_id(ops_test, primary_name)
 
+        assert storage_id_str == garbage_storage
+
         # Check if storage exists after application deployed
-        assert await is_storage_exists(ops_test, storage_id_str)
+        logger.info("werifing is storage exists")
+        for attempt in Retrying(stop=stop_after_delay(15 * 3), wait=wait_fixed(3)):
+            with attempt:
+                assert await is_storage_exists(ops_test, storage_id_str)
 
         # Check that test database is not exists for new unit
+        logger.info("checking db")
         assert not await check_db(ops_test, APPLICATION_NAME, TEST_DATABASE_RELATION_NAME)
 
-        await ops_test.model.destroy_unit(primary_name, destroy_storage=False, max_wait=1500)
+        logger.info("removing charm")
+        await ops_test.model.destroy_unit(primary_name)
 
 
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
-async def test_app_recoures_conflicts(ops_test: OpsTest, charm: str):
+@pytest.mark.skipif(juju_major_version < 3, reason="Requires juju 3 or higher")
+async def test_app_resources_conflicts_v3(ops_test: OpsTest, charm: str):
     """Test application deploy in dirty environment with garbage storage from another application."""
     async with ops_test.fast_forward():
-        garbadge_storage = None
+        logger.info("checking garbage storage")
+        garbage_storage = None
         for attempt in Retrying(stop=stop_after_delay(30 * 3), wait=wait_fixed(3)):
             with attempt:
-                garbadge_storage = await get_any_deatached_storage(ops_test)
-                assert garbadge_storage is not None
+                garbage_storage = await get_any_deatached_storage(ops_test)
+
+        logger.info("deploying duplicate application with attached storage")
+        await ops_test.model.deploy(
+            charm,
+            application_name=DUP_APPLICATION_NAME,
+            num_units=1,
+            series=CHARM_SERIES,
+            attach_storage=[tag.storage(garbage_storage)],
+            config={"profile": "testing"},
+        )
+
+        # Reducing the update status frequency to speed up the triggering of deferred events.
+        await ops_test.model.set_config({"update-status-hook-interval": "10s"})
+
+        logger.info("waiting for duplicate application to be blocked")
+        try:
+            await ops_test.model.wait_for_idle(
+                apps=[DUP_APPLICATION_NAME], timeout=1000, status="blocked"
+            )
+        except TimeoutError:
+            logger.info("Application is not in blocked state. Checking logs...")
+
+        # Since application have postgresql db in storage from external application it should not be able to connect due to new password
+        logger.info("checking operator password auth")
+        assert not await check_password_auth(
+            ops_test, ops_test.model.applications[DUP_APPLICATION_NAME].units[0].name
+        )
+
 
-        assert garbadge_storage is not None
+@pytest.mark.group(1)
+@pytest.mark.abort_on_fail
+@pytest.mark.skipif(juju_major_version != 2, reason="Requires juju 2")
+async def test_app_resources_conflicts_v2(ops_test: OpsTest, charm: str):
+    """Test application deploy in dirty environment with garbage storage from another application."""
+    async with ops_test.fast_forward():
+        logger.info("checking garbage storage")
+        garbage_storage = None
+        for attempt in Retrying(stop=stop_after_delay(30 * 3), wait=wait_fixed(3)):
+            with attempt:
+                garbage_storage = await get_any_deatached_storage(ops_test)
 
         # Deploy duplicaate charm
+        logger.info("deploying duplicate application")
         await ops_test.model.deploy(
             charm,
             application_name=DUP_APPLICATION_NAME,
             num_units=1,
             series=CHARM_SERIES,
             config={"profile": "testing"},
-            attach_storage=[tag.storage(garbadge_storage)],
         )
 
+        # Reducing the update status frequency to speed up the triggering of deferred events.
+        await ops_test.model.set_config({"update-status-hook-interval": "10s"})
+
+        logger.info("force removing charm")
+        await remove_unit_force(
+            ops_test, ops_test.model.applications[DUP_APPLICATION_NAME].units[0].name
+        )
+
+        # Add unit with garbage storage
+        logger.info("adding charm with attached storage")
+        add_unit_cmd = f"add-unit {DUP_APPLICATION_NAME} --model={ops_test.model.info.name} --attach-storage={garbage_storage}".split()
+        return_code, _, _ = await ops_test.juju(*add_unit_cmd)
+        assert return_code == 0, "Failed to add unit with storage"
+
+        logger.info("waiting for duplicate application to be blocked")
         try:
             await ops_test.model.wait_for_idle(
-                apps=[DUP_APPLICATION_NAME], timeout=500, status="blocked"
+                apps=[DUP_APPLICATION_NAME], timeout=1000, status="blocked"
             )
         except TimeoutError:
             logger.info("Application is not in blocked state. Checking logs...")
 
         # Since application have postgresql db in storage from external application it should not be able to connect due to new password
+        logger.info("checking operator password auth")
         assert not await check_password_auth(
             ops_test, ops_test.model.applications[DUP_APPLICATION_NAME].units[0].name
         )