fixed unit tests and tweaked integration tests

tests/integration/helpers.py

@@ -27,13 +27,17 @@ def put_key(
     value: str,
     user: str | None = None,
     password: str | None = None,
+    tls_enabled: bool = False,
 ) -> str:
     """Write data to etcd using `etcdctl` via `juju ssh`."""
     etcd_command = f"{SNAP_NAME}.etcdctl put {key} {value} --endpoints={endpoints}"
     if user:
         etcd_command = f"{etcd_command} --user={user}"
     if password:
         etcd_command = f"{etcd_command} --password={password}"
+    if tls_enabled:
+        etcd_command = f"{etcd_command} --cacert /var/snap/charmed-etcd/common/tls/client_ca.pem --cert /var/snap/charmed-etcd/common/tls/client.pem --key /var/snap/charmed-etcd/common/tls/client.key"
+
     juju_command = f"juju ssh --model={model} {unit} {etcd_command}"
 
     return subprocess.getoutput(juju_command).split("\n")[0]

tests/integration/test_tls.py

@@ -31,7 +31,7 @@
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy", "large"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
-async def test_build_and_deploy(ops_test: OpsTest) -> None:
+async def test_build_and_deploy_with_tls(ops_test: OpsTest) -> None:
     """Build the charm-under-test and deploy it with three units.
 
     The initial cluster should be formed and accessible.
@@ -46,14 +46,21 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
     assert model is not None
     # Deploy the charm and wait for active/idle status
     await ops_test.model.deploy(etcd_charm, num_units=NUM_UNITS)
+
+    # enable TLS and check if the cluster is still accessible
+    await ops_test.model.integrate(f"{APP_NAME}:peer-certificates", TLS_NAME)
+    await ops_test.model.integrate(f"{APP_NAME}:client-certificates", TLS_NAME)
     await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
 
     # check if all units have been added to the cluster
-    endpoints = get_cluster_endpoints(ops_test, APP_NAME)
+    endpoints = get_cluster_endpoints(ops_test, APP_NAME, tls_enabled=True)
     leader_unit = await get_juju_leader_unit_name(ops_test, APP_NAME)
 
-    cluster_members = get_cluster_members(model, leader_unit, endpoints)
+    cluster_members = get_cluster_members(model, leader_unit, endpoints, tls_enabled=True)
     assert len(cluster_members) == NUM_UNITS
+    for cluster_member in cluster_members:
+        assert cluster_member["clientURLs"][0].startswith("https")
+        assert cluster_member["peerURLs"][0].startswith("https")
 
     # make sure data can be written to the cluster
     secret = await get_secret_by_label(ops_test, label=f"{PEER_RELATION}.{APP_NAME}.app")
@@ -68,37 +75,47 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
             password=password,
             key=TEST_KEY,
             value=TEST_VALUE,
+            tls_enabled=True,
         )
         == "OK"
     )
     assert (
-        get_key(model, leader_unit, endpoints, user=INTERNAL_USER, password=password, key=TEST_KEY)
+        get_key(
+            model,
+            leader_unit,
+            endpoints,
+            user=INTERNAL_USER,
+            password=password,
+            key=TEST_KEY,
+            tls_enabled=True,
+        )
         == TEST_VALUE
     )
 
 
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy", "large"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
-async def test_turning_on_tls(ops_test: OpsTest) -> None:
+async def test_turning_off_tls(ops_test: OpsTest) -> None:
     assert ops_test.model
     model = ops_test.model_full_name
     assert model is not None
 
     # enable TLS and check if the cluster is still accessible
-    await ops_test.model.integrate(f"{APP_NAME}:peer-certificates", TLS_NAME)
-    await ops_test.model.integrate(f"{APP_NAME}:client-certificates", TLS_NAME)
+    etcd_app: Application = ops_test.model.applications[APP_NAME]  # type: ignore
+    await etcd_app.remove_relation("peer-certificates", f"{TLS_NAME}:certificates")
+    await etcd_app.remove_relation("client-certificates", f"{TLS_NAME}:certificates")
 
     await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
 
-    endpoints = get_cluster_endpoints(ops_test, APP_NAME, tls_enabled=True)
+    endpoints = get_cluster_endpoints(ops_test, APP_NAME)
     leader_unit = await get_juju_leader_unit_name(ops_test, APP_NAME)
-    cluster_members = get_cluster_members(model, leader_unit, endpoints, tls_enabled=True)
+    cluster_members = get_cluster_members(model, leader_unit, endpoints)
     assert len(cluster_members) == NUM_UNITS
 
     for cluster_member in cluster_members:
-        assert cluster_member["clientURLs"][0].startswith("https")
-        assert cluster_member["peerURLs"][0].startswith("https")
+        assert cluster_member["clientURLs"][0].startswith("http://")
+        assert cluster_member["peerURLs"][0].startswith("http://")
 
     secret = await get_secret_by_label(ops_test, label=f"{PEER_RELATION}.{APP_NAME}.app")
     password = secret.get(f"{INTERNAL_USER}-password")
@@ -110,7 +127,6 @@ async def test_turning_on_tls(ops_test: OpsTest) -> None:
             user=INTERNAL_USER,
             password=password,
             key=TEST_KEY,
-            tls_enabled=True,
         )
         == TEST_VALUE
     )
@@ -119,26 +135,25 @@ async def test_turning_on_tls(ops_test: OpsTest) -> None:
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy", "large"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
-async def test_turning_off_tls(ops_test: OpsTest) -> None:
+async def test_turning_on_tls(ops_test: OpsTest) -> None:
     assert ops_test.model
     model = ops_test.model_full_name
     assert model is not None
 
     # enable TLS and check if the cluster is still accessible
-    etcd_app: Application = ops_test.model.applications[APP_NAME]  # type: ignore
-    await etcd_app.remove_relation("peer-certificates", f"{TLS_NAME}:certificates")
-    await etcd_app.remove_relation("client-certificates", f"{TLS_NAME}:certificates")
+    await ops_test.model.integrate(f"{APP_NAME}:peer-certificates", TLS_NAME)
+    await ops_test.model.integrate(f"{APP_NAME}:client-certificates", TLS_NAME)
 
     await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
 
-    endpoints = get_cluster_endpoints(ops_test, APP_NAME)
+    endpoints = get_cluster_endpoints(ops_test, APP_NAME, tls_enabled=True)
     leader_unit = await get_juju_leader_unit_name(ops_test, APP_NAME)
-    cluster_members = get_cluster_members(model, leader_unit, endpoints)
+    cluster_members = get_cluster_members(model, leader_unit, endpoints, tls_enabled=True)
     assert len(cluster_members) == NUM_UNITS
 
     for cluster_member in cluster_members:
-        assert cluster_member["clientURLs"][0].startswith("http://")
-        assert cluster_member["peerURLs"][0].startswith("http://")
+        assert cluster_member["clientURLs"][0].startswith("https")
+        assert cluster_member["peerURLs"][0].startswith("https")
 
     secret = await get_secret_by_label(ops_test, label=f"{PEER_RELATION}.{APP_NAME}.app")
     password = secret.get(f"{INTERNAL_USER}-password")
@@ -150,6 +165,7 @@ async def test_turning_off_tls(ops_test: OpsTest) -> None:
             user=INTERNAL_USER,
             password=password,
             key=TEST_KEY,
+            tls_enabled=True,
         )
         == TEST_VALUE
     )

tests/unit/test_charm.py

@@ -29,8 +29,9 @@ def test_install_failure_blocked_status():
 def test_internal_user_creation():
     ctx = testing.Context(EtcdOperatorCharm)
     relation = testing.PeerRelation(id=1, endpoint=PEER_RELATION)
+    restart_relation = testing.PeerRelation(id=2, endpoint="restart")
 
-    state_in = testing.State(relations={relation}, leader=True)
+    state_in = testing.State(relations={relation, restart_relation}, leader=True)
     state_out = ctx.run(ctx.on.leader_elected(), 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")

tests/unit/test_tls.py

@@ -4,10 +4,16 @@
 
 from unittest.mock import patch
 
-from ops import testing
+from ops import WaitingStatus, testing
 
 from charm import EtcdOperatorCharm
-from literals import CLIENT_TLS_RELATION_NAME, PEER_RELATION, PEER_TLS_RELATION_NAME, Status
+from literals import (
+    CLIENT_TLS_RELATION_NAME,
+    PEER_RELATION,
+    PEER_TLS_RELATION_NAME,
+    RESTART_RELATION,
+    Status,
+)
 
 
 def test_enable_tls_on_start():
@@ -43,11 +49,12 @@ def test_certificates_broken():
             "tls-state": "tls",
         },
     )
+    restart_peer_relation = testing.PeerRelation(id=4, endpoint=RESTART_RELATION)
     peer_tls_relation = testing.Relation(id=2, endpoint=PEER_TLS_RELATION_NAME)
     client_tls_relation = testing.Relation(id=3, endpoint=CLIENT_TLS_RELATION_NAME)
 
     state_in = testing.State(
-        relations=[peer_relation, peer_tls_relation, client_tls_relation],
+        relations=[peer_relation, restart_peer_relation, peer_tls_relation, client_tls_relation],
     )
 
     state_out = ctx.run(ctx.on.relation_broken(relation=peer_tls_relation), state_in)
@@ -69,26 +76,16 @@ def test_certificates_broken():
         patch("managers.tls.TLSManager.delete_certificates"),
         patch("managers.config.ConfigManager.set_config_properties"),
         patch("workload.EtcdWorkload.restart"),
+        # patch("charm.EtcdOperatorCharm.rolling_restart"),
     ):
         state_out = ctx.run(ctx.on.relation_broken(relation=peer_tls_relation), state_in)
         state_out = ctx.run(ctx.on.relation_broken(relation=client_tls_relation), state_out)
-        assert state_out.unit_status == Status.ACTIVE.value.status
-        assert state_out.get_relation(peer_relation.id).local_unit_data["tls-state"] == "no-tls"
+        assert state_out.unit_status == WaitingStatus("Awaiting restart operation")
+        assert state_out.get_relation(peer_relation.id).local_unit_data["tls-state"] == "to-no-tls"
         assert (
             state_out.get_relation(peer_relation.id).local_unit_data["client-cert-ready"]
             == "False"
         )
         assert (
             state_out.get_relation(peer_relation.id).local_unit_data["peer-cert-ready"] == "False"
         )
-
-    with (
-        patch("managers.cluster.ClusterManager.broadcast_peer_url"),
-        patch("managers.cluster.ClusterManager.health_check", return_value=False),
-        patch("managers.tls.TLSManager.delete_certificates"),
-        patch("managers.config.ConfigManager.set_config_properties"),
-        patch("workload.EtcdWorkload.restart"),
-    ):
-        state_out = ctx.run(ctx.on.relation_broken(relation=peer_tls_relation), state_in)
-        state_out = ctx.run(ctx.on.relation_broken(relation=client_tls_relation), state_out)
-        assert state_out.unit_status == Status.HEALTH_CHECK_FAILED.value.status