Add integration tests

.github/workflows/build-and-test.yaml

@@ -42,6 +42,11 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v3 # v3
+      - uses: actions/setup-python@v4
+        with:
+          python-version: '3.11'
+      - name: Install dependencies
+        run: python -m pip install tox
       - name: Set channel
         run: |
           juju_channel=$(echo "${{ matrix.agent-versions }}" | cut -c 1-3)

charmcraft.yaml

@@ -3,6 +3,8 @@ bases:
 - build-on:
   - name: "ubuntu"
     channel: "20.04"
+  - name: "ubuntu"
+    channel: "22.04"
   run-on:
   - name: "ubuntu"
     channel: "20.04"

pyproject.toml

@@ -15,7 +15,6 @@ log_cli_level = "INFO"
 # Formatting tools configuration
 [tool.black]
 line-length = 99
-target-version = ["py38"]
 
 # Linting tools configuration
 [tool.ruff]

src/charm.py

@@ -69,24 +69,24 @@ def __init__(self, *args):
 
         self.framework.observe(self.on.install, self._on_install)
         self.framework.observe(self.on.start, self._on_start)
-        self.framework.observe(self.on.upgrade_charm, self._on_upgrade_charm)
-        self.framework.observe(self.on.config_changed, self._on_config_changed)
+        self.framework.observe(self.on.upgrade_charm, self._reconfigure_nats)
+        self.framework.observe(self.on.config_changed, self._reconfigure_nats)
 
         listen_on_all_addresses = self.model.config["listen-on-all-addresses"]
         self.cluster = NatsCluster(self, "cluster", listen_on_all_addresses)
-        self.framework.observe(self.on.cluster_relation_changed, self._on_cluster_relation_changed)
+        self.framework.observe(self.on.cluster_relation_changed, self._reconfigure_nats)
 
         self.client = NatsClient(
             self, "client", listen_on_all_addresses, self.model.config["client-port"]
         )
-        self.framework.observe(self.on.client_relation_joined, self._on_client_relation_joined)
+        self.framework.observe(self.on.client_relation_joined, self._reconfigure_nats)
 
         self.ca_client = CAClient(self, "ca-client")
         self.framework.observe(self.ca_client.on.tls_config_ready, self._on_tls_config_ready)
-        self.framework.observe(self.ca_client.on.ca_available, self._on_ca_available)
+        self.framework.observe(self.ca_client.on.ca_available, self._reconfigure_nats)
 
         self.nrpe_client = NRPEClient(self, "nrpe-external-master")
-        self.framework.observe(self.nrpe_client.on.nrpe_available, self._on_nrpe_available)
+        self.framework.observe(self.nrpe_client.on.nrpe_available, self._reconfigure_nats)
 
     def _on_install(self, _):
         try:
@@ -144,13 +144,7 @@ def handle_tls_config(self):
                 self.TLS_CA_CERT_PATH.write_text(tls_ca_cert)
                 self.client._set_tls_ca(tls_ca_cert)
 
-    def _on_nrpe_available(self, _):
-        self._reconfigure_nats()
-
-    def _on_ca_available(self, _):
-        self._reconfigure_nats()
-
-    def _on_tls_config_ready(self, _):
+    def _on_tls_config_ready(self, event):
         self.TLS_KEY_PATH.write_bytes(
             self.ca_client.key.private_bytes(
                 encoding=serialization.Encoding.PEM,
@@ -164,15 +158,15 @@ def _on_tls_config_ready(self, _):
         self.TLS_CA_CERT_PATH.write_bytes(
             self.ca_client.ca_certificate.public_bytes(encoding=serialization.Encoding.PEM)
         )
-        self._reconfigure_nats()
+        self._reconfigure_nats(event)
 
     def _generate_content_hash(self, content):
         m = hashlib.sha256()
         m.update(content.encode("utf-8"))
         return m.hexdigest()
 
     # FIXME: reduce this function's complexity to satisfy the linter
-    def _reconfigure_nats(self):  # noqa: C901
+    def _reconfigure_nats(self, event):  # noqa: C901
         logger.info("Reconfiguring NATS")
         self.handle_tls_config()
         ctxt = {
@@ -303,18 +297,6 @@ def _on_start(self, _):
         self.on.nats_started.emit()
         self.model.unit.status = ActiveStatus()
 
-    def _on_cluster_relation_changed(self, _):
-        self._reconfigure_nats()
-
-    def _on_client_relation_joined(self, _):
-        self._reconfigure_nats()
-
-    def _on_config_changed(self, _):
-        self._reconfigure_nats()
-
-    def _on_upgrade_charm(self, _):
-        self._reconfigure_nats()
-
     def _open_port(self, port):
         subprocess.check_call(["open-port", port])
 

tests/integration/relation_tests/application-charm/charmcraft.yaml

@@ -0,0 +1,19 @@
+type: charm
+bases:
+- build-on:
+  - name: "ubuntu"
+    channel: "20.04"
+  - name: "ubuntu"
+    channel: "22.04"
+  run-on:
+  - name: "ubuntu"
+    channel: "20.04"
+    architectures: [arm64, amd64]
+  - name: "ubuntu"
+    channel: "22.04"
+    architectures: [arm64, amd64]
+parts:
+  charm:
+    charm-requirements: ["requirements.txt"]
+    build-packages:
+    - git

tests/integration/relation_tests/application-charm/config.yaml

@@ -0,0 +1,5 @@
+options:
+    check_clustering:
+        type: boolean
+        default: false
+        description: Whether to test clustering for the NATS cluster

tests/integration/relation_tests/application-charm/metadata.yaml

@@ -0,0 +1,11 @@
+name: nats-tester
+display-name: nats-test-app
+description: a testing application for nats charm
+summary: A simple application to relate to the nats interface to test the nats charm
+subordinate: false
+series:
+    - focal
+    - jammy
+requires:
+    client:
+        interface: nats

tests/integration/relation_tests/application-charm/requirements.txt

@@ -0,0 +1,2 @@
+nats-py==2.4.0
+ops==2.6.0

tests/integration/relation_tests/application-charm/src/charm.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+
+import asyncio
+import logging
+import ssl
+
+import nats
+import ops
+
+logger = logging.getLogger(__name__)
+
+
+class ApplicationCharm(ops.CharmBase):
+    """Application charm that connects to database charms."""
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Default charm events.
+        self.framework.observe(self.on.start, self._on_start)
+        self.framework.observe(self.on.client_relation_changed, self._on_client_relation_changed)
+
+    def _on_start(self, _):
+        self.unit.status = ops.ActiveStatus()
+
+    def _on_client_relation_changed(self, event):
+        unit_data = event.relation.data.get(event.unit)
+        if not unit_data:
+            return
+        url = unit_data.get("url")
+        if not url:
+            return
+        connect_opts = {}
+        if url.startswith("tls"):
+            cert = event.relation.data.get(event.app).get("ca_cert")
+            if not cert:
+                return
+            tls = ssl.create_default_context(cadata=cert)
+            connect_opts.update({"tls": tls})
+
+        async def _verify_connection(url: str, opts: dict):
+            client = await nats.connect(url, **opts)
+            logger.info(f"connected to {url}")
+            if self.config["check_clustering"]:
+                logger.info("checking for clustering")
+                assert len(client.servers) > 1, f"NATS not clustered: {client.servers}"
+            channel_name = "test"
+            message = b"testing"
+            sub = await client.subscribe("test")
+            await client.publish("test", b"testing")
+            msg = await sub.next_msg()
+            assert (
+                msg.data == message
+            ), f"messages do not match. Expected: {message}, Got: {msg.data}"
+            assert (
+                msg.subject == channel_name
+            ), f"messages do not match. Expected: {channel_name}, Got: {msg.subject}"
+            logger.info("connection check complete")
+            await sub.unsubscribe()
+
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(_verify_connection(url, connect_opts))
+        self.unit.status = ops.ActiveStatus()
+
+
+if __name__ == "__main__":
+    ops.main(ApplicationCharm)

tests/integration/relation_tests/helpers.py

@@ -0,0 +1,156 @@
+from pathlib import Path
+from typing import Literal, Optional
+
+import yaml
+from pytest_operator.plugin import OpsTest
+from tenacity import RetryError, Retrying, stop_after_attempt, wait_exponential
+
+APPLICATION_APP_NAME = "nats-tester"
+TEST_APP_CHARM_PATH = "tests/integration/relation_tests/application-charm"
+CHARM_NAME = yaml.safe_load(Path("metadata.yaml").read_text())["name"]
+APP_NAMES = [APPLICATION_APP_NAME, CHARM_NAME]
+
+
+async def get_alias_from_relation_data(
+    ops_test: OpsTest, unit_name: str, related_unit_name: str
+) -> Optional[str]:
+    """Get the alias that the unit assigned to the related unit application/cluster.
+
+    Args:
+        ops_test: The ops test framework instance
+        unit_name: The name of the unit
+        related_unit_name: name of the related unit
+
+    Returns:
+        the alias for the application/cluster of
+            the related unit
+
+    Raises:
+        ValueError if it's not possible to get unit data
+            or if there is no alias on that.
+    """
+    raw_data = (await ops_test.juju("show-unit", related_unit_name))[1]
+    if not raw_data:
+        raise ValueError(f"no unit info could be grabbed for {related_unit_name}")
+    data = yaml.safe_load(raw_data)
+
+    # Retrieve the relation data from the unit.
+    relation_data = {}
+    for relation in data[related_unit_name]["relation-info"]:
+        for name, unit in relation["related-units"].items():
+            if name == unit_name:
+                relation_data = unit["data"]
+                break
+
+    # Check whether the unit has set an alias for the related unit application/cluster.
+    if "alias" not in relation_data:
+        raise ValueError(f"no alias could be grabbed for {related_unit_name} application/cluster")
+
+    return relation_data["alias"]
+
+
+async def get_relation_data(
+    ops_test: OpsTest,
+    application_name: str,
+    relation_name: str,
+    key: str,
+    databag: Literal["application", "unit"],
+    relation_id: str | None = None,
+    relation_alias: str | None = None,
+) -> Optional[str]:
+    """Get relation data for an application.
+
+    Args:
+        ops_test: The ops test framework instance
+        application_name: The name of the application
+        relation_name: name of the relation to get connection data from
+        key: key of data to be retrieved
+        databag: Type of data bag i.e application or unit, to check the key in. Defaults to "unit".
+        relation_id: id of the relation to get connection data from
+        relation_alias: alias of the relation (like a connection name)
+            to get connection data from
+
+    Returns:
+        the data that was requested or None
+            if no data in the relation
+
+    Raises:
+        ValueError if it's not possible to get application data
+            or if there is no data for the particular relation endpoint
+            and/or alias.
+    """
+    unit_name = ops_test.model.applications[application_name].units[0].name
+    raw_data = (await ops_test.juju("show-unit", unit_name))[1]
+    if not raw_data:
+        raise ValueError(f"no unit info could be grabbed for {unit_name}")
+    data = yaml.safe_load(raw_data)
+    # Filter the data based on the relation name.
+    relation_data = [v for v in data[unit_name]["relation-info"] if v["endpoint"] == relation_name]
+    if relation_id:
+        # Filter the data based on the relation id.
+        relation_data = [v for v in relation_data if v["relation-id"] == relation_id]
+    if relation_alias:
+        # Filter the data based on the cluster/relation alias.
+        relation_data = [
+            v
+            for v in relation_data
+            if await get_alias_from_relation_data(
+                ops_test, unit_name, next(iter(v["related-units"]))
+            )
+            == relation_alias
+        ]
+    if len(relation_data) == 0:
+        raise ValueError(
+            f"no relation data could be grabbed on relation with endpoint {relation_name} and alias {relation_alias}"
+        )
+    if databag == "application":
+        return relation_data[0]["application-data"].get(key)
+    elif databag == "unit":
+        related_unit = relation_data[0]["related-units"].popitem()
+        return related_unit[1]["data"].get(key)
+    else:
+        raise ValueError("databag can only be of type 'unit' or 'application'")
+
+
+async def check_relation_data_existence(
+    ops_test: OpsTest,
+    application_name: str,
+    relation_name: str,
+    key: str,
+    exists: bool = True,
+    databag: Literal["unit", "application"] = "unit",
+) -> bool:
+    """Check for the existence of a key in the relation data.
+
+    Args:
+        ops_test: The ops test framework instance
+        application_name: The name of the application
+        relation_name: Name of the relation to get relation data from
+        key: Key of data to be checked
+        exists: Whether to check for the existence or non-existence
+        databag: Type of data bag i.e application or unit, to check the key in. Defaults to "unit".
+
+    Returns:
+        whether the key exists in the relation data
+    """
+    try:
+        # Retry mechanism used to wait for some events to be triggered,
+        # like the relation departed event.
+        for attempt in Retrying(
+            stop=stop_after_attempt(10), wait=wait_exponential(multiplier=1, min=2, max=30)
+        ):
+            with attempt:
+                data = await get_relation_data(
+                    ops_test,
+                    application_name,
+                    relation_name,
+                    key,
+                    databag,
+                )
+                if exists:
+                    assert data is not None
+                else:
+                    assert data is None
+        return True
+    except RetryError:
+        return False