Integration test for templated notification by receiver (#74)

* Added support for setting alerts in Client

This commit adds support for setting alerts in the Alertmanager
client object. This support is needed to create integration tests
which involve triggering alerts and checking that these alerts
are correctly forwarded to the receiver.

* Fix double quoted template file path

This commit fixes a bug that results in a failure of alertmanager
templates configuration option from working as expected. The
commit also fixes a corresponding unit tests.

* An integration test altermanger templates

This commit adds and integration test to check that the alertmanger
templates configuration option works as expected. The test checks
that a fake alert receiver does get notifications that are customized
using a template.

Closes #40

* Added unit test for alertmanager clean set alerts

This commit adds a unit tests for alert setting api of
alertmanager client.

* Added type hints to Alertmanager client

This commit adds a few type hints to the new methods of alertmanager
client.

Co-authored-by: Leon <[email protected]>

src/alertmanager_client.py

@@ -155,3 +155,52 @@ def config(self) -> dict:
             return yaml.safe_load(config)
         except yaml.YAMLError as e:
             raise AlertmanagerBadResponse("Response is not a YAML string") from e
+
+    def _post(
+        self, url: str, post_data: bytes, headers: dict = None, timeout: int = None
+    ) -> bytes:
+        """Make a HTTP POST request to Alertmanager.
+
+        Args:
+            url: string URL where POST request is sent.
+            post_data: encoded string (bytes) of data to be posted.
+            headers: dictionary containing HTTP headers to be used for POST request.
+            timeout: numeric timeout value in seconds.
+
+        Returns:
+            urllib response object.
+        """
+        response = "".encode("utf-8")
+        timeout = timeout or self.timeout
+        request = urllib.request.Request(url, headers=headers or {}, data=post_data, method="POST")
+
+        try:
+            response = urllib.request.urlopen(request, timeout=timeout)
+        except urllib.error.HTTPError as error:
+            logger.debug(
+                "Failed posting to %s, reason: %s",
+                url,
+                error.reason,
+            )
+        except urllib.error.URLError as error:
+            logger.debug("Invalid URL %s : %s", url, error)
+        except TimeoutError:
+            logger.debug("Request timeout during posting to URL %s", url)
+        return response
+
+    def set_alerts(self, alerts: list) -> bytes:
+        """Send a set of new alerts to alertmanger.
+
+        Args:
+            alerts: a list of alerts to be set. Format of this list is
+               described here https://prometheus.io/docs/alerting/latest/clients/.
+
+        Returns:
+            urllib response object.
+        """
+        url = urllib.parse.urljoin(self.base_url, "/api/v1/alerts")
+        headers = {"Content-Type": "application/json"}
+        post_data = json.dumps(alerts).encode("utf-8")
+        response = self._post(url, post_data, headers=headers)
+
+        return response

src/charm.py

@@ -277,7 +277,7 @@ def _update_config(self) -> None:
 
         # add templates, if any
         if templates := self.config["templates_file"]:
-            config["templates"] = [f"'{self._templates_path}'"]
+            config["templates"] = [f"{self._templates_path}"]
             self.container.push(self._templates_path, templates, make_dirs=True)
 
         # add juju topology to "group_by"

tests/integration/conftest.py

@@ -2,15 +2,33 @@
 # Copyright 2021 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import socket
 from pathlib import Path
 
 import pytest
 from pytest_operator.plugin import OpsTest
 
+PYTEST_HTTP_SERVER_PORT = 8000
+
 
 @pytest.fixture(scope="module")
 async def charm_under_test(ops_test: OpsTest) -> Path:
     """Charm used for integration testing."""
     path_to_built_charm = await ops_test.build_charm(".")
 
     return path_to_built_charm
+
+
+@pytest.fixture(scope="session")
+def httpserver_listen_address():
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    s.settimeout(0)
+    try:
+        # ip address does not need to be reachable
+        s.connect(("8.8.8.8", 1))
+        local_ip_address = s.getsockname()[0]
+    except Exception:
+        local_ip_address = "127.0.0.1"
+    finally:
+        s.close()
+    return (local_ip_address, PYTEST_HTTP_SERVER_PORT)

tests/integration/test_templates.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+# Copyright 2022 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+import yaml
+from helpers import get_unit_address, is_alertmanager_up
+from pytest_operator.plugin import OpsTest
+from werkzeug.wrappers import Request, Response
+
+from alertmanager_client import Alertmanager
+
+logger = logging.getLogger(__name__)
+
+METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
+app_name = METADATA["name"]
+resources = {"alertmanager-image": METADATA["resources"]["alertmanager-image"]["upstream-source"]}
+
+
+def request_handler(request: Request):
+    response = Response("OK", status=200, content_type="text/plain")
+    logger.info("Got Request Data : %s", json.loads(request.data.decode("utf-8")))
+    return response
+
+
+@pytest.mark.abort_on_fail
+async def test_receiver_gets_alert(ops_test: OpsTest, charm_under_test, httpserver):
+
+    # deploy charm from local source folder
+    await ops_test.model.deploy(charm_under_test, resources=resources, application_name=app_name)
+    await ops_test.model.wait_for_idle(apps=[app_name], status="active", timeout=1000)
+    assert ops_test.model.applications[app_name].units[0].workload_status == "active"
+    assert await is_alertmanager_up(ops_test, app_name)
+
+    # define the alertmanager configuration
+    receiver_name = "fake-receiver"
+    aconfig = {
+        "global": {"http_config": {"tls_config": {"insecure_skip_verify": True}}},
+        "route": {
+            "group_by": ["alertname"],
+            "group_wait": "3s",
+            "group_interval": "5m",
+            "repeat_interval": "1h",
+            "receiver": receiver_name,
+        },
+        "receivers": [
+            {
+                "name": receiver_name,
+                "slack_configs": [
+                    {
+                        "api_url": httpserver.url_for("/"),
+                        "channel": "test",
+                        "text": r"https://localhost/alerts/{{ .GroupLabels.alertname }}",
+                    }
+                ],
+            }
+        ],
+    }
+
+    # use a template to define the slack callback id
+    atemplate = r'{{ define "slack.default.callbackid" }}2{{ end }}'
+    # set alertmanager configuration and template file
+    await ops_test.model.applications[app_name].set_config(
+        {"config_file": yaml.safe_dump(aconfig), "templates_file": atemplate}
+    )
+    await ops_test.model.wait_for_idle(apps=[app_name], status="active", timeout=60)
+
+    # create an alert
+    start_time = datetime.now(timezone.utc)
+    end_time = start_time + timedelta(minutes=5)
+    alert_name = "fake-alert"
+    model_uuid = "1234"
+    alerts = [
+        {
+            "startsAt": start_time.isoformat("T"),
+            "endsAt": end_time.isoformat("T"),
+            "status": "firing",
+            "annotations": {
+                "summary": "A fake alert",
+            },
+            "labels": {
+                "juju_model_uuid": model_uuid,
+                "juju_application": app_name,
+                "juju_model": ops_test.model_name,
+                "alertname": alert_name,
+            },
+            "generatorURL": f"http://localhost/{alert_name}",
+        }
+    ]
+
+    # define the expected slack notification for the alert
+    expected_notification = {
+        "channel": "test",
+        "username": "Alertmanager",
+        "attachments": [
+            {
+                "title": f"[FIRING:1] {alert_name} {app_name} {ops_test.model_name} {model_uuid} ",
+                "title_link": f"http://{app_name}-0:9093/#/alerts?receiver={receiver_name}",
+                "text": f"https://localhost/alerts/{alert_name}",
+                "fallback": f"[FIRING:1] {alert_name} {app_name} {ops_test.model_name} {model_uuid}  | "
+                f"http://{app_name}-0:9093/#/alerts?receiver={receiver_name}",
+                "callback_id": "2",
+                "footer": "",
+                "color": "danger",
+                "mrkdwn_in": ["fallback", "pretext", "text"],
+            }
+        ],
+    }
+
+    # set the alert
+    with httpserver.wait(timeout=120) as waiting:
+        # expect an alert to be forwarded to the receiver
+        httpserver.expect_oneshot_request(
+            "/", method="POST", json=expected_notification
+        ).respond_with_handler(request_handler)
+        client_address = await get_unit_address(ops_test, app_name, 0)
+        amanager = Alertmanager(address=client_address)
+        amanager.set_alerts(alerts)
+
+    # check receiver got an alert
+    assert waiting.result

tests/unit/test_alertmanager_client.py

@@ -4,6 +4,7 @@
 
 import json
 import unittest
+from datetime import datetime
 from unittest.mock import patch
 
 from alertmanager_client import Alertmanager, AlertmanagerBadResponse
@@ -59,3 +60,23 @@ def test_version(self, urlopen_mock):
         urlopen_mock.return_value.reason = "OK"
 
         self.assertEqual(self.api.version, "0.1.2")
+
+    @patch("alertmanager_client.urllib.request.urlopen")
+    def test_alerts_can_be_set(self, urlopen_mock):
+        msg = "HTTP 200 OK"
+        urlopen_mock.return_value = msg
+        alerts = [
+            {
+                "startsAt": datetime.now().isoformat("T"),
+                "status": "firing",
+                "annotations": {
+                    "summary": "A fake alert",
+                },
+                "labels": {
+                    "alertname": "fake alert",
+                },
+            }
+        ]
+        status = self.api.set_alerts(alerts)
+        urlopen_mock.assert_called()
+        self.assertEqual(status, msg)

tests/unit/test_charm.py

@@ -115,7 +115,7 @@ def test_templates_section_added_if_user_provided_templates(self, *unused):
         updated_config = yaml.safe_load(
             self.harness.charm.container.pull(self.harness.charm._config_path)
         )
-        self.assertEqual(updated_config["templates"], [f"'{self.harness.charm._templates_path}'"])
+        self.assertEqual(updated_config["templates"], [f"{self.harness.charm._templates_path}"])
 
 
 class TestWithoutInitialHooks(unittest.TestCase):

tox.ini

@@ -104,6 +104,7 @@ deps =
     pytest
     #git+https://github.com/charmed-kubernetes/pytest-operator.git
     pytest-operator
+    pytest-httpserver
 commands =
     pytest -v --tb native --log-cli-level=INFO -s {posargs} {toxinidir}/tests/integration