Merge branch 'main' into DPE-6431-break-main-charm

phvalguima · web-flow · commit d3af469bc308 · 2025-02-19T13:56:21.000+01:00
diff --git a/lib/charms/kafka/v0/client.py b/lib/charms/kafka/v0/client.py
@@ -77,7 +77,7 @@ def on_kafka_relation_created(self, event: RelationCreatedEvent):
 import time
 import sys
 from functools import cached_property
-from typing import Generator, List, Optional
+from typing import Generator, List, Optional, Any
 
 from kafka import KafkaAdminClient, KafkaConsumer, KafkaProducer
 from kafka.admin import NewTopic
@@ -184,6 +184,10 @@ def _consumer_client(self) -> KafkaConsumer:
             consumer_timeout_ms=15000,
         )
 
+    def describe_cluster(self) -> Any:
+        """Returns the cluster metadata."""
+        return self._admin_client.describe_cluster()
+
     def create_topic(self, topic: NewTopic) -> None:
         """Creates a new topic on the Kafka cluster.
 
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
@@ -3,12 +3,15 @@
 # See LICENSE file for licensing details.
 
 
+import pathlib
 import subprocess
+from types import SimpleNamespace
 
 import juju
 import pytest
 import yaml
 from pytest_operator.plugin import OpsTest
+from tenacity import Retrying, stop_after_delay, wait_fixed
 
 from .helpers import (
     K8S_DB_MODEL_NAME,
@@ -43,3 +46,48 @@ async def kafka_benchmark_charm(ops_test: OpsTest):
     """Kafka charm used for integration testing."""
     charm = await ops_test.build_charm(".")
     return charm
+
+
+@pytest.fixture(scope="module")
+async def microk8s(ops_test: OpsTest) -> SimpleNamespace:
+    controller = yaml.safe_load(subprocess.check_output(["juju", "show-controller"]))
+
+    for controller_name in controller.keys():
+        # controller_data = details["details"]
+        try:
+            subprocess.run(["sudo", "snap", "install", "--classic", "microk8s"], check=True)
+            subprocess.run(["sudo", "snap", "install", "--classic", "kubectl"], check=True)
+            subprocess.run(["sudo", "microk8s", "enable", "dns", "hostpath-storage"], check=True)
+
+            # Configure kubectl now
+            subprocess.run(["mkdir", "-p", str(pathlib.Path.home() / ".kube")], check=True)
+            kubeconfig = subprocess.check_output(["sudo", "microk8s", "config"])
+            with open(str(pathlib.Path.home() / ".kube" / "config"), "w") as f:
+                f.write(kubeconfig.decode())
+            for attempt in Retrying(stop=stop_after_delay(150), wait=wait_fixed(15)):
+                with attempt:
+                    if (
+                        len(
+                            subprocess.check_output(
+                                "kubectl get po -A  --field-selector=status.phase!=Running",
+                                shell=True,
+                                stderr=subprocess.DEVNULL,
+                            ).decode()
+                        )
+                        != 0
+                    ):  # We got sth different than "No resources found." in stderr
+                        raise Exception()
+
+            # Get controller name
+            ctlname = controller_name
+
+            # Add microk8s to the kubeconfig
+            subprocess.run(
+                ["juju", "add-k8s", MICROK8S_CLOUD_NAME, "--client", "--controller", ctlname],
+                check=True,
+            )
+
+        except subprocess.CalledProcessError as e:
+            pytest.exit(str(e))
+
+    return SimpleNamespace(cloud_name=MICROK8S_CLOUD_NAME)
diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py
@@ -7,6 +7,7 @@
 import subprocess
 from types import SimpleNamespace
 
+import pytest
 from pytest_operator.plugin import OpsTest
 from tenacity import Retrying, stop_after_delay, wait_fixed
 
@@ -20,9 +21,52 @@
 CONFIG_OPTS = {"workload_name": "test_mode", "parallel_processes": 1}
 SERIES = "jammy"
 KAFKA = "kafka"
+KAFKA_K8S = "kafka-k8s"
 APP_NAME = "kafka-benchmark"
 KAFKA_CHANNEL = "3/edge"
 DEFAULT_NUM_UNITS = 2
+K8S_DB_MODEL_NAME = "database"
+MICROK8S_CLOUD_NAME = "uk8s"
+
+
+DEPLOY_MARKS = [
+    (
+        pytest.param(
+            use_tls,
+            cloud,
+            id=str(use_tls) + f"-{cloud}",
+            marks=pytest.mark.group(str(use_tls) + f"-{cloud}"),
+        )
+    )
+    for use_tls in [True, False]
+    for cloud in ["vm", MICROK8S_CLOUD_NAME]
+]
+
+K8S_MARKS = [
+    (
+        pytest.param(
+            use_tls,
+            cloud,
+            id=str(use_tls) + f"-{cloud}",
+            marks=pytest.mark.group(str(use_tls) + f"-{cloud}"),
+        )
+    )
+    for use_tls in [True, False]
+    for cloud in [MICROK8S_CLOUD_NAME]
+]
+
+VM_MARKS = [
+    (
+        pytest.param(
+            use_tls,
+            cloud,
+            id=str(use_tls) + f"-{cloud}",
+            marks=pytest.mark.group(str(use_tls) + f"-{cloud}"),
+        )
+    )
+    for use_tls in [True, False]
+    for cloud in ["vm"]
+]
 
 
 KRAFT_CONFIG = {
diff --git a/tests/integration/test_charm.py b/tests/integration/test_charm.py
@@ -4,6 +4,7 @@
 
 import logging
 
+import juju
 import pytest
 from pytest_operator.plugin import OpsTest
 from tenacity import Retrying, stop_after_attempt, wait_fixed
@@ -12,11 +13,16 @@
     APP_NAME,
     CONFIG_OPTS,
     DEFAULT_NUM_UNITS,
+    DEPLOY_MARKS,
+    K8S_DB_MODEL_NAME,
+    K8S_MARKS,
     KAFKA,
     KAFKA_CHANNEL,
+    KAFKA_K8S,
     KRAFT_CONFIG,
     MODEL_CONFIG,
     SERIES,
+    VM_MARKS,
     check_service,
     get_leader_unit_id,
     run_action,
@@ -25,23 +31,78 @@
 logger = logging.getLogger(__name__)
 
 
-TLS_MARK = {
-    (use_tls): pytest.param(
-        use_tls,
-        id=str(use_tls),
-        marks=[
-            pytest.mark.group(str(use_tls)),
-        ],
+model_db = None
+
+
+@pytest.mark.parametrize("use_tls,cloud", K8S_MARKS)
+@pytest.mark.abort_on_fail
+@pytest.mark.skip_if_deployed
+async def test_build_and_deploy_k8s_only(
+    ops_test: OpsTest, microk8s, kafka_benchmark_charm, use_tls, cloud
+) -> None:
+    """Build and deploy with and without TLS on k8s."""
+    logging.info(f"Creating k8s model {K8S_DB_MODEL_NAME}")
+    controller = juju.controller.Controller()
+    await controller.connect()
+    await controller.add_model(K8S_DB_MODEL_NAME, cloud_name=microk8s.cloud_name)
+
+    global model_db
+    model_db = juju.model.Model()
+    await model_db.connect(model_name=K8S_DB_MODEL_NAME)
+
+    await ops_test.model.set_config(MODEL_CONFIG)
+    await ops_test.model.deploy(
+        kafka_benchmark_charm,
+        num_units=DEFAULT_NUM_UNITS,
+        series=SERIES,
+        config=CONFIG_OPTS,
+    )
+    await model_db.deploy(
+        KAFKA_K8S,
+        channel=KAFKA_CHANNEL,
+        config=KRAFT_CONFIG | {"expose_external": "nodeport"},
+        num_units=DEFAULT_NUM_UNITS,
+        series=SERIES,
+        trust=True,
+    )
+    await model_db.create_offer(
+        endpoint="kafka-client",
+        offer_name="kafka-client",
+        application_name=KAFKA_K8S,
     )
-    for use_tls in [True, False]
-}
-USE_TLS = list(TLS_MARK.values())
+    await ops_test.model.consume(f"admin/{model_db.name}.kafka-client")
+    await ops_test.model.integrate("kafka-client", f"{APP_NAME}:kafka")
 
+    if use_tls:
+        await ops_test.model.deploy(
+            "self-signed-certificates",
+            num_units=1,
+            series=SERIES,
+        )
+        await ops_test.model.integrate(APP_NAME, "self-signed-certificates")
 
-@pytest.mark.parametrize("use_tls", USE_TLS)
+        await ops_test.model.create_offer(
+            endpoint="certificates",
+            offer_name="certificates",
+            application_name="self-signed-certificates",
+        )
+        await model_db.consume(f"admin/{ops_test.model.name}.certificates")
+        await model_db.integrate("certificates", f"{KAFKA_K8S}:certificates")
+
+    await ops_test.model.wait_for_idle(apps=[APP_NAME], status="waiting", timeout=2000)
+    await model_db.wait_for_idle(apps=[KAFKA_K8S], status="active", timeout=2000)
+
+    assert len(ops_test.model.applications[APP_NAME].units) == DEFAULT_NUM_UNITS
+    await controller.disconnect()
+    await model_db.disconnect()
+
+
+@pytest.mark.parametrize("use_tls,cloud", VM_MARKS)
 @pytest.mark.abort_on_fail
 @pytest.mark.skip_if_deployed
-async def test_deploy(ops_test: OpsTest, kafka_benchmark_charm, use_tls) -> None:
+async def test_build_and_deploy_vm_only(
+    ops_test: OpsTest, kafka_benchmark_charm, use_tls, cloud
+) -> None:
     """Build and deploy with and without TLS."""
     await ops_test.model.set_config(MODEL_CONFIG)
 
@@ -73,10 +134,14 @@ async def test_deploy(ops_test: OpsTest, kafka_benchmark_charm, use_tls) -> None
 
     assert len(ops_test.model.applications[APP_NAME].units) == DEFAULT_NUM_UNITS
 
+    # set the model to the global model_db
+    global model_db
+    model_db = ops_test.model
+
 
-@pytest.mark.parametrize("use_tls", USE_TLS)
+@pytest.mark.parametrize("use_tls,cloud", DEPLOY_MARKS)
 @pytest.mark.abort_on_fail
-async def test_prepare(ops_test: OpsTest, use_tls) -> None:
+async def test_prepare(ops_test: OpsTest, use_tls, cloud) -> None:
     """Test prepare action."""
     leader_id = await get_leader_unit_id(ops_test)
 
@@ -93,9 +158,9 @@ async def test_prepare(ops_test: OpsTest, use_tls) -> None:
     )
 
 
-@pytest.mark.parametrize("use_tls", USE_TLS)
+@pytest.mark.parametrize("use_tls,cloud", DEPLOY_MARKS)
 @pytest.mark.abort_on_fail
-async def test_run(ops_test: OpsTest, use_tls) -> None:
+async def test_run(ops_test: OpsTest, use_tls, cloud) -> None:
     """Test run action."""
     leader_id = await get_leader_unit_id(ops_test)
 
@@ -111,9 +176,9 @@ async def test_run(ops_test: OpsTest, use_tls) -> None:
     assert check_service("dpe_benchmark", unit_id=leader_id)
 
 
-@pytest.mark.parametrize("use_tls", USE_TLS)
+@pytest.mark.parametrize("use_tls,cloud", DEPLOY_MARKS)
 @pytest.mark.abort_on_fail
-async def test_stop(ops_test: OpsTest, use_tls) -> None:
+async def test_stop(ops_test: OpsTest, use_tls, cloud) -> None:
     """Test stop action."""
     leader_id = await get_leader_unit_id(ops_test)
 
@@ -129,9 +194,9 @@ async def test_stop(ops_test: OpsTest, use_tls) -> None:
     assert not check_service("dpe_benchmark", unit_id=leader_id)
 
 
-@pytest.mark.parametrize("use_tls", USE_TLS)
+@pytest.mark.parametrize("use_tls,cloud", DEPLOY_MARKS)
 @pytest.mark.abort_on_fail
-async def test_restart(ops_test: OpsTest, use_tls) -> None:
+async def test_restart(ops_test: OpsTest, use_tls, cloud) -> None:
     """Test stop and restart the benchmark."""
     leader_id = await get_leader_unit_id(ops_test)
 
@@ -147,9 +212,9 @@ async def test_restart(ops_test: OpsTest, use_tls) -> None:
     assert check_service("dpe_benchmark", unit_id=leader_id)
 
 
-@pytest.mark.parametrize("use_tls", USE_TLS)
+@pytest.mark.parametrize("use_tls,cloud", DEPLOY_MARKS)
 @pytest.mark.abort_on_fail
-async def test_clean(ops_test: OpsTest, use_tls) -> None:
+async def test_clean(ops_test: OpsTest, use_tls, cloud) -> None:
     """Test cleanup action."""
     leader_id = await get_leader_unit_id(ops_test)
 
