diff --git a/README.md b/README.md
index 29cd036..46b2d71 100644
--- a/README.md
+++ b/README.md
@@ -2,20 +2,28 @@
# Alertbot
-This bot uses the webhook functionality of monitoring/alerting solutions like Grafana or Alertmanager to forward alerts to matrix rooms.
-This means that you no longer have to use E-Mail or Slack to receive alerts.
+A [maubot](https://github.com/maubot/maubot) plugin that receives webhooks from monitoring systems and forwards them as formatted messages to [Matrix](https://matrix.org) rooms.
+
+## Supported webhook types
+
+| Type | Detection | Source |
+|------|-----------|--------|
+| Alertmanager | `alerts[].labels.alertname` | Prometheus, custom rules, any Alertmanager source |
+| Grafana | `alerts[].labels.grafana_folder` | Grafana alerting |
+| Uptime Kuma | `heartbeat.status` | Uptime Kuma |
+| Slack webhook | `text` + `attachments` | Slack-format payloads |
## Getting Started
You can either [invite](#invite-the-official-instance) an official instance of the bot, or [self-host](#self-host-an-instance) it on your own matrix server.
-## Invite the Official Instance
+### Invite the Official Instance
* Create a Matrix room and invite @alertbot:hyteck.de
* Send `!url` to the room. The bot will answer with the webhook URL
* Put the Webhook URL into your monitoring solution (see below)
-## Self-host an Instance
+### Self-host an Instance
**Prerequisites:**
* A Matrix server where you have access to a maubot instance
@@ -35,7 +43,7 @@ cd alertbot
mbc build
```
-It's possible to upload the build to you maubot instance from the CLI. This is especially helpful when developing.
+It's possible to upload the build to your maubot instance from the CLI. This is especially helpful when developing.
First login to your instance, then add the `-u` flag to upload after build.
```shell
mbc login
@@ -44,30 +52,55 @@ mbc build -u
**Configure the Bot**
-Next, once the plugin is installed, set up a [client](https://docs.mau.fi/maubot/usage/basic.html#creating-clients) and then create an [instance](https://docs.mau.fi/maubot/usage/basic.html#creating-instances) which connects the client and plugin.
+Next, once the plugin is installed, set up a [client](https://docs.mau.fi/maubot/usage/basic.html#creating-clients) and then create an [instance](https://docs.mau.fi/maubot/usage/basic.html#creating-instances) which connects the client and plugin.
-Finally, invite the bot to an encrypted room where alerts should be sent and query the bot for the room id with the command `!roomid`.
+Finally, invite the bot to a Matrix room where alerts should be sent and query the bot for the room id with the command `!roomid`.
## Setup Alertmanager
-This configuration will send all your alerts to the room `!zOcbWjsWzdREnihgeC:example.com` (if the bot has access to it).
-Put in your own room-id (`!roomid`) behind the webhook base url (`!url`):
-```yaml
+This configuration will send all your alerts to a Matrix room. Put in your own room-id (`!roomid`) behind the webhook base url (`!url`):
+```yaml
receivers:
-- name: alertbot
- webhook_configs:
- - url: https://synapse.hyteck.de/_matrix/maubot/plugin/alertbot/webhook/!zOcbWjsWzdREnihgeC:example.com
+ - name: alertbot
+ webhook_configs:
+ - url: 'https://maubot.example.com/_matrix/maubot/plugin/alertbot/webhook/!roomid:example.com'
+ send_resolved: true
+
route:
- group_by:
- - alertname
- - cluster
- - service
- group_interval: 5m
- group_wait: 30s
receiver: alertbot
+ group_by: ['alertname']
+ group_wait: 30s
+ group_interval: 5m
repeat_interval: 3h
+```
+
+**Example messages:**
+
+Firing:
+```
+🔥 **InstanceDown** · critical
+
+webserver.example.com of job node_exporter has been down for more than 5 minutes.
+
+• **Instance:** webserver.example.com
+• **Job:** node_exporter
+• **Since:** 2024-03-30 08:00 UTC
+
+Source · Silence
+```
+
+Resolved:
+```
+✅ **InstanceDown** · resolved
+
+webserver.example.com of job node_exporter has been down for more than 5 minutes.
+
+• **Instance:** webserver.example.com
+• **Job:** node_exporter
+• **Duration:** 2h 15m · **Resolved:** 2024-03-30 10:15 UTC
+Source
```
## Setup Grafana
@@ -78,14 +111,19 @@ The grafana setup is fairly simple and can be used to forward grafana alerts to
## Send Test Alerts
-Use an example from the `alert_examples/` to test your setup
+Use an example from `alert_examples/` to test your setup:
+
```shell
curl --header "Content-Type: application/json" \
--request POST \
- --data "@alert_examples/prometheus.json" \
- https://webhook.example.com/_matrix/maubot/plugin/maubot/webhook/!zOcbWjsWzdREnihreC:example.com
+ --data "@alert_examples/prometheus_alert.json" \
+ https://maubot.example.com/_matrix/maubot/plugin/alertbot/webhook/!roomid:example.com
```
+## Dependencies
+
+- `python-dateutil>=2.8.2` (for timestamp parsing and duration calculation)
+
## Matrix Room
This project has a dedicated [chat room](https://matrix.to/#/#alertbot:hyteck.de)
diff --git a/alertbot.py b/alertbot.py
index 705e0f2..f8cf6b5 100644
--- a/alertbot.py
+++ b/alertbot.py
@@ -1,4 +1,5 @@
import json
+from urllib.parse import quote
import dateutil.parser
from aiohttp.web import Request, Response, json_response
@@ -58,10 +59,15 @@ def convert_slack_webhook_to_markdown(data):
def get_alert_type(data):
- """
- Currently supported are ["grafana-alert", "grafana-resolved", "prometheus-alert", "not-found"]
-
- :return: alert type
+ """Detect the type of incoming webhook payload.
+
+ Supported types:
+ - slack-webhook: Slack-format payload with text + attachments
+ - uptime-kuma-alert / uptime-kuma-resolved: Uptime Kuma heartbeat
+ - grafana-alert / grafana-resolved: Grafana alert with grafana_folder label
+ - alertmanager-alert / alertmanager-resolved: Any Alertmanager webhook
+ (includes Prometheus-originated alerts — the "job" label is NOT required)
+ - not-found: Unrecognized format
"""
if ("text" in data) and ("attachments" in data):
@@ -73,7 +79,7 @@ def get_alert_type(data):
return "uptime-kuma-alert"
elif data["heartbeat"]["status"] == 1:
return "uptime-kuma-resolved"
- except KeyError:
+ except (KeyError, TypeError):
pass
# Grafana
@@ -83,22 +89,160 @@ def get_alert_type(data):
return "grafana-alert"
else:
return "grafana-resolved"
- except KeyError:
+ except (KeyError, IndexError, TypeError):
pass
- # Prometheus
+ # Generic Alertmanager webhook: has alerts[] with labels.alertname.
+ # This covers Prometheus-originated alerts, custom alerts, and any other
+ # source that sends through Alertmanager. No "job" label required.
try:
- if data["alerts"][0]["labels"]["job"]:
+ if data["alerts"][0]["labels"]["alertname"]:
if data['status'] == "firing":
- return "prometheus-alert"
+ return "alertmanager-alert"
else:
- return "prometheus-resolved"
- except KeyError:
+ return "alertmanager-resolved"
+ except (KeyError, IndexError, TypeError):
pass
return "not-found"
+def _format_duration(start_str, end_str):
+ """Calculate human-readable duration between two ISO 8601 timestamps."""
+ try:
+ start = dateutil.parser.isoparse(start_str)
+ end = dateutil.parser.isoparse(end_str)
+ delta = end - start
+ total_seconds = int(delta.total_seconds())
+ if total_seconds < 0:
+ return None
+
+ days, remainder = divmod(total_seconds, 86400)
+ hours, remainder = divmod(remainder, 3600)
+ minutes, _ = divmod(remainder, 60)
+
+ parts = []
+ if days:
+ parts.append(f"{days}d")
+ if hours:
+ parts.append(f"{hours}h")
+ if minutes or not parts:
+ parts.append(f"{minutes}m")
+ return " ".join(parts)
+ except (ValueError, TypeError):
+ return None
+
+
+def _format_timestamp(iso_str):
+ """Format an ISO 8601 timestamp as 'YYYY-MM-DD HH:MM UTC'."""
+ try:
+ dt = dateutil.parser.isoparse(iso_str)
+ return dt.strftime("%Y-%m-%d %H:%M UTC")
+ except (ValueError, TypeError):
+ return iso_str
+
+
+def _build_silence_url(external_url, alertname):
+ """Build a pre-filled Alertmanager silence URL for a given alertname."""
+ if not external_url:
+ return None
+ base = external_url.rstrip("/")
+ filter_param = quote('{alertname="' + alertname + '"}')
+ return f"{base}/#/silences/new?filter={filter_param}"
+
+
+# Labels already shown in the title line — no need to repeat in metadata
+_SKIP_LABELS = {"alertname", "severity"}
+
+
+def alertmanager_to_markdown(alert_data: dict) -> list:
+ """Convert an Alertmanager webhook payload to formatted markdown messages.
+
+ Produces one message per alert in the payload. Each message follows
+ a structured format optimized for mobile notification clients:
+
+ 1. Status emoji + alert name + severity (= push notification text)
+ 2. Description
+ 3. Label metadata (one per line)
+ 4. Timestamps
+ 5. Source + silence links
+
+ Args:
+ alert_data: The full Alertmanager webhook payload dict.
+
+ Returns:
+ List of markdown-formatted message strings.
+ """
+ messages = []
+ external_url = alert_data.get("externalURL", "")
+
+ for alert in alert_data.get("alerts", []):
+ status = alert.get("status", "unknown")
+ labels = alert.get("labels", {})
+ annotations = alert.get("annotations", {})
+
+ alertname = labels.get("alertname", "Unknown")
+ severity = labels.get("severity", "")
+
+ # --- Title line ---
+ emoji = "\u2705" if status == "resolved" else "\U0001f525"
+ severity_text = f" \u00b7 {severity}" if severity and status != "resolved" else ""
+ status_text = " \u00b7 resolved" if status == "resolved" else ""
+ title = f"{emoji} **{alertname}**{severity_text}{status_text}"
+
+ # --- Description ---
+ description = annotations.get("description") or annotations.get("summary", "")
+
+ # --- Label metadata ---
+ # Use • character (not - markdown list marker) so CommonMark
+ # emits +
instead of
- . List markup renders
+ # with inconsistent indentation on Element X iOS.
+ meta_lines = []
+ for key, value in labels.items():
+ if key not in _SKIP_LABELS:
+ display_key = key.replace("_", " ").title()
+ meta_lines.append(f"• **{display_key}:** {value}")
+
+ # --- Timestamps ---
+ starts_at = alert.get("startsAt", "")
+ ends_at = alert.get("endsAt", "")
+
+ if status == "resolved":
+ time_parts = []
+ duration = _format_duration(starts_at, ends_at)
+ if duration:
+ time_parts.append(f"**Duration:** {duration}")
+ time_parts.append(f"**Resolved:** {_format_timestamp(ends_at)}")
+ meta_lines.append("• " + " \u00b7 ".join(time_parts))
+ elif starts_at:
+ meta_lines.append(f"• **Since:** {_format_timestamp(starts_at)}")
+
+ # --- Links ---
+ links = []
+ generator_url = alert.get("generatorURL", "")
+ if generator_url:
+ links.append(f"[Source]({generator_url})")
+ silence_url = _build_silence_url(external_url, alertname)
+ if silence_url and status != "resolved":
+ links.append(f"[Silence]({silence_url})")
+ link_line = " \u00b7 ".join(links)
+
+ # --- Assemble message ---
+ parts = [title, ""]
+ if description:
+ parts.append(description)
+ if meta_lines:
+ parts.append("")
+ parts.append(" \n".join(meta_lines))
+ if link_line:
+ parts.append("")
+ parts.append(link_line)
+
+ messages.append("\n".join(parts).strip())
+
+ return messages
+
+
def get_alert_messages(alert_data: dict, raw_mode=False) -> list:
"""
Returns a list of messages in markdown format
@@ -118,21 +262,17 @@ def get_alert_messages(alert_data: dict, raw_mode=False) -> list:
try:
if alert_type == "slack-webhook":
messages = convert_slack_webhook_to_markdown(alert_data)
- if alert_type == "grafana-alert":
+ elif alert_type in ("grafana-alert", "grafana-resolved"):
messages = grafana_alert_to_markdown(alert_data)
- elif alert_type == "grafana-resolved":
- messages = grafana_alert_to_markdown(alert_data)
- elif alert_type == "prometheus-alert":
- messages = prometheus_alert_to_markdown(alert_data)
- elif alert_type == "prometheus-resolved":
- messages = prometheus_alert_to_markdown(alert_data)
+ elif alert_type in ("alertmanager-alert", "alertmanager-resolved"):
+ messages = alertmanager_to_markdown(alert_data)
elif alert_type == "uptime-kuma-alert":
messages = uptime_kuma_alert_to_markdown(alert_data)
elif alert_type == "uptime-kuma-resolved":
messages = uptime_kuma_resolved_to_markdown(alert_data)
except KeyError as e:
messages = ["**Data received**\n```\n" + str(alert_data).strip(
- "\n").strip() + f"\n```\nThe data was detected as {alert_type} but was not in an expected format. If you want to help the development of this bot, file a bug report [here](https://github.com/moan0s/alertbot/issues)\n{e.with_traceback()}"]
+ "\n").strip() + f"\n```\nThe data was detected as {alert_type} but was not in an expected format. If you want to help the development of this bot, file a bug report [here](https://github.com/moan0s/alertbot/issues)\n{e}"]
return messages
@@ -213,35 +353,13 @@ def grafana_alert_to_markdown(alert_data: dict) -> list:
return messages
-def prometheus_alert_to_markdown(alert_data: dict) -> str:
- """
- Converts a prometheus alert json to markdown
-
- :param alert_data:
- :return: Alert as fomatted markdown
- """
- messages = []
- known_labels = ['alertname', 'instance', 'job']
- for alert in alert_data["alerts"]:
- title = alert['annotations']['description'] if hasattr(alert['annotations'], 'description') else \
- alert['annotations']['summary']
- message = f"""**{alert['status']}** {'💚' if alert['status'] == 'resolved' else '🔥'}: {title}"""
- for label_name in known_labels:
- try:
- message += "\n* **{0}**: {1}".format(label_name.capitalize(), alert["labels"][label_name])
- except:
- pass
- messages.append(message)
- return messages
-
-
class AlertBot(Plugin):
raw_mode = False
- async def send_alert(self, text, room):
+ async def send_alert(self, req, room):
+ text = await req.text()
self.log.info(text)
content = json.loads(text)
-
for message in get_alert_messages(content, self.raw_mode):
self.log.debug(f"Sending alert to {room}")
await self.client.send_markdown(room, message)
@@ -249,15 +367,14 @@ async def send_alert(self, text, room):
@web.post("/webhook/{room_id}")
async def webhook_room(self, req: Request) -> Response:
room_id = req.match_info["room_id"].strip()
- text = await req.text()
try:
- await self.send_alert(text, room=room_id)
+ await self.send_alert(req, room=room_id)
except MForbidden:
self.log.error(f"Could not send to {room_id}: Forbidden. Most likely the bot is not invited in the room.")
return json_response('{"status": "forbidden", "error": "forbidden"}', status=403)
- except json.decoder.JSONDecodeError:
- self.log.error(f"Decoding the data failed. {text}")
- return json_response({"status": "failed", "error": "JSON decoding failed", "data": text}, status=400)
+ except json.JSONDecodeError:
+ self.log.error(f"Decoding the data failed for {room_id}")
+ return json_response({"status": "failed", "error": "JSON decoding failed"}, status=400)
return json_response({"status": "ok"})
@command.new()
diff --git a/tests/test_alertbot.py b/tests/test_alertbot.py
index 6947af8..7d0b860 100644
--- a/tests/test_alertbot.py
+++ b/tests/test_alertbot.py
@@ -1,39 +1,244 @@
import unittest
-import alertbot
import json
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+import alertbot
+
+EXAMPLES_DIR = os.path.join(os.path.dirname(__file__), "..", "alert_examples")
+
+
+class TestAlertTypeClassification(unittest.TestCase):
+ """Test get_alert_type detection for all supported webhook formats."""
+
+ examples = [
+ ("grafana_alert.json", "grafana-alert"),
+ ("grafana_resolved.json", "grafana-resolved"),
+ ("uptime-kuma-503-alert.json", "uptime-kuma-alert"),
+ ("prometheus_alert.json", "alertmanager-alert"),
+ ("slack-webhook.json", "slack-webhook"),
+ ]
-examples = [{"name": "Grafana Alert",
- "filepath": "../alert_examples/grafana_alert.json",
- "expected_response": "",
- "type": "grafana-alert"},
- {"name": "Grafana Resolved",
- "filepath": "../alert_examples/grafana_resolved.json",
- "expected_response": "",
- "type": "grafana-resolved"},
- {"name": "Uptime Kuma 503 Alert",
- "filepath": "../alert_examples/uptime-kuma-503-alert.json",
- "expected_response": "",
- "type": "uptime-kuma-alert"},
- {"name": "Prometheus Alert",
- "filepath": "../alert_examples/prometheus_alert.json",
- "expected_response": "",
- "type": "prometheus-alert"},
- {"name": "Slack Alert",
- "filepath": "../alert_examples/slack-webhook.json",
- "expected_response": "",
- "type": "slack-webhook"},
- ]
-
-
-class ParseResponses(unittest.TestCase):
def test_classification(self):
- for example in examples:
- print(f"Example: {example['name']}")
- with open(example["filepath"]) as file:
- alert_data = json.load(file)
- found_type = alertbot.find_alert_type(alert_data)
- self.assertEqual(found_type, example["type"])
+ for filename, expected_type in self.examples:
+ with self.subTest(filename=filename):
+ with open(os.path.join(EXAMPLES_DIR, filename)) as f:
+ data = json.load(f)
+ self.assertEqual(alertbot.get_alert_type(data), expected_type)
+
+ def test_alertmanager_resolved(self):
+ with open(os.path.join(EXAMPLES_DIR, "prometheus_alert.json")) as f:
+ data = json.load(f)
+ data["status"] = "resolved"
+ data["alerts"][0]["status"] = "resolved"
+ self.assertEqual(alertbot.get_alert_type(data), "alertmanager-resolved")
+
+ def test_alertmanager_without_job_label(self):
+ """Alertmanager alerts that don't originate from Prometheus have no job label."""
+ data = {
+ "status": "firing",
+ "alerts": [{"labels": {"alertname": "CustomAlert", "severity": "warning"}}],
+ }
+ self.assertEqual(alertbot.get_alert_type(data), "alertmanager-alert")
+
+ def test_not_found(self):
+ self.assertEqual(alertbot.get_alert_type({}), "not-found")
+ self.assertEqual(alertbot.get_alert_type({"random": "data"}), "not-found")
+
+ def test_malformed_payloads(self):
+ """Malformed data should return not-found, not raise."""
+ cases = [
+ {"alerts": []},
+ {"alerts": [{"labels": {}}]},
+ {"alerts": "not a list"},
+ {"alerts": [None]},
+ {"heartbeat": "not a dict"},
+ ]
+ for data in cases:
+ with self.subTest(data=data):
+ result = alertbot.get_alert_type(data)
+ self.assertIsInstance(result, str)
+
+
+class TestAlertmanagerToMarkdown(unittest.TestCase):
+ """Test the rich Alertmanager formatter."""
+
+ def _make_payload(self, status="firing", alertname="TestAlert",
+ severity="warning", annotations=None, labels=None,
+ starts_at="2026-03-30T08:00:00Z",
+ ends_at="0001-01-01T00:00:00Z",
+ generator_url="", external_url=""):
+ alert_labels = {"alertname": alertname, "severity": severity}
+ if labels:
+ alert_labels.update(labels)
+ return {
+ "status": status,
+ "alerts": [{
+ "status": status,
+ "labels": alert_labels,
+ "annotations": annotations or {},
+ "startsAt": starts_at,
+ "endsAt": ends_at,
+ "generatorURL": generator_url,
+ }],
+ "externalURL": external_url,
+ }
+
+ def test_firing_alert_basic(self):
+ payload = self._make_payload(
+ annotations={"description": "Something is on fire"},
+ )
+ messages = alertbot.alertmanager_to_markdown(payload)
+ self.assertEqual(len(messages), 1)
+ msg = messages[0]
+ self.assertIn("\U0001f525", msg)
+ self.assertIn("**TestAlert**", msg)
+ self.assertIn("warning", msg)
+ self.assertIn("Something is on fire", msg)
+
+ def test_resolved_alert_shows_duration(self):
+ payload = self._make_payload(
+ status="resolved",
+ starts_at="2026-03-30T08:00:00Z",
+ ends_at="2026-03-30T10:15:00Z",
+ )
+ messages = alertbot.alertmanager_to_markdown(payload)
+ msg = messages[0]
+ self.assertIn("\u2705", msg)
+ self.assertIn("resolved", msg)
+ self.assertIn("2h 15m", msg)
+
+ def test_description_preferred_over_summary(self):
+ payload = self._make_payload(
+ annotations={"description": "Detailed info", "summary": "Short info"},
+ )
+ msg = alertbot.alertmanager_to_markdown(payload)[0]
+ self.assertIn("Detailed info", msg)
+
+ def test_summary_fallback(self):
+ payload = self._make_payload(
+ annotations={"summary": "Short info"},
+ )
+ msg = alertbot.alertmanager_to_markdown(payload)[0]
+ self.assertIn("Short info", msg)
+
+ def test_extra_labels_shown(self):
+ payload = self._make_payload(
+ labels={"instance": "server-01:9090", "job": "node"},
+ )
+ msg = alertbot.alertmanager_to_markdown(payload)[0]
+ self.assertIn("server-01:9090", msg)
+ self.assertIn("node", msg)
+ # alertname and severity should NOT appear in metadata
+ self.assertNotIn("\u2022 **Alertname", msg)
+ self.assertNotIn("\u2022 **Severity", msg)
+
+ def test_silence_url(self):
+ payload = self._make_payload(
+ external_url="https://alertmanager.example.com",
+ )
+ msg = alertbot.alertmanager_to_markdown(payload)[0]
+ self.assertIn("[Silence]", msg)
+ self.assertIn("alertmanager.example.com", msg)
+
+ def test_no_silence_url_when_resolved(self):
+ payload = self._make_payload(
+ status="resolved",
+ external_url="https://alertmanager.example.com",
+ ends_at="2026-03-30T10:00:00Z",
+ )
+ msg = alertbot.alertmanager_to_markdown(payload)[0]
+ self.assertNotIn("[Silence]", msg)
+
+ def test_source_link(self):
+ payload = self._make_payload(
+ generator_url="http://prometheus:9090/graph?g0.expr=up",
+ )
+ msg = alertbot.alertmanager_to_markdown(payload)[0]
+ self.assertIn("[Source]", msg)
+
+ def test_multiple_alerts_in_group(self):
+ payload = self._make_payload()
+ payload["alerts"].append({
+ "status": "firing",
+ "labels": {"alertname": "SecondAlert", "severity": "critical"},
+ "annotations": {},
+ "startsAt": "2026-03-30T09:00:00Z",
+ "endsAt": "0001-01-01T00:00:00Z",
+ "generatorURL": "",
+ })
+ messages = alertbot.alertmanager_to_markdown(payload)
+ self.assertEqual(len(messages), 2)
+ self.assertIn("TestAlert", messages[0])
+ self.assertIn("SecondAlert", messages[1])
+
+ def test_real_prometheus_example(self):
+ with open(os.path.join(EXAMPLES_DIR, "prometheus_alert.json")) as f:
+ data = json.load(f)
+ messages = alertbot.alertmanager_to_markdown(data)
+ self.assertEqual(len(messages), 1)
+ msg = messages[0]
+ self.assertIn("InstanceDown", msg)
+ self.assertIn("critical", msg)
+ self.assertIn("webserver.example.com", msg)
+
+
+class TestFormatDuration(unittest.TestCase):
+
+ def test_hours_and_minutes(self):
+ self.assertEqual(
+ alertbot._format_duration("2026-03-30T08:00:00Z", "2026-03-30T10:15:00Z"),
+ "2h 15m",
+ )
+
+ def test_days(self):
+ result = alertbot._format_duration("2026-03-30T00:00:00Z", "2026-04-01T02:30:00Z")
+ self.assertEqual(result, "2d 2h 30m")
+
+ def test_zero_duration(self):
+ self.assertEqual(
+ alertbot._format_duration("2026-03-30T08:00:00Z", "2026-03-30T08:00:00Z"),
+ "0m",
+ )
+
+ def test_negative_duration(self):
+ self.assertIsNone(
+ alertbot._format_duration("2026-03-30T10:00:00Z", "2026-03-30T08:00:00Z")
+ )
+
+ def test_invalid_input(self):
+ self.assertIsNone(alertbot._format_duration("not-a-date", "also-not"))
+
+
+class TestFormatTimestamp(unittest.TestCase):
+
+ def test_iso_format(self):
+ self.assertEqual(
+ alertbot._format_timestamp("2026-03-30T08:00:00Z"),
+ "2026-03-30 08:00 UTC",
+ )
+
+ def test_invalid_returns_input(self):
+ self.assertEqual(alertbot._format_timestamp("garbage"), "garbage")
+
+
+class TestBuildSilenceUrl(unittest.TestCase):
+
+ def test_basic(self):
+ url = alertbot._build_silence_url("https://am.example.com", "HighCPU")
+ self.assertIn("am.example.com", url)
+ self.assertIn("silences/new", url)
+ self.assertIn("HighCPU", url)
+
+ def test_trailing_slash_stripped(self):
+ url = alertbot._build_silence_url("https://am.example.com/", "Test")
+ self.assertNotIn("//", url.split("://", 1)[1])
+
+ def test_no_external_url(self):
+ self.assertIsNone(alertbot._build_silence_url("", "Test"))
+ self.assertIsNone(alertbot._build_silence_url(None, "Test"))
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main()