add test coverage

tests/integration/helpers.py

@@ -11,7 +11,7 @@
 import yaml
 from pytest_operator.plugin import OpsTest
 
-from literals import CLIENT_PORT, PEER_RELATION, SNAP_NAME
+from literals import CLIENT_PORT, SNAP_NAME
 
 logger = logging.getLogger(__name__)
 

tests/integration/test_charm.py

@@ -7,7 +7,7 @@
 import pytest
 from pytest_operator.plugin import OpsTest
 
-from literals import INTERNAL_USER
+from literals import INTERNAL_USER, PEER_RELATION
 
 from .helpers import (
     APP_NAME,
@@ -48,7 +48,9 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
     assert len(cluster_members) == NUM_UNITS
 
     # make sure data can be written to the cluster
-    password = await get_secret_by_label(ops_test, label=f"{APP_NAME}:app:") # get_user_password(ops_test, user=INTERNAL_USER, unit=leader_unit)
+    secret = await get_secret_by_label(ops_test, label=f"{PEER_RELATION}.{APP_NAME}.app")
+    password = secret.get(f"{INTERNAL_USER}-password")
+
     test_key = "test_key"
     test_value = "42"
     assert (
@@ -73,11 +75,7 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
 async def test_authentication(ops_test: OpsTest) -> None:
-    """Assert authentication is enabled by default.
-
-    Test reading and writing data without providing credentials.
-    Test updating the password of the internal admin user and make sure it can be used.
-    """
+    """Assert authentication is enabled by default."""
     model = ops_test.model_full_name
     endpoints = get_cluster_endpoints(ops_test, APP_NAME)
     leader_unit = await get_juju_leader_unit_name(ops_test, APP_NAME)
@@ -87,3 +85,50 @@ async def test_authentication(ops_test: OpsTest) -> None:
     # check that reading/writing data without credentials fails
     assert get_key(model, leader_unit, endpoints, key=test_key) != test_value
     assert put_key(model, leader_unit, endpoints, key=test_key, value=test_value) != "OK"
+
+
+@pytest.mark.runner(["self-hosted", "linux", "X64", "jammy", "large"])
+@pytest.mark.group(1)
+@pytest.mark.abort_on_fail
+async def test_update_admin_password(ops_test: OpsTest) -> None:
+    """Assert the admin password is updated when adding a user secret to the config."""
+    model = ops_test.model_full_name
+    endpoints = get_cluster_endpoints(ops_test, APP_NAME)
+    leader_unit = await get_juju_leader_unit_name(ops_test, APP_NAME)
+    test_key = "test_key"
+    test_value = "42"
+
+    # create a user secret and grant it to the application
+    secret_name = "test_secret"
+    new_password = "some-password"
+
+    secret_id = await ops_test.model.add_secret(
+        name=secret_name, data_args=[f"admin-password={new_password}"]
+    )
+    await ops_test.model.grant_secret(secret_name=secret_name, application=APP_NAME)
+
+    # update the application config to include the secret
+    await ops_test.model.applications[APP_NAME].set_config({"admin-password": secret_id})
+
+    await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
+
+    # perform read/write operation with the updated password
+    assert (
+        get_key(
+            model, leader_unit, endpoints, user=INTERNAL_USER, password=new_password, key=test_key
+        )
+        == test_value
+    )
+
+    assert (
+        put_key(
+            model,
+            leader_unit,
+            endpoints,
+            user=INTERNAL_USER,
+            password=new_password,
+            key=test_key,
+            value="43",
+        )
+        == "OK"
+    )

tests/unit/test_charm.py

@@ -138,3 +138,24 @@ def test_get_leader():
     with patch("managers.cluster.EtcdClient.get_endpoint_status", return_value=test_data):
         with ctx(ctx.on.relation_joined(relation=relation), state_in) as context:
             assert context.charm.cluster_manager.get_leader() == f"http://{test_ip}:{CLIENT_PORT}"
+
+
+def test_config_changed():
+    secret_key = "admin-password"
+    secret_value = "123"
+    secret_content = {secret_key: secret_value}
+    secret = ops.testing.Secret(tracked_content=secret_content, remote_grants=APP_NAME)
+    relation = testing.PeerRelation(id=1, endpoint=PEER_RELATION)
+
+    ctx = testing.Context(EtcdOperatorCharm)
+    state_in = testing.State(
+        secrets=[secret],
+        config={secret_key: secret.id},
+        relations={relation},
+        leader=True
+    )
+
+    with patch("subprocess.run"):
+        state_out = ctx.run(ctx.on.config_changed(), state_in)
+        secret_out = state_out.get_secret(label=f"{PEER_RELATION}.{APP_NAME}.app")
+        assert secret_out.latest_content.get(f"{INTERNAL_USER}-password") == secret_value