this repo has no description
more webhook support
+36
-3
+20
-2
netdata_zulip_bot/formatter.py
+20
-2
netdata_zulip_bot/formatter.py
···
2
2
3
3
from typing import Union
4
4
5
-
from .models import AlertNotification, ReachabilityNotification
5
+
from .models import AlertNotification, ReachabilityNotification, TestNotification
6
6
7
7
8
8
class ZulipMessageFormatter:
···
64
64
65
65
return topic, message
66
66
67
-
def format_notification(self, notification: Union[AlertNotification, ReachabilityNotification]) -> tuple[str, str]:
67
+
def format_test(self, notification: TestNotification) -> tuple[str, str]:
68
+
"""Format test notification for Zulip.
69
+
70
+
Returns:
71
+
Tuple of (topic, message_content)
72
+
"""
73
+
topic = "test"
74
+
75
+
message = f"""🧪 **Netdata Webhook Test**
76
+
77
+
{notification.message}
78
+
79
+
Your webhook integration is working correctly! ✅"""
80
+
81
+
return topic, message
82
+
83
+
def format_notification(self, notification: Union[AlertNotification, ReachabilityNotification, TestNotification]) -> tuple[str, str]:
68
84
"""Format any notification type for Zulip.
69
85
70
86
Returns:
···
74
90
return self.format_alert(notification)
75
91
elif isinstance(notification, ReachabilityNotification):
76
92
return self.format_reachability(notification)
93
+
elif isinstance(notification, TestNotification):
94
+
return self.format_test(notification)
77
95
else:
78
96
raise ValueError(f"Unknown notification type: {type(notification)}")
+16
-1
netdata_zulip_bot/models.py
+16
-1
netdata_zulip_bot/models.py
···
37
37
severity: AlertSeverity
38
38
date: datetime
39
39
alert_url: str
40
+
# Additional fields from full schema
41
+
Rooms: Optional[dict] = None
42
+
family: Optional[str] = None
43
+
class_: Optional[str] = Field(None, alias="class") # 'class' is a Python keyword
44
+
duration: Optional[str] = None
45
+
additional_active_critical_alerts: Optional[int] = None
46
+
additional_active_warning_alerts: Optional[int] = None
40
47
41
48
@field_validator('date', mode='before')
42
49
@classmethod
···
55
62
status: ReachabilityStatus
56
63
57
64
65
+
class TestNotification(BaseModel):
66
+
"""Test notification payload from Netdata Cloud."""
67
+
message: str
68
+
69
+
58
70
class WebhookPayload(BaseModel):
59
71
"""Union type for webhook payloads."""
60
72
61
73
@classmethod
62
-
def parse(cls, data: dict) -> Union[AlertNotification, ReachabilityNotification]:
74
+
def parse(cls, data: dict) -> Union[AlertNotification, ReachabilityNotification, TestNotification]:
63
75
"""Parse webhook payload and determine notification type."""
64
76
if 'alert' in data and 'chart' in data:
65
77
return AlertNotification(**data)
66
78
elif 'status' in data and 'host' in data:
67
79
return ReachabilityNotification(**data)
80
+
elif len(data) == 1 and 'message' in data:
81
+
# Test notification - only has a message field
82
+
return TestNotification(**data)
68
83
else:
69
84
raise ValueError(f"Unknown notification type: {data}")
70
85