[DPE-6374] Add integration tests (#19)

Author: phvalguima

.github/workflows/ci.yaml

@@ -70,3 +70,18 @@ jobs:
       charmcraft-snap-channel: latest/beta/data-platform  # TODO: remove after charmcraft 3.3 stable release
       path-to-charm-directory: .
       cache: false  # TODO: change this to true once we are in charmcraftcache-hub
+
+  integration-test:
+    name: Integration test charm
+    needs:
+      - lint
+      - unit-test
+      - build
+    uses: canonical/data-platform-workflows/.github/workflows/[email protected]
+    with:
+      juju-agent-version: 3.6.1 # renovate: juju-agent-pin-minor
+      _beta_allure_report: true
+      artifact-prefix: ${{ needs.build.outputs.artifact-prefix }}
+      cloud: lxd
+    permissions:
+      contents: write

.github/workflows/release.yaml

@@ -16,7 +16,6 @@ jobs:
       pull-requests: write # Need to create PR
       actions: write
 
-
   release:
     name: Release to Charmhub
     needs:

poetry.lock

@@ -66,6 +66,7 @@ parameterized = "^0.9.0"
 [tool.poetry.group.integration.dependencies]
 pytest = "^8.3.3"
 pytest-github-secrets = {git = "https://github.com/canonical/data-platform-workflows", tag = "v26.0.0", subdirectory = "python/pytest_plugins/github_secrets"}
+pytest-asyncio = "^0.21.2"
 pytest-operator = "^0.37.0"
 pytest-operator-cache = {git = "https://github.com/canonical/data-platform-workflows", tag = "v26.0.0", subdirectory = "python/pytest_plugins/pytest_operator_cache"}
 pytest-operator-groups = {git = "https://github.com/canonical/data-platform-workflows", tag = "v26.0.0", subdirectory = "python/pytest_plugins/pytest_operator_groups"}
@@ -75,6 +76,8 @@ ops = "^2.17.0"
 pyyaml = "^6.0.2"
 urllib3 = "^2.2.3"
 protobuf = "5.28.2"
+allure-pytest = "^2.13.5"
+allure-pytest-collection-report = {git = "https://github.com/canonical/data-platform-workflows", tag = "v29.0.0", subdirectory = "python/pytest_plugins/allure_pytest_collection_report"}
 
 [tool.coverage.run]
 branch = true

pyproject.toml

@@ -348,22 +348,18 @@ def _render_params(
     def prepare(self) -> bool:
         """Prepare the benchmark service."""
         super().prepare()
-
-        # First, clean if a topic already existed
-        self.clean()
         try:
-            for attempt in Retrying(stop=stop_after_attempt(4), wait=wait_fixed(wait=15)):
+            for attempt in Retrying(
+                stop=stop_after_attempt(4), wait=wait_fixed(wait=20), reraise=True
+            ):
                 with attempt:
-                    if model := self.database_state.model():
+                    if (model := self.database_state.model()) and not self.is_prepared():
                         topic = NewTopic(
                             name=model.db_name,
                             num_partitions=self.config.parallel_processes * (len(self.peers) + 1),
                             replication_factor=self.client.replication_factor,
                         )
                         self.client.create_topic(topic)
-            else:
-                logger.warning("No database model found")
-                return False
         except Exception as e:
             logger.debug(f"Error creating topic: {e}")
 

src/charm.py

@@ -1,4 +1,4 @@
-# Copyright 2022 Canonical Ltd.
+# Copyright 2025 Canonical Ltd.
 # See LICENSE file for licensing details.
 
 import argparse

tests/conftest.py

@@ -0,0 +1,2 @@
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.

tests/integration/__init__.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+
+import subprocess
+
+import juju
+import pytest
+import yaml
+from pytest_operator.plugin import OpsTest
+
+from .helpers import (
+    K8S_DB_MODEL_NAME,
+    MICROK8S_CLOUD_NAME,
+)
+
+
+@pytest.fixture(scope="module", autouse=True)
+async def destroy_model_in_k8s(ops_test):
+    yield
+
+    if ops_test.keep_model:
+        return
+    controller = juju.controller.Controller()
+    await controller.connect()
+    await controller.destroy_model(K8S_DB_MODEL_NAME)
+    await controller.disconnect()
+
+    ctlname = list(yaml.safe_load(subprocess.check_output(["juju", "show-controller"])).keys())[0]
+
+    # We have deployed microk8s, and we do not need it anymore
+    subprocess.run(["sudo", "snap", "remove", "--purge", "microk8s"], check=True)
+    subprocess.run(["sudo", "snap", "remove", "--purge", "kubectl"], check=True)
+    subprocess.run(
+        ["juju", "remove-cloud", "--client", "--controller", ctlname, MICROK8S_CLOUD_NAME],
+        check=True,
+    )
+
+
+@pytest.fixture(scope="module")
+async def kafka_benchmark_charm(ops_test: OpsTest):
+    """Kafka charm used for integration testing."""
+    charm = await ops_test.build_charm(".")
+    return charm

tests/integration/conftest.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+
+import logging
+import subprocess
+from types import SimpleNamespace
+
+from pytest_operator.plugin import OpsTest
+from tenacity import Retrying, stop_after_delay, wait_fixed
+
+logger = logging.getLogger(__name__)
+
+
+K8S_DB_MODEL_NAME = "database"
+MICROK8S_CLOUD_NAME = "cloudk8s"
+
+
+CONFIG_OPTS = {"workload_name": "test_mode", "parallel_processes": 1}
+SERIES = "jammy"
+KAFKA = "kafka"
+APP_NAME = "kafka-benchmark"
+KAFKA_CHANNEL = "3/edge"
+DEFAULT_NUM_UNITS = 2
+
+
+KRAFT_CONFIG = {
+    "profile": "testing",
+    "roles": "broker,controller",
+}
+
+
+MODEL_CONFIG = {
+    "logging-config": "<root>=INFO;unit=DEBUG",
+    "update-status-hook-interval": "1m",
+}
+
+
+def check_service(svc_name: str, unit_id: int = 0, retry_if_fail: bool = True) -> bool | None:
+    def __check():
+        try:
+            return (
+                subprocess.check_output(
+                    [
+                        "juju",
+                        "ssh",
+                        f"{APP_NAME}/{unit_id}",
+                        "--",
+                        "sudo",
+                        "systemctl",
+                        "is-active",
+                        svc_name,
+                    ],
+                    text=True,
+                ).rstrip()
+                == "active"
+            )
+        except Exception:
+            return False
+
+    if not retry_if_fail:
+        return __check()
+    for attempt in Retrying(stop=stop_after_delay(150), wait=wait_fixed(15)):
+        with attempt:
+            return __check()
+
+
+async def run_action(
+    ops_test, action_name: str, unit_name: str, timeout: int = 30, **action_kwargs
+):
+    """Runs the given action on the given unit."""
+    client_unit = ops_test.model.units.get(unit_name)
+    action = await client_unit.run_action(action_name, **action_kwargs)
+    result = await action.wait()
+    logging.info(f"request results: {result.results}")
+    return SimpleNamespace(status=result.status or "completed", response=result.results)
+
+
+async def get_leader_unit_id(ops_test: OpsTest, app: str = APP_NAME) -> int:
+    """Helper function that retrieves the leader unit ID."""
+    leader_unit = None
+    for unit in ops_test.model.applications[app].units:
+        if await unit.is_leader_from_status():
+            leader_unit = unit
+            break
+
+    if not leader_unit:
+        raise ValueError("Leader unit not found")
+
+    return int(leader_unit.name.split("/")[1])