Netdata.cloud bot for Zulip
anil.recoil.org
1"""Tests for webhook functionality."""
2
3import pytest
4from datetime import datetime
5from unittest.mock import Mock, patch
6
7from netdata_zulip_bot.models import AlertNotification, ReachabilityNotification, WebhookPayload
8from netdata_zulip_bot.formatter import ZulipMessageFormatter
9
10
11class TestWebhookPayload:
12 """Test webhook payload parsing."""
13
14 def test_parse_alert_notification(self):
15 """Test parsing alert notification payload."""
16 data = {
17 "message": "Critical alert on production server",
18 "alert": "high_cpu_usage",
19 "info": "CPU usage exceeded 90%",
20 "chart": "system.cpu",
21 "context": "cpu.utilization",
22 "space": "production",
23 "severity": "critical",
24 "date": "2024-01-15T14:30:00Z",
25 "alert_url": "https://app.netdata.cloud/spaces/abc/alerts/123"
26 }
27
28 notification = WebhookPayload.parse(data)
29 assert isinstance(notification, AlertNotification)
30 assert notification.severity.value == "critical"
31 assert notification.alert == "high_cpu_usage"
32 assert isinstance(notification.date, datetime)
33
34 def test_parse_reachability_notification(self):
35 """Test parsing reachability notification payload."""
36 data = {
37 "message": "Host web-server-01 is unreachable",
38 "url": "https://app.netdata.cloud/hosts/web-server-01",
39 "host": "web-server-01",
40 "severity": "critical",
41 "status": {
42 "reachable": False,
43 "text": "unreachable"
44 }
45 }
46
47 notification = WebhookPayload.parse(data)
48 assert isinstance(notification, ReachabilityNotification)
49 assert notification.severity.value == "critical"
50 assert notification.host == "web-server-01"
51 assert not notification.status.reachable
52
53 def test_parse_unknown_payload_raises_error(self):
54 """Test that unknown payload types raise ValueError."""
55 data = {"unknown_field": "value"}
56
57 with pytest.raises(ValueError, match="Unknown notification type"):
58 WebhookPayload.parse(data)
59
60
61class TestZulipMessageFormatter:
62 """Test Zulip message formatting."""
63
64 def setup_method(self):
65 """Set up test fixtures."""
66 self.formatter = ZulipMessageFormatter()
67
68 def test_format_critical_alert(self):
69 """Test formatting critical alert notification."""
70 alert = AlertNotification(
71 message="Critical alert on production server",
72 alert="high_cpu_usage",
73 info="CPU usage exceeded 90% for 5 minutes",
74 chart="system.cpu",
75 context="cpu.utilization",
76 space="production",
77 severity="critical",
78 date=datetime(2024, 1, 15, 14, 30, 0),
79 alert_url="https://app.netdata.cloud/spaces/abc/alerts/123"
80 )
81
82 topic, content = self.formatter.format_alert(alert)
83
84 assert topic == "critical"
85 assert "🔴" in content
86 assert "**high_cpu_usage**" in content
87 assert "production" in content
88 assert "Critical" in content
89 assert "2024-01-15 14:30:00 UTC" in content
90 assert "[View Alert](https://app.netdata.cloud/spaces/abc/alerts/123)" in content
91
92 def test_format_warning_alert(self):
93 """Test formatting warning alert notification."""
94 alert = AlertNotification(
95 message="Warning: High memory usage",
96 alert="high_memory_usage",
97 info="Memory usage at 85%",
98 chart="system.ram",
99 context="memory.utilization",
100 space="staging",
101 severity="warning",
102 date=datetime(2024, 1, 15, 10, 15, 0),
103 alert_url="https://app.netdata.cloud/spaces/def/alerts/456"
104 )
105
106 topic, content = self.formatter.format_alert(alert)
107
108 assert topic == "warning"
109 assert "⚠️" in content
110 assert "Warning" in content
111
112 def test_format_clear_alert(self):
113 """Test formatting clear alert notification."""
114 alert = AlertNotification(
115 message="Alert cleared: CPU usage normal",
116 alert="high_cpu_usage",
117 info="CPU usage returned to normal levels",
118 chart="system.cpu",
119 context="cpu.utilization",
120 space="production",
121 severity="clear",
122 date=datetime(2024, 1, 15, 15, 0, 0),
123 alert_url="https://app.netdata.cloud/spaces/abc/alerts/123"
124 )
125
126 topic, content = self.formatter.format_alert(alert)
127
128 assert topic == "clear"
129 assert "✅" in content
130 assert "Clear" in content
131
132 def test_format_reachability_unreachable(self):
133 """Test formatting unreachable host notification."""
134 notification = ReachabilityNotification(
135 message="Host web-server-01 is unreachable",
136 url="https://app.netdata.cloud/hosts/web-server-01",
137 host="web-server-01",
138 severity="critical",
139 status={
140 "reachable": False,
141 "text": "unreachable"
142 }
143 )
144
145 topic, content = self.formatter.format_reachability(notification)
146
147 assert topic == "reachability"
148 assert "🔴" in content # critical severity
149 assert "❌" in content # unreachable status
150 assert "web-server-01" in content
151 assert "Unreachable" in content
152 assert "[View Host](https://app.netdata.cloud/hosts/web-server-01)" in content
153
154 def test_format_reachability_reachable(self):
155 """Test formatting reachable host notification."""
156 notification = ReachabilityNotification(
157 message="Host web-server-01 is reachable again",
158 url="https://app.netdata.cloud/hosts/web-server-01",
159 host="web-server-01",
160 severity="info",
161 status={
162 "reachable": True,
163 "text": "reachable"
164 }
165 )
166
167 topic, content = self.formatter.format_reachability(notification)
168
169 assert topic == "reachability"
170 assert "ℹ️" in content # info severity
171 assert "✅" in content # reachable status
172 assert "Reachable" in content