Integration tests

tests/integration/helpers.py

@@ -29,6 +29,7 @@
     wait_fixed,
 )
 
+APP_NAME = "app"
 CHARM_SERIES = "jammy"
 METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
 DATABASE_APP_NAME = METADATA["name"]
@@ -697,3 +698,11 @@ def wait_for_relation_removed_between(
                     break
     except RetryError:
         assert False, "Relation failed to exit after 3 minutes."
+
+
+async def get_leader_id(ops_test: OpsTest) -> int:
+    """Returns the unit number of the juju leader unit."""
+    for unit in ops_test.model.applications[APP_NAME].units:
+        if await unit.is_leader_from_status():
+            return int(unit.name.split("/")[1])
+    return -1

tests/integration/test_charm.py

@@ -2,7 +2,9 @@
 # Copyright 2021 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import json
 import logging
+from uuid import uuid4
 
 import psycopg2
 import pytest
@@ -23,10 +25,12 @@
     get_cluster_members,
     get_existing_k8s_resources,
     get_expected_k8s_resources,
+    get_leader_id,
     get_password,
     get_primary,
     get_unit_address,
     scale_application,
+    set_password,
 )
 
 logger = logging.getLogger(__name__)
@@ -401,3 +405,70 @@ async def test_storage_with_more_restrictive_permissions(ops_test: OpsTest):
 
         # This check is enough to ensure that the charm/workload is working for this specific test.
         await ops_test.model.wait_for_idle(apps=[app_name], status="active", timeout=1000)
+
+
+async def test_only_leader_can_set_while_all_can_read_password_secret(ops_test: OpsTest) -> None:
+    """Test verifies that only the leader can set a password, while all units can read it."""
+    # Setting existing password
+    leader_id = await get_leader_id(ops_test)
+    non_leaders = list(UNIT_IDS)
+    non_leaders.remove(leader_id)
+    password = "blablabla"
+    await set_password(ops_test, unit_id=non_leaders[0], username="rewind", password=password)
+    password1 = await get_password(ops_test, unit_id=leader_id, username="rewind")
+    assert password1 != password
+    await set_password(ops_test, unit_id=leader_id, username="rewind", password=password)
+    for unit_id in UNIT_IDS:
+        password2 = await get_password(ops_test, unit_id=unit_id, username="rewind")
+        assert password2 == password
+
+
+async def test_reset_and_get_password_secret_same_as_cli(ops_test: OpsTest) -> None:
+    """Test verifies that we can set and retrieve the correct password using Juju 3.x secrets."""
+    new_password = str(uuid4())
+    # Re=setting existing password
+    leader_id = await get_leader_id(ops_test)
+    result = await set_password(
+        ops_test, unit_id=leader_id, username="rewind", password=new_password
+    )
+    secret_id = result["secret-id"].split("/")[-1]
+    # Getting back the pw programmatically
+    password = await get_password(ops_test, unit_id=leader_id, username="rewind")
+
+    #
+    # No way to retrieve a secet by label for now (https://bugs.launchpad.net/juju/+bug/2037104)
+    # Therefore we take advantage of the fact, that we only have ONE single secret a this point
+    # So we take the single member of the list
+    # NOTE: This would BREAK if for instance units had secrets at the start...
+    #
+    complete_command = "list-secrets"
+    _, stdout, _ = await ops_test.juju(*complete_command.split())
+    secret_id = stdout.split("\n")[1].split(" ")[0]
+
+    # Getting back the pw from juju CLI
+    complete_command = f"show-secret {secret_id} --reveal --format=json"
+    _, stdout, _ = await ops_test.juju(*complete_command.split())
+    data = json.loads(stdout)
+    assert password == new_password
+    assert data[secret_id]["content"]["Data"]["rewind-password"] == password
+
+
+async def test_empty_password(ops_test: OpsTest) -> None:
+    """Test that the password can't be set to an empty string."""
+    leader_id = await get_leader_id(ops_test)
+    password1 = await get_password(ops_test, unit_id=leader_id, username="rewind")
+    await set_password(ops_test, unit_id=leader_id, username="rewind", password="")
+    password2 = await get_password(ops_test, unit_id=leader_id, username="rewind")
+    # The password remained unchanged
+    assert password1 == password2
+
+
+async def test_no_password_change_on_invalid_password(ops_test: OpsTest) -> None:
+    """Test that in general, there is no change when password validation fails."""
+    leader_id = await get_leader_id(ops_test)
+    password1 = await get_password(ops_test, unit_id=leader_id, username="rewind")
+    # The password has to be minimum 3 characters
+    await set_password(ops_test, unit_id=leader_id, username="rewind", password="ca" * 1000000)
+    password2 = await get_password(ops_test, unit_id=leader_id, username="rewind")
+    # The password didn't change
+    assert password1 == password2