Add itest for multi unit alerts

bundle.yaml.j2

@@ -49,14 +49,17 @@ applications:
   {%- if testing %}
   avalanche:
     charm: {{ avalanche|default('avalanche-k8s', true) }}
-    scale: 1
+    scale: 2
     trust: true
     {%- if avalanche is defined and avalanche.endswith('.charm') %}
     resources:
       avalanche-image: "quay.io/freshtracks.io/avalanche"
     {%- else %}
     channel: {{ channel|default('edge', true) }}
     {%- endif %}
+    options:
+      metric_count: 10
+      series_count: 2
   {%- endif %}
 
 relations:

tests/integration/helpers.py

@@ -8,10 +8,12 @@
 from pathlib import Path
 from typing import List
 
+from pytest_operator.plugin import OpsTest
+
 log = logging.getLogger(__name__)
 
 
-async def cli_deploy_bundle(ops_test, name: str, channel: str = "edge"):
+async def cli_deploy_bundle(ops_test: OpsTest, name: str, channel: str = "edge"):
     # use CLI to deploy bundle until https://github.com/juju/python-libjuju/issues/511 is fixed.
     run_args = [
         "juju",
@@ -30,13 +32,13 @@ async def cli_deploy_bundle(ops_test, name: str, channel: str = "edge"):
     await ops_test.model.wait_for_idle(timeout=1000)
 
 
-async def get_unit_address(ops_test, app_name: str, unit_num: int) -> str:
+async def get_unit_address(ops_test: OpsTest, app_name: str, unit_num: int) -> str:
     # return ops_test.model.applications[app_name].units[unit_num].data["private-address"]
     status = await ops_test.model.get_status()  # noqa: F821
     return status["applications"][app_name]["units"][f"{app_name}/{unit_num}"]["address"]
 
 
-async def get_alertmanager_alerts(ops_test, unit_name, unit_num, retries=3) -> List[dict]:
+async def get_alertmanager_alerts(ops_test: OpsTest, unit_name, unit_num, retries=3) -> List[dict]:
     """Get a list of alerts.
 
     Response looks like this:
@@ -75,7 +77,7 @@ async def get_alertmanager_alerts(ops_test, unit_name, unit_num, retries=3) -> L
     return alerts
 
 
-async def get_alertmanager_groups(ops_test, unit_name, unit_num, retries=3) -> List[dict]:
+async def get_alertmanager_groups(ops_test: OpsTest, unit_name, unit_num, retries=3) -> List[dict]:
     """Get a list of groups of alerts.
 
     Response looks like this:

tests/integration/test_bundle.py

@@ -20,13 +20,15 @@
 import urllib.request
 from pathlib import Path
 
+import juju
 import pytest
 from helpers import (
     cli_deploy_bundle,
     get_alertmanager_alerts,
     get_alertmanager_groups,
     get_unit_address,
 )
+from pytest_operator.plugin import OpsTest
 
 log = logging.getLogger(__name__)
 juju_topology_keys = {"juju_model_uuid", "juju_model", "juju_application"}
@@ -39,7 +41,7 @@ def get_this_script_dir() -> Path:
 
 
 @pytest.mark.abort_on_fail
-async def test_build_and_deploy(ops_test, pytestconfig):
+async def test_build_and_deploy(ops_test: OpsTest, pytestconfig):
     """Build the charm-under-test and deploy it together with related charms.
 
     Assert on the unit status before any relations/configurations take place.
@@ -92,7 +94,7 @@ async def build_charm_if_is_dir(option: str) -> str:
 
 
 @pytest.mark.abort_on_fail
-async def test_alertmanager_is_up(ops_test):
+async def test_alertmanager_is_up(ops_test: OpsTest):
     address = await get_unit_address(ops_test, "alertmanager", 0)
     url = f"http://{address}:9093"
     log.info("am public address: %s", url)
@@ -103,7 +105,7 @@ async def test_alertmanager_is_up(ops_test):
 
 
 @pytest.mark.abort_on_fail
-async def test_prometheus_is_up(ops_test):
+async def test_prometheus_is_up(ops_test: OpsTest):
     address = await get_unit_address(ops_test, "prometheus", 0)
     url = f"http://{address}:9090"
     log.info("prom public address: %s", url)
@@ -113,7 +115,7 @@ async def test_prometheus_is_up(ops_test):
 
 
 @pytest.mark.abort_on_fail
-async def test_prometheus_sees_alertmanager(ops_test):
+async def test_prometheus_sees_alertmanager(ops_test: OpsTest):
     am_address = await get_unit_address(ops_test, "alertmanager", 0)
     prom_address = await get_unit_address(ops_test, "prometheus", 0)
 
@@ -132,7 +134,7 @@ async def test_prometheus_sees_alertmanager(ops_test):
     )
 
 
-async def test_juju_topology_labels_in_alerts(ops_test):
+async def test_juju_topology_labels_in_alerts(ops_test: OpsTest):
     alerts = await get_alertmanager_alerts(ops_test, "alertmanager", 0, retries=100)
 
     i = -1
@@ -148,7 +150,7 @@ async def test_juju_topology_labels_in_alerts(ops_test):
     log.info("juju topology test passed for %s alerts", i + 1)
 
 
-async def test_alerts_are_grouped(ops_test):
+async def test_alerts_are_grouped(ops_test: OpsTest):
     groups = await get_alertmanager_groups(ops_test, "alertmanager", 0, retries=100)
     i = -1
     for i, group in enumerate(groups):
@@ -157,3 +159,20 @@ async def test_alerts_are_grouped(ops_test):
 
     assert i >= 0  # should have at least one group listed (the "AlwaysFiring" alarm rule)
     log.info("juju topology grouping test passed for %s groups", i + 1)
+
+
+async def test_alerts_are_fired_from_non_leader_units_too(ops_test: OpsTest):
+    """The list of alerts must include an "AlwaysFiring" alert from each avalanche unit."""
+
+    async def all_alerts_fire():
+        alerts = await get_alertmanager_alerts(ops_test, "alertmanager", 0, retries=100)
+        alerts = list(
+            filter(
+                lambda itm: itm["labels"]["alertname"] == "AlwaysFiringDueToNumericValue", alerts
+            )
+        )
+        units_firing = sorted([alert["labels"]["juju_unit"] for alert in alerts])
+        log.info("Units firing as of this moment: %s", units_firing)
+        return units_firing == ["avalanche/0", "avalanche/1"]
+
+    await juju.utils.block_until_with_coroutine(all_alerts_fire, timeout=300, wait_period=15)

tox.ini

@@ -67,8 +67,8 @@ commands =
 description = Run integration tests
 deps =
     jinja2
-    #git+https://github.com/juju/python-libjuju.git
-    juju
+    git+https://github.com/juju/python-libjuju.git
+    #juju
     pytest
     git+https://github.com/charmed-kubernetes/pytest-operator.git
 commands =