Update Channel.webhook_spec to return an object instead of a dict

This simplifies type annotations, as object's fields
can be type-annotated easily, and JSON->object parsing can be handled
by Pydantic.

hc/api/models.py

@@ -8,7 +8,7 @@
 from datetime import datetime
 from datetime import timedelta as td
 from datetime import timezone
-from typing import TypedDict
+from typing import Dict, TypedDict
 from urllib.parse import urlencode
 
 from cronsim import CronSim
@@ -19,6 +19,7 @@
 from django.urls import reverse
 from django.utils.functional import cached_property
 from django.utils.timezone import now
+from pydantic import BaseModel
 
 from hc.accounts.models import Project
 from hc.api import transports
@@ -684,6 +685,13 @@ def fget(instance):
     return property(fget)
 
 
+class WebhookSpec(BaseModel):
+    method: str
+    url: str
+    body: str
+    headers: Dict[str, str]
+
+
 class Channel(models.Model):
     name = models.CharField(max_length=100, blank=True)
     code = models.UUIDField(default=uuid.uuid4, editable=False, unique=True)
@@ -827,41 +835,26 @@ def po_priority(self):
         prio = int(parts[1])
         return PO_PRIORITIES[prio]
 
-    def webhook_spec(self, status: str) -> dict[str, str | dict[str, str]]:
+    def webhook_spec(self, status: str) -> WebhookSpec:
         assert self.kind == "webhook"
+        assert status in ("up", "down")
 
         doc = json.loads(self.value)
-        if status == "down":
-            return {
-                "method": doc["method_down"],
-                "url": doc["url_down"],
-                "body": doc["body_down"],
-                "headers": doc["headers_down"],
-            }
-        elif status == "up":
-            return {
-                "method": doc["method_up"],
-                "url": doc["url_up"],
-                "body": doc["body_up"],
-                "headers": doc["headers_up"],
-            }
+        return WebhookSpec(
+            method=doc[f"method_{status}"],
+            url=doc[f"url_{status}"],
+            body=doc[f"body_{status}"],
+            headers=doc[f"headers_{status}"],
+        )
 
     @property
-    def down_webhook_spec(self):
+    def down_webhook_spec(self) -> WebhookSpec:
         return self.webhook_spec("down")
 
     @property
-    def up_webhook_spec(self):
+    def up_webhook_spec(self) -> WebhookSpec:
         return self.webhook_spec("up")
 
-    @property
-    def url_down(self):
-        return self.down_webhook_spec["url"]
-
-    @property
-    def url_up(self):
-        return self.up_webhook_spec["url"]
-
     cmd_down = json_property("shell", "cmd_down")
     cmd_up = json_property("shell", "cmd_up")
 

hc/api/tests/test_channel_model.py

@@ -2,7 +2,7 @@
 
 import json
 
-from hc.api.models import Channel
+from hc.api.models import Channel, WebhookSpec
 from hc.test import BaseTestCase
 
 
@@ -24,22 +24,22 @@ def test_webhook_spec_handles_mixed(self) -> None:
 
         self.assertEqual(
             c.down_webhook_spec,
-            {
-                "method": "GET",
-                "url": "http://example.org",
-                "body": "",
-                "headers": {"X-Status": "X"},
-            },
+            WebhookSpec(
+                method="GET",
+                url="http://example.org",
+                body="",
+                headers={"X-Status": "X"},
+            ),
         )
 
         self.assertEqual(
             c.up_webhook_spec,
-            {
-                "method": "POST",
-                "url": "http://example.org/up/",
-                "body": "hello world",
-                "headers": {"X-Status": "OK"},
-            },
+            WebhookSpec(
+                method="POST",
+                url="http://example.org/up/",
+                body="hello world",
+                headers={"X-Status": "OK"},
+            ),
         )
 
     def test_it_handles_legacy_opsgenie_value(self) -> None:

hc/api/transports.py

@@ -309,10 +309,8 @@ def safe(s: str) -> str:
         return result
 
     def is_noop(self, check: Check) -> bool:
-        if check.status == "down" and not self.channel.url_down:
-            return True
-
-        if check.status == "up" and not self.channel.url_up:
+        spec = self.channel.webhook_spec(check.status)
+        if not spec.url:
             return True
 
         return False
@@ -322,20 +320,17 @@ def notify(self, check: Check, notification: Notification | None = None) -> None
             raise TransportError("Webhook notifications are not enabled.")
 
         spec = self.channel.webhook_spec(check.status)
-        if not spec["url"]:
+        if not spec.url:
             raise TransportError("Empty webhook URL")
 
-        assert isinstance(spec["url"], str)
-        url = self.prepare(spec["url"], check, urlencode=True)
+        url = self.prepare(spec.url, check, urlencode=True)
         headers = {}
-        assert isinstance(spec["headers"], dict)
-        for key, value in spec["headers"].items():
+        for key, value in spec.headers.items():
             # Header values should contain ASCII and latin-1 only
             headers[key] = self.prepare(value, check, latin1=True)
 
-        body, body_bytes = spec["body"], None
+        body, body_bytes = spec.body, None
         if body:
-            assert isinstance(body, str)
             body = self.prepare(body, check, allow_ping_body=True)
             body_bytes = body.encode()
 
@@ -344,11 +339,11 @@ def notify(self, check: Check, notification: Notification | None = None) -> None
         if notification and notification.owner is None:
             use_retries = False  # this is a test notification
 
-        if spec["method"] == "GET":
+        if spec.method == "GET":
             self.get(url, use_retries=use_retries, headers=headers)
-        elif spec["method"] == "POST":
+        elif spec.method == "POST":
             self.post(url, use_retries=use_retries, data=body_bytes, headers=headers)
-        elif spec["method"] == "PUT":
+        elif spec.method == "PUT":
             self.put(url, use_retries=use_retries, data=body_bytes, headers=headers)
 
 

hc/front/tests/test_add_webhook.py

@@ -50,8 +50,8 @@ def test_it_adds_two_webhook_urls_and_redirects(self) -> None:
 
         c = Channel.objects.get()
         self.assertEqual(c.project, self.project)
-        self.assertEqual(c.down_webhook_spec["url"], "http://foo.com")
-        self.assertEqual(c.up_webhook_spec["url"], "https://bar.com")
+        self.assertEqual(c.down_webhook_spec.url, "http://foo.com")
+        self.assertEqual(c.up_webhook_spec.url, "https://bar.com")
 
     def test_it_adds_webhook_using_team_access(self) -> None:
         form = {
@@ -68,8 +68,8 @@ def test_it_adds_webhook_using_team_access(self) -> None:
 
         c = Channel.objects.get()
         self.assertEqual(c.project, self.project)
-        self.assertEqual(c.down_webhook_spec["url"], "http://foo.com")
-        self.assertEqual(c.up_webhook_spec["url"], "https://bar.com")
+        self.assertEqual(c.down_webhook_spec.url, "http://foo.com")
+        self.assertEqual(c.up_webhook_spec.url, "https://bar.com")
 
     def test_it_accepts_good_urls(self) -> None:
         urls = [
@@ -127,8 +127,8 @@ def test_it_handles_empty_down_url(self) -> None:
         self.client.post(self.url, form)
 
         c = Channel.objects.get()
-        self.assertEqual(c.down_webhook_spec["url"], "")
-        self.assertEqual(c.up_webhook_spec["url"], "http://foo.com")
+        self.assertEqual(c.down_webhook_spec.url, "")
+        self.assertEqual(c.up_webhook_spec.url, "http://foo.com")
 
     def test_it_adds_request_body(self) -> None:
         form = {
@@ -143,7 +143,7 @@ def test_it_adds_request_body(self) -> None:
         self.assertRedirects(r, self.channels_url)
 
         c = Channel.objects.get()
-        self.assertEqual(c.down_webhook_spec["body"], "hello")
+        self.assertEqual(c.down_webhook_spec.body, "hello")
 
     def test_it_adds_headers(self) -> None:
         form = {
@@ -158,9 +158,7 @@ def test_it_adds_headers(self) -> None:
         self.assertRedirects(r, self.channels_url)
 
         c = Channel.objects.get()
-        self.assertEqual(
-            c.down_webhook_spec["headers"], {"test": "123", "test2": "abc"}
-        )
+        self.assertEqual(c.down_webhook_spec.headers, {"test": "123", "test2": "abc"})
 
     def test_it_rejects_bad_headers(self) -> None:
         self.client.login(username="alice@example.org", password="password")
@@ -201,7 +199,7 @@ def test_it_strips_headers(self) -> None:
         self.assertRedirects(r, self.channels_url)
 
         c = Channel.objects.get()
-        self.assertEqual(c.down_webhook_spec["headers"], {"test": "123"})
+        self.assertEqual(c.down_webhook_spec.headers, {"test": "123"})
 
     def test_it_rejects_both_empty(self) -> None:
         self.client.login(username="alice@example.org", password="password")

hc/front/tests/test_edit_webhook.py

@@ -67,16 +67,16 @@ def test_it_saves_form_and_redirects(self) -> None:
         self.assertEqual(self.channel.name, "Call foo.com / bar.com")
 
         down_spec = self.channel.down_webhook_spec
-        self.assertEqual(down_spec["method"], "POST")
-        self.assertEqual(down_spec["url"], "http://foo.com")
-        self.assertEqual(down_spec["body"], "going down")
-        self.assertEqual(down_spec["headers"], {"X-Foo": "1", "X-Bar": "2"})
+        self.assertEqual(down_spec.method, "POST")
+        self.assertEqual(down_spec.url, "http://foo.com")
+        self.assertEqual(down_spec.body, "going down")
+        self.assertEqual(down_spec.headers, {"X-Foo": "1", "X-Bar": "2"})
 
         up_spec = self.channel.up_webhook_spec
-        self.assertEqual(up_spec["method"], "POST")
-        self.assertEqual(up_spec["url"], "https://bar.com")
-        self.assertEqual(up_spec["body"], "going up")
-        self.assertEqual(up_spec["headers"], {"Content-Type": "text/plain"})
+        self.assertEqual(up_spec.method, "POST")
+        self.assertEqual(up_spec.url, "https://bar.com")
+        self.assertEqual(up_spec.body, "going up")
+        self.assertEqual(up_spec.headers, {"Content-Type": "text/plain"})
 
         # Make sure it does not call assign_all_checks
         self.assertFalse(self.channel.checks.exists())

templates/front/event_summary.html

@@ -11,7 +11,7 @@
         ({{ event.channel.name }})
     {% endif %}
 {% elif event.channel.kind == "webhook" %}
-    Called webhook {{ event.channel.url_down }}
+    Called webhook {{ event.channel.down_webhook_spec.url }}
 {% elif event.channel.kind == "msteams" %}
     Sent alert to Microsoft Teams
     {% if event.channel.name %}