run the etcdctl commands through juju ssh

tests/integration/helpers.py

@@ -3,6 +3,7 @@
 # See LICENSE file for licensing details.
 
 import json
+import logging
 import subprocess
 from pathlib import Path
 
@@ -11,55 +12,34 @@
 
 from literals import CLIENT_PORT
 
+logger = logging.getLogger(__name__)
+
 METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
 APP_NAME = METADATA["name"]
 
 
-def put_key(endpoints: str, key: str, value: str) -> str:
-    """Write data to etcd using `etcdctl`."""
-    return subprocess.run(
-        args=[
-            "etcdctl",
-            "put",
-            key,
-            value,
-            f"--endpoints={endpoints}",
-        ],
-        check=True,
-        capture_output=True,
-        text=True,
-    ).stdout.split("\n")[0]
-
-
-def get_key(endpoints: str, key: str) -> str:
-    """Read data from etcd using `etcdctl`."""
-    return subprocess.run(
-        args=[
-            "etcdctl",
-            "get",
-            key,
-            f"--endpoints={endpoints}",
-        ],
-        check=True,
-        capture_output=True,
-        text=True,
-    ).stdout.split("\n")[1]
-
-
-def get_cluster_members(endpoints: str) -> list[dict]:
-    """Query all cluster members from etcd using `etcdctl`."""
-    result = subprocess.run(
-        args=[
-            "etcdctl",
-            "member",
-            "list",
-            f"--endpoints={endpoints}",
-            "-w=json",
-        ],
-        check=True,
-        capture_output=True,
-        text=True,
-    ).stdout
+def put_key(model: str, unit: str, endpoints: str, key: str, value: str) -> str:
+    """Write data to etcd using `etcdctl` via `juju ssh`."""
+    etcd_command = f"etcdctl put {key} {value} --endpoints={endpoints}"
+    juju_command = f"juju ssh --model={model} {unit} {etcd_command}"
+
+    return subprocess.getoutput(juju_command).split("\n")[0]
+
+
+def get_key(model: str, unit: str, endpoints: str, key: str) -> str:
+    """Read data from etcd using `etcdctl` via `juju ssh`."""
+    etcd_command = f"etcdctl get {key} --endpoints={endpoints}"
+    juju_command = f"juju ssh --model={model} {unit} {etcd_command}"
+
+    return subprocess.getoutput(juju_command).split("\n")[1]
+
+
+def get_cluster_members(model: str, unit: str, endpoints: str) -> list[dict]:
+    """Query all cluster members from etcd using `etcdctl` via `juju ssh`."""
+    etcd_command = f"etcdctl member list --endpoints={endpoints} -w=json"
+    juju_command = f"juju ssh --model={model} {unit} {etcd_command}"
+
+    result = subprocess.getoutput(juju_command).split("\n")[0]
 
     return json.loads(result)["members"]
 
@@ -68,7 +48,14 @@ def get_cluster_endpoints(ops_test: OpsTest, app_name: str = APP_NAME) -> str:
     """Resolve the etcd endpoints for a given juju application."""
     return ",".join(
         [
-            f"http://{unit.public_address}:{(CLIENT_PORT)}"
+            f"http://{unit.public_address}:{CLIENT_PORT}"
             for unit in ops_test.model.applications[app_name].units
         ]
     )
+
+
+async def get_juju_leader_unit_name(ops_test: OpsTest, app_name: str = APP_NAME) -> str:
+    """Retrieve the leader unit name."""
+    for unit in ops_test.model.applications[app_name].units:
+        if await unit.is_leader_from_status():
+            return unit.name

tests/integration/test_charm.py

@@ -3,29 +3,46 @@
 # See LICENSE file for licensing details.
 
 import logging
-from pathlib import Path
 
 import pytest
-import yaml
 from pytest_operator.plugin import OpsTest
 
-logger = logging.getLogger(__name__)
+from .helpers import (
+    APP_NAME,
+    get_cluster_endpoints,
+    get_cluster_members,
+    get_juju_leader_unit_name,
+    get_key,
+    put_key,
+)
 
-METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
-APP_NAME = METADATA["name"]
+logger = logging.getLogger(__name__)
 
 
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
 async def test_build_and_deploy(ops_test: OpsTest) -> None:
-    """Build the charm-under-test and deploy it together with related charms.
+    """Build the charm-under-test and deploy it with three units.
 
-    Assert on the status before any relations/configurations take place.
+    The initial cluster should be formed and accessible.
     """
     # Build and deploy charm from local source folder
     etcd_charm = await ops_test.build_charm(".")
+    model = ops_test.model_full_name
 
     # Deploy the charm and wait for active/idle status
-    await ops_test.model.deploy(etcd_charm, num_units=1)
-
+    await ops_test.model.deploy(etcd_charm, num_units=3)
     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)
+    leader_unit = await get_juju_leader_unit_name(ops_test, APP_NAME)
+
+    cluster_members = get_cluster_members(model, leader_unit, endpoints)
+    assert len(cluster_members) == 3
+
+    # make sure data can be written to the cluster
+    test_key = "test_key"
+    test_value = "42"
+    assert put_key(model, leader_unit, endpoints, key=test_key, value=test_value) == "OK"
+    assert get_key(model, leader_unit, endpoints, key=test_key) == test_value