archive/healthchecks_healthchecks
0
0
Fork 0
mirror of https://github.com/healthchecks/healthchecks.git synced 2025-04-08 06:30:05 +00:00
Code Issues Projects Releases Wiki Activity

Change Channel.notify() signature to take Flip object as an argument

Browse source 
... and pass it to Transport.notify_flip().

This allows us to pass flip-specific information (the flip timestamp,
the new status) to transport classes.
This commit is contained in:
Pēteris Caune 2024-04-12 13:54:16 +03:00
parent 6e130f1749
commit 28fdfd1362
No known key found for this signature in database
GPG key ID: E28D7679E9A9EDE2
35 changed files with 467 additions and 308 deletions

10
hc/accounts/management/commands/senddeletionscheduled.py
View file

@ -11,7 +11,7 @@ from django.db.models import QuerySet
from django.utils.timezone import now
from hc.accounts.models import Profile
from hc.api.models import Channel, Check
from hc.api.models import Channel, Check, Flip
from hc.lib import emails
@ -51,12 +51,16 @@ class Command(BaseCommand):
dummy.last_ping = now() - td(days=1)
dummy.n_pings = 1
dummy_flip = Flip(owner=dummy)
dummy_flip.old_status = "up"
dummy_flip.new_status = "down"
self.stdout.write(f" * Sending notification to {channel.kind}")
error = channel.notify(dummy, is_test=True)
error = channel.notify(dummy_flip, is_test=True)
if error == "no-op":
# This channel may be configured to send "up" notifications only.
dummy.status = "up"
error = channel.notify(dummy, is_test=True)
error = channel.notify(dummy_flip, is_test=True)
if error:
self.stdout.write(f" Error sending notification: {error}")

2
hc/api/management/commands/sendalerts.py
View file

@ -39,7 +39,7 @@ def notify(flip_id: int, stdout: TextIOBase) -> None:
send_start = now()
for ch in channels:
notify_start = time.time()
error = ch.notify(check)
error = ch.notify(flip)
secs = time.time() - notify_start
label = "ERR" if error else "OK"
s = " * %-3s %4.1fs %-10s %s %s\n" % (label, secs, ch.kind, ch.code, error)

11
hc/api/models.py
View file

@ -930,8 +930,8 @@ class Channel(models.Model):
_, cls = TRANSPORTS[self.kind]
return cls(self)
def notify(self, check: Check, is_test: bool = False) -> str:
if self.transport.is_noop(check.status):
def notify(self, flip: "Flip", is_test: bool = False) -> str:
if self.transport.is_noop(flip.new_status):
return "no-op"
n = Notification(channel=self)
@ -940,15 +940,16 @@ class Channel(models.Model):
# (the passed check is a dummy, unsaved Check instance)
pass
else:
n.owner = check
n.owner = flip.owner
n.check_status = check.status
n.check_status = flip.new_status
n.error = "Sending"
n.save()
start, error, disabled = now(), "", self.disabled
try:
self.transport.notify(check, notification=n)
self.transport.notify_flip(flip, notification=n)
except transports.TransportError as e:
disabled = True if e.permanent else disabled
error = e.message

21
hc/api/tests/test_notify.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -29,11 +29,16 @@ class NotifyTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = status
@patch("hc.api.transports.curl.request", autospec=True)
def test_pagerteam(self, mock_post: Mock) -> None:
self._setup_data("pagerteam", "123")
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_not_called()
self.assertEqual(Notification.objects.count(), 0)
@ -41,7 +46,7 @@ class NotifyTestCase(BaseTestCase):
def test_hipchat(self, mock_post: Mock) -> None:
self._setup_data("hipchat", "123")
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_not_called()
self.assertEqual(Notification.objects.count(), 0)
@ -50,7 +55,7 @@ class NotifyTestCase(BaseTestCase):
self.channel.kind = "invalid"
with self.assertRaises(NotImplementedError):
self.channel.notify(self.check)
self.channel.notify(self.flip)
@patch("hc.api.transports.os.system")
@override_settings(SHELL_ENABLED=True)
@ -59,7 +64,7 @@ class NotifyTestCase(BaseTestCase):
self._setup_data("shell", json.dumps(definition))
mock_system.return_value = 0
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_system.assert_called_with("logger hello")
@patch("hc.api.transports.os.system")
@ -69,7 +74,7 @@ class NotifyTestCase(BaseTestCase):
self._setup_data("shell", json.dumps(definition))
mock_system.return_value = 123
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Command returned exit code 123")
@ -83,7 +88,7 @@ class NotifyTestCase(BaseTestCase):
self.check.name = "Database"
self.check.tags = "foo bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_system.assert_called_with("logger Database is down (foo)")
@ -93,7 +98,7 @@ class NotifyTestCase(BaseTestCase):
definition = {"cmd_down": "logger hello", "cmd_up": ""}
self._setup_data("shell", json.dumps(definition))
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_system.assert_not_called()
n = Notification.objects.get()

13
hc/api/tests/test_notify_apprise.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
try:
@ -37,6 +37,11 @@ class NotifyAppriseTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("apprise.Apprise")
@override_settings(APPRISE_ENABLED=True)
def test_it_works(self, mock_apprise: Mock) -> None:
@ -44,7 +49,7 @@ class NotifyAppriseTestCase(BaseTestCase):
mock_aobj.add.return_value = True
mock_aobj.notify.return_value = True
mock_apprise.return_value = mock_aobj
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 1)
body = mock_apprise.return_value.notify.call_args.kwargs["body"]
@ -54,7 +59,7 @@ class NotifyAppriseTestCase(BaseTestCase):
@patch("apprise.Apprise")
@override_settings(APPRISE_ENABLED=False)
def test_apprise_disabled(self, mock_apprise: Mock) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Apprise is disabled and/or not installed")
@ -69,7 +74,7 @@ class NotifyAppriseTestCase(BaseTestCase):
mock_aobj.add.return_value = True
mock_aobj.notify.return_value = True
mock_apprise.return_value = mock_aobj
self.channel.notify(self.check)
self.channel.notify(self.flip)
body = mock_apprise.return_value.notify.call_args.kwargs["body"]
self.assertIn("Foo is DOWN", body)

17
hc/api/tests/test_notify_call.py
View file

@ -10,7 +10,7 @@ from django.core import mail
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -30,6 +30,11 @@ class NotifyCallTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_call(self, mock_post: Mock) -> None:
self.profile.call_limit = 1
@ -37,7 +42,7 @@ class NotifyCallTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["To"], "+1234567890")
@ -48,7 +53,7 @@ class NotifyCallTestCase(BaseTestCase):
@override_settings(TWILIO_ACCOUNT=None)
def test_it_requires_twilio_configuration(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Call notifications are not enabled")
@ -60,7 +65,7 @@ class NotifyCallTestCase(BaseTestCase):
self.profile.calls_sent = 50
self.profile.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_not_called()
n = Notification.objects.get()
@ -83,7 +88,7 @@ class NotifyCallTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_called_once()
@override_settings(TWILIO_FROM="+000")
@ -97,7 +102,7 @@ class NotifyCallTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b"""{"code": 21211}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
# Make sure the HTTP request was made only once (no retries):
self.channel.refresh_from_db()

13
hc/api/tests/test_notify_discord.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.test import BaseTestCase
@ -29,13 +29,18 @@ class NotifyDiscordTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = status
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
v = json.dumps({"webhook": {"url": "https://example.org"}})
self._setup_data(v)
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
url = mock_post.call_args.args[1]
@ -51,7 +56,7 @@ class NotifyDiscordTestCase(BaseTestCase):
self._setup_data(v)
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
url = mock_post.call_args.args[1]
@ -68,7 +73,7 @@ class NotifyDiscordTestCase(BaseTestCase):
self.ping.kind = "fail"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]

41
hc/api/tests/test_notify_email.py
View file

@ -12,7 +12,7 @@ from django.core.mail import EmailMessage, EmailMultiAlternatives
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.test import BaseTestCase
@ -42,6 +42,11 @@ class NotifyEmailTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
def get_html(self, email: EmailMessage) -> str:
assert isinstance(email, EmailMultiAlternatives)
html, _ = email.alternatives[0]
@ -50,7 +55,7 @@ class NotifyEmailTestCase(BaseTestCase):
@override_settings(DEFAULT_FROM_EMAIL="alerts@example.org")
def test_it_works(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -107,7 +112,7 @@ class NotifyEmailTestCase(BaseTestCase):
@override_settings(DEFAULT_FROM_EMAIL='"Alerts" <alerts@example.org>')
def test_it_message_id_generation_handles_angle_brackets(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
self.assertTrue(email.extra_headers["Message-ID"].endswith("@example.org>"))
@ -121,7 +126,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.ping.body_raw = None
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
html = self.get_html(mail.outbox[0])
self.assertIn("Line 1<br>Line2", html)
@ -136,7 +141,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
html = self.get_html(email)
@ -149,7 +154,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.ping.body = "X" * 10000 + ", and the rest gets cut off"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
html = self.get_html(email)
@ -160,7 +165,7 @@ class NotifyEmailTestCase(BaseTestCase):
def test_it_handles_missing_ping_object(self) -> None:
self.ping.delete()
self.channel.notify(self.check)
self.channel.notify(self.flip)
html = self.get_html(mail.outbox[0])
self.assertIn("Daily Backup", html)
@ -169,7 +174,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.channel.value = "alice+notifications@example.org"
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
self.assertEqual(email.to[0], "alice+notifications@example.org")
@ -185,7 +190,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.channel.value = json.dumps(payload)
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
# And email should have been sent
self.assertEqual(len(mail.outbox), 1)
@ -197,7 +202,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.channel.email_verified = False
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
# If an email is not verified, it should say so in the notification:
n = Notification.objects.get()
@ -208,7 +213,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.channel.value = json.dumps(payload)
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
# This channel should not notify on "down" events:
self.assertEqual(Notification.objects.count(), 0)
@ -218,7 +223,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.check.name = "Foo & Bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
self.assertEqual(email.subject, "DOWN | Foo & Bar")
@ -233,7 +238,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.ping.save()
with patch("hc.api.transports.time.sleep"):
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
html = self.get_html(email)
@ -245,7 +250,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
html = self.get_html(email)
@ -256,7 +261,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.ping.kind = "log"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
html = self.get_html(email)
@ -268,7 +273,7 @@ class NotifyEmailTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
html = self.get_html(email)
@ -277,7 +282,7 @@ class NotifyEmailTestCase(BaseTestCase):
@override_settings(EMAIL_MAIL_FROM_TMPL="%s@bounces.example.org")
def test_it_sets_custom_mail_from(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
self.assertTrue(email.from_email.startswith("n."))
@ -298,7 +303,7 @@ tempor incididunt ut labore et dolore magna aliqua.
"""
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
email = mail.outbox[0]
self.assertIn("Lorem ipsum", email.body)

23
hc/api/tests/test_notify_gotify.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -28,11 +28,16 @@ class NotifyGotidyTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
method, url = mock_post.call_args.args
@ -50,7 +55,7 @@ class NotifyGotidyTestCase(BaseTestCase):
{"url": "https://example.org/sub/", "token": "abc"}
)
self.channel.notify(self.check)
self.channel.notify(self.flip)
method, url = mock_post.call_args.args
self.assertEqual(url, "https://example.org/sub/message?token=abc")
@ -62,7 +67,7 @@ class NotifyGotidyTestCase(BaseTestCase):
{"url": "https://example.org/sub", "token": "abc"}
)
self.channel.notify(self.check)
self.channel.notify(self.flip)
method, url = mock_post.call_args.args
self.assertEqual(url, "https://example.org/sub/message?token=abc")
@ -77,7 +82,7 @@ class NotifyGotidyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("All the other checks are up.", payload["message"])
@ -92,7 +97,7 @@ class NotifyGotidyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("The following checks are also down", payload["message"])
@ -106,7 +111,7 @@ class NotifyGotidyTestCase(BaseTestCase):
Check.objects.create(project=self.project, status="down")
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("(last ping: never)", payload["message"])
@ -122,7 +127,7 @@ class NotifyGotidyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertNotIn("Foobar", payload["message"])
@ -134,7 +139,7 @@ class NotifyGotidyTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertNotIn("Last ping was", payload["message"])

15
hc/api/tests/test_notify_group.py
View file

@ -9,7 +9,7 @@ from django.core import mail
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -35,8 +35,13 @@ class NotifyGroupTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
def test_it_works(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertEqual(self.channel.last_error, "")
@ -57,7 +62,7 @@ class NotifyGroupTestCase(BaseTestCase):
)
self.channel_email.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertEqual(self.channel.last_error, "")
@ -72,7 +77,7 @@ class NotifyGroupTestCase(BaseTestCase):
)
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
assert self.channel.last_error == ""
@ -86,7 +91,7 @@ class NotifyGroupTestCase(BaseTestCase):
self.channel_email.email_verified = False
self.channel_email.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
assert self.channel.last_error == "1 out of 1 notifications failed"

13
hc/api/tests/test_notify_line.py
View file

@ -7,7 +7,7 @@ from unittest.mock import Mock, patch
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -27,11 +27,16 @@ class NotifyLineTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
headers = mock_post.call_args.kwargs["headers"]
@ -47,7 +52,7 @@ class NotifyLineTestCase(BaseTestCase):
self.check.status = "up"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
params = mock_post.call_args.kwargs["params"]
self.assertEqual(params["message"], 'The check "Foo & Bar" is now UP.')
@ -58,7 +63,7 @@ class NotifyLineTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
params = mock_post.call_args.kwargs["params"]
self.assertNotIn("Last ping was", params["message"])

11
hc/api/tests/test_notify_matrix.py
View file

@ -9,7 +9,7 @@ from urllib.parse import quote
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -32,11 +32,16 @@ class NotifyMatrixTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
method, url = mock_post.call_args.args
@ -55,7 +60,7 @@ class NotifyMatrixTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]

19
hc/api/tests/test_notify_mattermost.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.test import BaseTestCase
@ -31,11 +31,16 @@ class NotifyMattermostTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
url = mock_post.call_args.args[1]
@ -47,7 +52,7 @@ class NotifyMattermostTestCase(BaseTestCase):
@override_settings(MATTERMOST_ENABLED=False)
def test_it_requires_mattermost_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Mattermost notifications are not enabled.")
@ -56,7 +61,7 @@ class NotifyMattermostTestCase(BaseTestCase):
def test_it_does_not_disable_channel_on_404(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 404
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertFalse(self.channel.disabled)
@ -68,7 +73,7 @@ class NotifyMattermostTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -83,7 +88,7 @@ class NotifyMattermostTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World" * 1000
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -98,7 +103,7 @@ class NotifyMattermostTestCase(BaseTestCase):
self.ping.body_raw = b"Hello ``` World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]

27
hc/api/tests/test_notify_msteams.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.test import BaseTestCase
@ -32,13 +32,18 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.check.name = "_underscores_ & more"
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -63,7 +68,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
facts = {f["name"]: f["value"] for f in payload["sections"][0]["facts"]}
self.assertEqual(facts["Schedule:"], "\u034f* \u034f* \u034f* \u034f* \u034f*")
@ -76,7 +81,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.check.kind = "cron"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
facts = {f["name"]: f["value"] for f in payload["sections"][0]["facts"]}
@ -84,7 +89,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
@override_settings(MSTEAMS_ENABLED=False)
def test_it_requires_msteams_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "MS Teams notifications are not enabled.")
@ -96,7 +101,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.ping.kind = "fail"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -110,7 +115,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.ping.kind = "log"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
facts = {f["name"]: f["value"] for f in payload["sections"][0]["facts"]}
@ -124,7 +129,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -138,7 +143,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -152,7 +157,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World" * 1000
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -166,7 +171,7 @@ class NotifyMsTeamsTestCase(BaseTestCase):
self.ping.body_raw = b"Hello ``` World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]

25
hc/api/tests/test_notify_ntfy.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.test import BaseTestCase
@ -42,11 +42,16 @@ class NotifyNtfyTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -68,7 +73,7 @@ class NotifyNtfyTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("Last Ping: Exit status 123", payload["message"])
@ -90,7 +95,7 @@ class NotifyNtfyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(payload["title"], "<Name> is DOWN")
@ -105,7 +110,7 @@ class NotifyNtfyTestCase(BaseTestCase):
self.check.kind = "cron"
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("Schedule: * * * * *", payload["message"])
@ -121,7 +126,7 @@ class NotifyNtfyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("All the other checks are up.", payload["message"])
@ -136,7 +141,7 @@ class NotifyNtfyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("The following checks are also down", payload["message"])
@ -149,7 +154,7 @@ class NotifyNtfyTestCase(BaseTestCase):
Check.objects.create(project=self.project, status="down")
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("(last ping: never)", payload["message"])
@ -165,7 +170,7 @@ class NotifyNtfyTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertNotIn("Foobar", payload["message"])
@ -186,7 +191,7 @@ class NotifyNtfyTestCase(BaseTestCase):
)
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
headers = mock_post.call_args.kwargs["headers"]

23
hc/api/tests/test_notify_opsgenie.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -30,12 +30,17 @@ class NotifyOpsgenieTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
self._setup_data(json.dumps({"key": "123", "region": "us"}))
mock_post.return_value.status_code = 202
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -50,7 +55,7 @@ class NotifyOpsgenieTestCase(BaseTestCase):
self._setup_data(json.dumps({"key": "123", "region": "us"}))
mock_post.return_value.status_code = 202
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -65,7 +70,7 @@ class NotifyOpsgenieTestCase(BaseTestCase):
self._setup_data(json.dumps({"key": "123", "region": "us"}), status="up")
mock_post.return_value.status_code = 202
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -78,7 +83,7 @@ class NotifyOpsgenieTestCase(BaseTestCase):
self._setup_data(json.dumps({"key": "456", "region": "eu"}))
mock_post.return_value.status_code = 202
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -92,7 +97,7 @@ class NotifyOpsgenieTestCase(BaseTestCase):
mock_post.return_value.status_code = 403
mock_post.return_value.content = b"""{"message": "Nice try"}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, 'Received status code 403 with a message: "Nice try"')
@ -102,14 +107,14 @@ class NotifyOpsgenieTestCase(BaseTestCase):
mock_post.return_value.status_code = 403
mock_post.return_value.json = Mock(side_effect=ValueError)
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 403")
@override_settings(OPSGENIE_ENABLED=False)
def test_it_requires_opsgenie_enabled(self) -> None:
self._setup_data(json.dumps({"key": "123", "region": "us"}))
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Opsgenie notifications are not enabled.")
@ -121,7 +126,7 @@ class NotifyOpsgenieTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 202
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")

15
hc/api/tests/test_notify_pagertree.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -28,11 +28,16 @@ class NotifyPagertreeTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -42,7 +47,7 @@ class NotifyPagertreeTestCase(BaseTestCase):
@override_settings(PAGERTREE_ENABLED=False)
def test_it_requires_pagertree_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "PagerTree notifications are not enabled.")
@ -54,7 +59,7 @@ class NotifyPagertreeTestCase(BaseTestCase):
self.check.name = "Foo & Bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(payload["title"], "Foo & Bar is DOWN")
@ -65,7 +70,7 @@ class NotifyPagertreeTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertNotIn("Last ping was", payload["description"])

17
hc/api/tests/test_notify_pd.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -31,12 +31,17 @@ class NotifyPdTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = status
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
self._setup_data("123")
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -53,7 +58,7 @@ class NotifyPdTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(payload["details"]["Schedule"], "* * * * *")
self.assertEqual(payload["details"]["Time zone"], "Europe/Riga")
@ -63,7 +68,7 @@ class NotifyPdTestCase(BaseTestCase):
self._setup_data(json.dumps({"service_key": "456"}))
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -73,7 +78,7 @@ class NotifyPdTestCase(BaseTestCase):
@override_settings(PD_ENABLED=False)
def test_it_requires_pd_enabled(self) -> None:
self._setup_data("123")
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "PagerDuty notifications are not enabled.")
@ -86,7 +91,7 @@ class NotifyPdTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(payload["description"], "Foo & Bar is DOWN")

15
hc/api/tests/test_notify_pushbullet.py
View file

@ -7,7 +7,7 @@ from unittest.mock import Mock, patch
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -27,13 +27,18 @@ class NotifyPushbulletTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
self.check.status = "down"
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
_, kwargs = mock_post.call_args
@ -47,7 +52,7 @@ class NotifyPushbulletTestCase(BaseTestCase):
def test_it_handles_up(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
_, kwargs = mock_post.call_args
@ -63,7 +68,7 @@ class NotifyPushbulletTestCase(BaseTestCase):
self.check.name = "Foo & Bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
_, kwargs = mock_post.call_args
self.assertEqual(
@ -78,7 +83,7 @@ class NotifyPushbulletTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
_, kwargs = mock_post.call_args
payload = kwargs["json"]

37
hc/api/tests/test_notify_pushover.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, TokenBucket
from hc.api.models import Channel, Check, Flip, Notification, TokenBucket
from hc.test import BaseTestCase
API = "https://api.pushover.net/1"
@ -32,12 +32,17 @@ class NotifyPushoverTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = status
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
self._setup_data("123|0")
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 1)
url = mock_post.call_args.args[1]
@ -54,7 +59,7 @@ class NotifyPushoverTestCase(BaseTestCase):
@override_settings(PUSHOVER_API_TOKEN=None)
def test_it_requires_pushover_api_token(self) -> None:
self._setup_data("123|0")
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Pushover notifications are not enabled.")
@ -63,7 +68,7 @@ class NotifyPushoverTestCase(BaseTestCase):
self._setup_data("123|0|2", status="up")
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 1)
payload = mock_post.call_args.kwargs["data"]
@ -82,7 +87,7 @@ class NotifyPushoverTestCase(BaseTestCase):
obj.tokens = 0
obj.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Rate limit exceeded")
@ -91,7 +96,7 @@ class NotifyPushoverTestCase(BaseTestCase):
self._setup_data("123|2|0", status="up")
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 1)
self.assertEqual(mock_post.call_count, 2)
@ -115,7 +120,7 @@ class NotifyPushoverTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertIn("All the other checks are up.", payload["message"])
@ -131,7 +136,7 @@ class NotifyPushoverTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertIn("The following checks are also down", payload["message"])
@ -145,7 +150,7 @@ class NotifyPushoverTestCase(BaseTestCase):
Check.objects.create(project=self.project, status="down")
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertIn("(last ping: never)", payload["message"])
@ -162,7 +167,7 @@ class NotifyPushoverTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertNotIn("Foobar", payload["message"])
@ -175,7 +180,7 @@ class NotifyPushoverTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["title"], "🔴 Foo & Bar")
@ -184,7 +189,7 @@ class NotifyPushoverTestCase(BaseTestCase):
def test_it_handles_disabled_priority(self, mock_post: Mock) -> None:
self._setup_data("123|-3")
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 0)
mock_post.assert_not_called()
@ -192,7 +197,7 @@ class NotifyPushoverTestCase(BaseTestCase):
def test_it_handles_disabled_up_priority(self, mock_post: Mock) -> None:
self._setup_data("123|0|-3", status="up")
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 0)
mock_post.assert_not_called()
@ -201,7 +206,7 @@ class NotifyPushoverTestCase(BaseTestCase):
self._setup_data("123|0")
mock_post.return_value.status_code = 400
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 400")
@ -214,7 +219,7 @@ class NotifyPushoverTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b"""{"user": "invalid"}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 400 (invalid user)")
@ -227,7 +232,7 @@ class NotifyPushoverTestCase(BaseTestCase):
mock_post.return_value.status_code = 500
mock_post.return_value.content = b"""{"user": "invalid"}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 500")

25
hc/api/tests/test_notify_rocketchat.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.test import BaseTestCase
@ -31,11 +31,16 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
url = mock_post.call_args.args[1]
@ -47,7 +52,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
@override_settings(ROCKETCHAT_ENABLED=False)
def test_it_requires_rocketchat_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Rocket.Chat notifications are not enabled.")
@ -56,7 +61,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
def test_it_does_not_disable_channel_on_404(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 404
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertFalse(self.channel.disabled)
self.assertEqual(self.channel.last_error, "Received status code 404")
@ -68,7 +73,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
fields = {f["title"]: f["value"] for f in attachment["fields"]}
self.assertEqual(fields["Schedule"], "\u034f* \u034f* \u034f* \u034f* \u034f*")
@ -82,7 +87,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -98,7 +103,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.ping.body_raw = b"X"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -112,7 +117,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.ping.kind = "fail"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -127,7 +132,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -142,7 +147,7 @@ class NotifyRocketChatTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]

51
hc/api/tests/test_notify_signal.py
View file

@ -13,7 +13,7 @@ from django.core.mail import EmailMessage, EmailMultiAlternatives
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping, TokenBucket
from hc.api.models import Channel, Check, Flip, Notification, Ping, TokenBucket
from hc.test import BaseTestCase
# Address is either a string (the path to the unix socket)
@ -96,6 +96,11 @@ class NotifySignalTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
def get_html(self, email: EmailMessage) -> str:
assert isinstance(email, EmailMultiAlternatives)
html, _ = email.alternatives[0]
@ -106,7 +111,7 @@ class NotifySignalTestCase(BaseTestCase):
def test_it_works(self, socket: Mock) -> None:
socketobj = setup_mock(socket, {})
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(socketobj.address, "/tmp/socket")
n = Notification.objects.get()
@ -136,7 +141,7 @@ class NotifySignalTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(socketobj.address, "/tmp/socket")
n = Notification.objects.get()
@ -153,7 +158,7 @@ class NotifySignalTestCase(BaseTestCase):
self.check.kind = "cron"
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert socketobj.req
params = socketobj.req["params"]
@ -171,7 +176,7 @@ class NotifySignalTestCase(BaseTestCase):
self.check.tags = "foo a&b"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(socketobj.address, "/tmp/socket")
n = Notification.objects.get()
@ -188,7 +193,7 @@ class NotifySignalTestCase(BaseTestCase):
def test_it_handles_host_port(self, socket: Mock) -> None:
socketobj = setup_mock(socket, {})
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(socketobj.address, ("example.org", 1234))
n = Notification.objects.get()
@ -200,7 +205,7 @@ class NotifySignalTestCase(BaseTestCase):
self.channel.value = json.dumps(payload)
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
# This channel should not notify on "down" events:
self.assertEqual(Notification.objects.count(), 0)
@ -209,7 +214,7 @@ class NotifySignalTestCase(BaseTestCase):
@patch("hc.api.transports.socket.socket")
def test_it_requires_signal_cli_socket(self, socket: Mock) -> None:
with override_settings(SIGNAL_CLI_SOCKET=None):
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Signal notifications are not enabled")
@ -222,7 +227,7 @@ class NotifySignalTestCase(BaseTestCase):
self.check.name = "Foo & Bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert socketobj.req
self.assertIn("Foo & Bar", socketobj.req["params"]["message"])
@ -235,7 +240,7 @@ class NotifySignalTestCase(BaseTestCase):
obj.tokens = 0
obj.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Rate limit exceeded")
socket.assert_not_called()
@ -250,7 +255,7 @@ class NotifySignalTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert socketobj.req
message = socketobj.req["params"]["message"]
@ -266,7 +271,7 @@ class NotifySignalTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert socketobj.req
message = socketobj.req["params"]["message"]
@ -280,7 +285,7 @@ class NotifySignalTestCase(BaseTestCase):
Check.objects.create(project=self.project, status="down")
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert socketobj.req
message = socketobj.req["params"]["message"]
@ -297,7 +302,7 @@ class NotifySignalTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert socketobj.req
message = socketobj.req["params"]["message"]
@ -308,7 +313,7 @@ class NotifySignalTestCase(BaseTestCase):
@patch("hc.api.transports.socket.socket")
def test_it_handles_unexpected_payload(self, socket: Mock, logger: Mock) -> None:
setup_mock(socket, "surprise")
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "signal-cli call failed (unexpected response)")
@ -335,7 +340,7 @@ class NotifySignalTestCase(BaseTestCase):
}
setup_mock(socket, msg)
self.channel.notify(self.check)
self.channel.notify(self.flip)
# It should disable the channel, so we don't attempt deliveries to
# this recipient in the future
@ -368,7 +373,7 @@ class NotifySignalTestCase(BaseTestCase):
}
setup_mock(socket, msg)
self.channel.notify(self.check)
self.channel.notify(self.flip)
# It should disable the channel, so we don't attempt deliveries to
# this recipient in the future
@ -382,7 +387,7 @@ class NotifySignalTestCase(BaseTestCase):
def test_it_handles_error_code(self, socket: Mock, logger: Mock) -> None:
setup_mock(socket, {"error": {"code": 123}})
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "signal-cli call failed (123)")
@ -394,7 +399,7 @@ class NotifySignalTestCase(BaseTestCase):
setup_mock(socket, {}, side_effect=OSError("oops"))
logging.disable(logging.CRITICAL)
self.channel.notify(self.check)
self.channel.notify(self.flip)
logging.disable(logging.NOTSET)
n = Notification.objects.get()
@ -407,7 +412,7 @@ class NotifySignalTestCase(BaseTestCase):
# The socket reader should skip over it.
socketobj.outbox += b'{"id": "surprise"}\n'
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -421,7 +426,7 @@ class NotifySignalTestCase(BaseTestCase):
# Add a message with no id in the outbox. The socket reader should skip over it.
socketobj.outbox += b"{}\n"
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "")
@ -454,7 +459,7 @@ class NotifySignalTestCase(BaseTestCase):
self.check.name = "Foo & Co"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "CAPTCHA proof required")
@ -490,7 +495,7 @@ class NotifySignalTestCase(BaseTestCase):
}
setup_mock(socket, msg)
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "signal-cli call failed (-32602)")

43
hc/api/tests/test_notify_slack.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.lib.curl import CurlError
from hc.test import BaseTestCase
@ -35,13 +35,18 @@ class NotifySlackTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = status
@override_settings(SITE_ROOT="http://testserver", SITE_LOGO_URL=None)
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
self._setup_data("https://example.org")
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
url = mock_post.call_args.args[1]
@ -64,7 +69,7 @@ class NotifySlackTestCase(BaseTestCase):
self._setup_data(v)
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
url = mock_post.call_args.args[1]
@ -75,7 +80,7 @@ class NotifySlackTestCase(BaseTestCase):
self._setup_data("123")
mock_post.return_value.status_code = 500
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 500")
@ -88,7 +93,7 @@ class NotifySlackTestCase(BaseTestCase):
def test_it_handles_timeout(self, mock_post: Mock) -> None:
self._setup_data("123")
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Timed out")
@ -103,7 +108,7 @@ class NotifySlackTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
attachment = payload["attachments"][0]
fields = {f["title"]: f["value"] for f in attachment["fields"]}
@ -118,14 +123,14 @@ class NotifySlackTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 1)
mock_post.assert_called_once()
@override_settings(SLACK_ENABLED=False)
def test_it_requires_slack_enabled(self) -> None:
self._setup_data("123")
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Slack notifications are not enabled.")
@ -135,7 +140,7 @@ class NotifySlackTestCase(BaseTestCase):
self._setup_data("123")
mock_post.return_value.status_code = 404
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 404")
@ -146,7 +151,7 @@ class NotifySlackTestCase(BaseTestCase):
self._setup_data("123")
mock_post.return_value.status_code = 404
self.channel.notify(self.check)
self.channel.notify(self.flip)
# Make sure the HTTP request was made only once (no retries):
self.assertEqual(mock_post.call_count, 1)
self.channel.refresh_from_db()
@ -161,7 +166,7 @@ class NotifySlackTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b"invalid_token"
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertTrue(self.channel.disabled)
@ -178,7 +183,7 @@ class NotifySlackTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b"surprise"
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertTrue(debug.called)
@override_settings(SITE_ROOT="http://testserver")
@ -190,7 +195,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.kind = "fail"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -207,7 +212,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
fields = {f["title"]: f["value"] for f in attachment["fields"]}
self.assertEqual(fields["Last Ping"], "Exit status 123, an hour ago")
@ -221,7 +226,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.kind = "log"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
fields = {f["title"]: f["value"] for f in attachment["fields"]}
@ -237,7 +242,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -253,7 +258,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -269,7 +274,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World" * 1000
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]
@ -285,7 +290,7 @@ class NotifySlackTestCase(BaseTestCase):
self.ping.body_raw = b"Hello ``` World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
attachment = mock_post.call_args.kwargs["json"]["attachments"][0]

31
hc/api/tests/test_notify_sms.py
View file

@ -10,7 +10,7 @@ from django.core import mail
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -31,12 +31,17 @@ class NotifySmsTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@override_settings(TWILIO_FROM="+000", TWILIO_MESSAGING_SERVICE_SID=None)
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["To"], "+1234567890")
@ -55,7 +60,7 @@ class NotifySmsTestCase(BaseTestCase):
@override_settings(TWILIO_ACCOUNT=None)
def test_it_requires_twilio_configuration(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "SMS notifications are not enabled")
@ -66,7 +71,7 @@ class NotifySmsTestCase(BaseTestCase):
self.check.last_ping = now() - td(hours=2)
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["MessagingServiceSid"], "dummy-sid")
@ -79,7 +84,7 @@ class NotifySmsTestCase(BaseTestCase):
self.profile.sms_sent = 50
self.profile.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_not_called()
n = Notification.objects.get()
@ -102,7 +107,7 @@ class NotifySmsTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_called_once()
@override_settings(TWILIO_FROM="+000")
@ -113,7 +118,7 @@ class NotifySmsTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertIn("Foo > Bar & Co", payload["Body"])
@ -123,7 +128,7 @@ class NotifySmsTestCase(BaseTestCase):
payload = {"value": "+123123123", "up": True, "down": False}
self.channel.value = json.dumps(payload)
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_not_called()
@override_settings(TWILIO_FROM="+000")
@ -132,11 +137,11 @@ class NotifySmsTestCase(BaseTestCase):
payload = {"value": "+123123123", "up": True, "down": False}
self.channel.value = json.dumps(payload)
self.check.last_ping = now()
self.check.status = "up"
self.flip.old_status = "down"
self.flip.new_status = "up"
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
assert isinstance(payload["Body"], str)
@ -150,7 +155,7 @@ class NotifySmsTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b"""{"code": 21211}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
# Make sure the HTTP request was made only once (no retries):
self.channel.refresh_from_db()
@ -172,7 +177,7 @@ class NotifySmsTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertNotIn("""Last ping was""", payload["Body"])

15
hc/api/tests/test_notify_spike.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -28,11 +28,16 @@ class NotifySpikeTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -43,7 +48,7 @@ class NotifySpikeTestCase(BaseTestCase):
@override_settings(SPIKE_ENABLED=False)
def test_it_requires_spike_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Spike notifications are not enabled.")
@ -55,7 +60,7 @@ class NotifySpikeTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(payload["title"], "Foo & Bar is UP")
@ -67,7 +72,7 @@ class NotifySpikeTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertNotIn("Last ping was", payload["message"])

41
hc/api/tests/test_notify_telegram.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping, TokenBucket
from hc.api.models import Channel, Check, Flip, Notification, Ping, TokenBucket
from hc.test import BaseTestCase
@ -34,11 +34,16 @@ class NotifyTelegramTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -66,7 +71,7 @@ class NotifyTelegramTestCase(BaseTestCase):
self.ping.exitstatus = 123
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("<b>Last Ping:</b> Exit status 123, an hour ago", payload["text"])
@ -77,7 +82,7 @@ class NotifyTelegramTestCase(BaseTestCase):
self.channel.value = json.dumps({"id": 123, "thread_id": 456})
self.channel.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -93,7 +98,7 @@ class NotifyTelegramTestCase(BaseTestCase):
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn(
@ -106,7 +111,7 @@ class NotifyTelegramTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b'{"description": "Hi"}'
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, 'Received status code 400 with a message: "Hi"')
@ -115,7 +120,7 @@ class NotifyTelegramTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.json = Mock(side_effect=ValueError)
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 400")
@ -129,7 +134,7 @@ class NotifyTelegramTestCase(BaseTestCase):
mock_post.side_effect = [error_response, Mock(status_code=200)]
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(mock_post.call_count, 2)
# The chat id should have been updated
@ -143,7 +148,7 @@ class NotifyTelegramTestCase(BaseTestCase):
def test_it_obeys_rate_limit(self) -> None:
TokenBucket.objects.create(value="tg-123", tokens=0)
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Rate limit exceeded")
@ -157,7 +162,7 @@ class NotifyTelegramTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("All the other checks are up.", payload["text"])
@ -172,7 +177,7 @@ class NotifyTelegramTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("The following checks are also down", payload["text"])
@ -186,7 +191,7 @@ class NotifyTelegramTestCase(BaseTestCase):
Check.objects.create(project=self.project, status="down")
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("(last ping: never)", payload["text"])
@ -202,7 +207,7 @@ class NotifyTelegramTestCase(BaseTestCase):
other.last_ping = now() - td(minutes=61)
other.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertNotIn("Foobar", payload["text"])
@ -215,7 +220,7 @@ class NotifyTelegramTestCase(BaseTestCase):
"description": "Forbidden: the group chat was deleted"
}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertTrue(self.channel.disabled)
@ -226,7 +231,7 @@ class NotifyTelegramTestCase(BaseTestCase):
"description": "Forbidden: bot was blocked by the user"
}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertTrue(self.channel.disabled)
@ -237,7 +242,7 @@ class NotifyTelegramTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("<b>Last Ping Body:</b>\n", payload["text"])
@ -250,7 +255,7 @@ class NotifyTelegramTestCase(BaseTestCase):
self.ping.body_raw = b"Hello World" * 100
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("[truncated]", payload["text"])
@ -262,7 +267,7 @@ class NotifyTelegramTestCase(BaseTestCase):
self.ping.body_raw = b"<b>bold</b>\nfoo & bar"
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertIn("&lt;b&gt;bold&lt;/b&gt;\n", payload["text"])

17
hc/api/tests/test_notify_trello.py
View file

@ -7,7 +7,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -35,11 +35,16 @@ class NotifyTrelloTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
params = mock_post.call_args.kwargs["params"]
@ -52,7 +57,7 @@ class NotifyTrelloTestCase(BaseTestCase):
@override_settings(TRELLO_APP_KEY=None)
def test_it_requires_trello_app_key(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Trello notifications are not enabled.")
@ -64,7 +69,7 @@ class NotifyTrelloTestCase(BaseTestCase):
self.check.tz = "Europe/Riga"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
params = mock_post.call_args.kwargs["params"]
a = "\u034f*"
@ -78,7 +83,7 @@ class NotifyTrelloTestCase(BaseTestCase):
self.check.name = "Foo & Bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
params = mock_post.call_args.kwargs["params"]
self.assertEqual(params["name"], "Down: Foo & Bar")
@ -89,7 +94,7 @@ class NotifyTrelloTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
params = mock_post.call_args.kwargs["params"]
self.assertIn("**Last Ping:** never", params["desc"])

19
hc/api/tests/test_notify_victorops.py
View file

@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -28,11 +28,16 @@ class NotifyVictorOpsTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]
@ -42,7 +47,7 @@ class NotifyVictorOpsTestCase(BaseTestCase):
@override_settings(VICTOROPS_ENABLED=False)
def test_it_requires_victorops_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Splunk On-Call notifications are not enabled.")
@ -55,7 +60,7 @@ class NotifyVictorOpsTestCase(BaseTestCase):
self.check.status = "up"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(
@ -66,7 +71,7 @@ class NotifyVictorOpsTestCase(BaseTestCase):
def test_it_does_not_retry_404(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 404
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 404")
@ -76,7 +81,7 @@ class NotifyVictorOpsTestCase(BaseTestCase):
def test_it_disables_channel_on_404(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 404
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.channel.refresh_from_db()
self.assertTrue(self.channel.disabled)
@ -86,7 +91,7 @@ class NotifyVictorOpsTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
payload = mock_post.call_args.kwargs["json"]

63
hc/api/tests/test_notify_webhook.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
from hc.lib.curl import CurlError
from hc.test import BaseTestCase
@ -30,6 +30,11 @@ class NotifyWebhookTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = status
@patch("hc.api.transports.curl.request", autospec=True)
def test_webhook(self, mock_get: Mock) -> None:
definition = {
@ -42,7 +47,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
mock_get.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_get.call_args
self.assertEqual(args, ("get", "http://example"))
@ -60,7 +65,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition))
self.channel.notify(self.check)
self.channel.notify(self.flip)
# The transport should have retried 3 times
self.assertEqual(mock_get.call_count, 3)
@ -83,7 +88,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
mock_get.return_value.status_code = 500
self.channel.notify(self.check)
self.channel.notify(self.flip)
# The transport should have retried 3 times
self.assertEqual(mock_get.call_count, 3)
@ -107,7 +112,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition))
self.channel.notify(self.check, is_test=True)
self.channel.notify(self.flip, is_test=True)
# is_test flag is set, the transport should not retry:
self.assertEqual(mock_get.call_count, 1)
@ -129,7 +134,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.check.tags = "foo bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
url = "http://host/%s/down/foo/bar/?name=Hello%%20World" % self.check.code
@ -152,7 +157,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.check.tags = "foo bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
# $$NAMETAG1 should *not* get transformed to "foo"
url = mock_get.call_args.args[1]
@ -170,7 +175,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
method, url = mock_request.call_args.args
self.assertEqual(method, "post")
self.assertEqual(url, "http://example.com")
@ -195,7 +200,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.check.tags = "foo"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_get.call_args
self.assertEqual(args, ("get", "http://host/%24TAG1"))
@ -210,7 +215,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition), status="up")
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_get.call_args
self.assertEqual(args, ("get", "http://bar"))
@ -225,7 +230,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition), status="up")
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_get.assert_not_called()
self.assertEqual(Notification.objects.count(), 0)
@ -242,7 +247,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
# unicode should be encoded into utf-8
payload = mock_request.call_args.kwargs["data"]
@ -258,7 +263,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition))
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_request.call_args
self.assertEqual(args, ("post", "http://foo.com"))
@ -275,7 +280,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition))
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_request.call_args
self.assertEqual(args, ("get", "http://foo.com"))
@ -291,7 +296,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition))
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_request.call_args
self.assertEqual(args, ("get", "http://foo.com"))
@ -310,7 +315,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.check.name = "Foo"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
args, kwargs = mock_request.call_args
self.assertEqual(args, ("get", "http://foo.com"))
@ -326,7 +331,7 @@ class NotifyWebhookTestCase(BaseTestCase):
}
self._setup_data(json.dumps(definition))
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Webhook notifications are not enabled.")
@ -343,7 +348,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
headers = mock_request.call_args.kwargs["headers"]
self.assertEqual(headers["X-Foo"], "b&#257;r")
@ -360,7 +365,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
headers = mock_request.call_args.kwargs["headers"]
self.assertEqual(headers["X-Foo"], "½")
@ -379,7 +384,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.check.tags = "foo bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
body = json.loads(payload)
@ -401,7 +406,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.ping.body_raw = ping_body
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, ping_body)
@ -424,7 +429,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.ping.body_raw = ping_body
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
url = mock_post.call_args.args[1]
self.assertTrue(url.endswith("$BODY"))
@ -442,7 +447,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
Ping.objects.create(owner=self.check, exitstatus=123)
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, b"Exit status 123")
@ -459,7 +464,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
# Note - does not create Ping object
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, b"Exit status -1")
@ -474,7 +479,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
Ping.objects.create(owner=self.check) # does not set exitstatus
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, b"Exit status -1")
@ -489,7 +494,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
self.check.name = 'Project "Foo"'
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, self.check.name.encode())
@ -504,7 +509,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self._setup_data(json.dumps(definition))
self.check.name = 'Project "Foo"'
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, json.dumps(self.check.name).encode())
@ -523,7 +528,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.ping.body_raw = b'Project "Foo"'
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, b'Project "Foo"')
@ -544,7 +549,7 @@ class NotifyWebhookTestCase(BaseTestCase):
self.ping.body_raw = ping_body.encode()
self.ping.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload, json.dumps(ping_body).encode())

23
hc/api/tests/test_notify_whatsapp.py
View file

@ -10,7 +10,7 @@ from django.core import mail
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -40,11 +40,16 @@ class NotifyWhatsAppTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
@patch("hc.api.transports.curl.request", autospec=True)
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["From"], "whatsapp:+000")
@ -68,7 +73,7 @@ class NotifyWhatsAppTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.check.last_ping = now()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
variables = json.loads(payload["ContentVariables"])
@ -76,14 +81,14 @@ class NotifyWhatsAppTestCase(BaseTestCase):
@override_settings(TWILIO_ACCOUNT=None)
def test_it_requires_twilio_account(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "WhatsApp notifications are not enabled")
@override_settings(WHATSAPP_UP_CONTENT_SID=None)
def test_it_requires_content_template_sids(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "WhatsApp notifications are not enabled")
@ -96,7 +101,7 @@ class NotifyWhatsAppTestCase(BaseTestCase):
self.check.last_ping = now() - td(hours=2)
self.channel.notify(self.check)
self.channel.notify(self.flip)
self.assertEqual(Notification.objects.count(), 0)
mock_post.assert_not_called()
@ -107,7 +112,7 @@ class NotifyWhatsAppTestCase(BaseTestCase):
self.profile.sms_sent = 50
self.profile.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
mock_post.assert_not_called()
n = Notification.objects.get()
@ -126,7 +131,7 @@ class NotifyWhatsAppTestCase(BaseTestCase):
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertIn("Foo > Bar & Co", payload["ContentVariables"])
@ -138,7 +143,7 @@ class NotifyWhatsAppTestCase(BaseTestCase):
mock_post.return_value.status_code = 400
mock_post.return_value.content = b"""{"code": 21211}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
# Make sure the HTTP request was made only once (no retries):
self.channel.refresh_from_db()

23
hc/api/tests/test_notify_zulip.py
View file

@ -9,7 +9,7 @@ from unittest.mock import Mock, patch
from django.test.utils import override_settings
from django.utils.timezone import now
from hc.api.models import Channel, Check, Notification
from hc.api.models import Channel, Check, Flip, Notification
from hc.test import BaseTestCase
@ -29,6 +29,11 @@ class NotifyZulipTestCase(BaseTestCase):
self.channel.save()
self.channel.checks.add(self.check)
self.flip = Flip(owner=self.check)
self.flip.created = now()
self.flip.old_status = "new"
self.flip.new_status = "down"
def definition(self, **kwargs: str) -> dict[str, str]:
d = {
"bot_email": "bot@example.org",
@ -43,7 +48,7 @@ class NotifyZulipTestCase(BaseTestCase):
def test_it_works(self, mock_post: Mock) -> None:
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
method, url = mock_post.call_args.args
@ -64,7 +69,7 @@ class NotifyZulipTestCase(BaseTestCase):
self.channel.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["topic"], "foo")
@ -74,7 +79,7 @@ class NotifyZulipTestCase(BaseTestCase):
mock_post.return_value.status_code = 403
mock_post.return_value.content = b"""{"msg": "Nice try"}"""
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, 'Received status code 403 with a message: "Nice try"')
@ -84,7 +89,7 @@ class NotifyZulipTestCase(BaseTestCase):
mock_post.return_value.status_code = 403
mock_post.return_value.json = Mock(side_effect=ValueError)
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Received status code 403")
@ -100,7 +105,7 @@ class NotifyZulipTestCase(BaseTestCase):
}
self.channel.value = json.dumps(definition)
self.channel.notify(self.check)
self.channel.notify(self.flip)
assert Notification.objects.count() == 1
method, url = mock_post.call_args.args
@ -111,7 +116,7 @@ class NotifyZulipTestCase(BaseTestCase):
@override_settings(ZULIP_ENABLED=False)
def test_it_requires_zulip_enabled(self) -> None:
self.channel.notify(self.check)
self.channel.notify(self.flip)
n = Notification.objects.get()
self.assertEqual(n.error, "Zulip notifications are not enabled.")
@ -123,7 +128,7 @@ class NotifyZulipTestCase(BaseTestCase):
self.check.name = "Foo & Bar"
self.check.save()
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertEqual(payload["topic"], "Foo & Bar is DOWN")
@ -134,7 +139,7 @@ class NotifyZulipTestCase(BaseTestCase):
self.check.save()
mock_post.return_value.status_code = 200
self.channel.notify(self.check)
self.channel.notify(self.flip)
payload = mock_post.call_args.kwargs["data"]
self.assertNotIn("Last ping was", payload["content"])

2
hc/api/tests/test_sendalerts.py
View file

@ -1,10 +1,8 @@
from __future__ import annotations
from datetime import timedelta as td
from io import StringIO
from unittest.mock import Mock, patch
from django.core.management import call_command
from django.utils.timezone import now
from hc.api.management.commands.sendalerts import Command, notify

38
hc/api/transports.py
View file

@ -33,7 +33,7 @@ from hc.lib.string import replace
from hc.lib.typealias import JSONDict, JSONList, JSONValue
if TYPE_CHECKING:
from hc.api.models import Channel, Check, Notification, Ping
from hc.api.models import Channel, Check, Flip, Notification, Ping
try:
import apprise
@ -94,7 +94,11 @@ class Transport(object):
raise NotImplementedError()
def is_noop(self, status: str) -> bool:
def notify_flip(self, flip: Flip, notification: Notification) -> None:
# The default implementation calls self.notify
self.notify(flip.owner, notification)
def is_noop(self, check: Check) -> bool:
"""Return True if transport will ignore check's current status.
This method is overridden in Webhook subclass where the user can
@ -328,7 +332,7 @@ class Webhook(HttpTransport):
def prepare(
self,
template: str,
check: Check,
flip: Flip,
urlencode: bool = False,
latin1: bool = False,
allow_ping_body: bool = False,
@ -338,10 +342,11 @@ class Webhook(HttpTransport):
def safe(s: str) -> str:
return quote(s) if urlencode else s
check = flip.owner
ctx = {
"$CODE": str(check.code),
"$STATUS": check.status,
"$NOW": safe(now().replace(microsecond=0).isoformat()),
"$STATUS": flip.new_status,
"$NOW": safe(flip.created.replace(microsecond=0).isoformat()),
"$NAME_JSON": safe(json.dumps(check.name)),
"$NAME": safe(check.name),
"$TAGS": safe(check.tags),
@ -377,23 +382,23 @@ class Webhook(HttpTransport):
return False
def notify(self, check: Check, notification: Notification) -> None:
def notify_flip(self, flip: Flip, notification: Notification) -> None:
if not settings.WEBHOOKS_ENABLED:
raise TransportError("Webhook notifications are not enabled.")
spec = self.channel.webhook_spec(check.status)
spec = self.channel.webhook_spec(flip.new_status)
if not spec.url:
raise TransportError("Empty webhook URL")
url = self.prepare(spec.url, check, urlencode=True)
url = self.prepare(spec.url, flip, urlencode=True)
headers = {}
for key, value in spec.headers.items():
# Header values should contain ASCII and latin-1 only
headers[key] = self.prepare(value, check, latin1=True)
headers[key] = self.prepare(value, flip, latin1=True)
body, body_bytes = spec.body, None
if body and spec.method in ("POST", "PUT"):
body = self.prepare(body, check, allow_ping_body=True)
body = self.prepare(body, flip, allow_ping_body=True)
body_bytes = body.encode()
retry = True
@ -911,7 +916,7 @@ class Sms(HttpTransport):
else:
return not self.channel.phone.notify_up
def notify(self, check: Check, notification: Notification) -> None:
def notify_flip(self, flip: Flip, notification: Notification) -> None:
if not settings.TWILIO_ACCOUNT or not settings.TWILIO_AUTH:
raise TransportError("SMS notifications are not enabled")
@ -922,7 +927,12 @@ class Sms(HttpTransport):
url = self.URL % settings.TWILIO_ACCOUNT
auth = (settings.TWILIO_ACCOUNT, settings.TWILIO_AUTH)
text = tmpl("sms_message.html", check=check, site_name=settings.SITE_NAME)
text = tmpl(
"sms_message.html",
check=flip.owner,
status=flip.new_status,
site_name=settings.SITE_NAME,
)
data = {
"To": self.channel.phone.value,
@ -1392,14 +1402,14 @@ class Gotify(HttpTransport):
class Group(Transport):
def notify(self, check: Check, notification: Notification) -> None:
def notify_flip(self, flip: Flip, notification: Notification) -> None:
channels = self.channel.group_channels
# If notification's owner field is None then this is a test notification,
# and we should pass is_test=True to channel.notify() calls
is_test = notification.owner is None
error_count = 0
for channel in channels:
error = channel.notify(check, is_test=is_test)
error = channel.notify(flip, is_test=is_test)
if error and error != "no-op":
error_count += 1
if error_count:

12
hc/front/views.py
View file

@ -1293,16 +1293,22 @@ def send_test_notification(
dummy.last_ping = now() - td(days=1)
dummy.n_pings = 42
dummy_flip = Flip(owner=dummy)
dummy_flip.created = now()
dummy_flip.old_status = "up"
dummy_flip.new_status = "down"
# Delete all older test notifications for this channel
Notification.objects.filter(channel=channel, owner=None).delete()
# Send the test notification
error = channel.notify(dummy, is_test=True)
error = channel.notify(dummy_flip, is_test=True)
if error == "no-op":
# This channel may be configured to send "up" notifications only.
dummy.status = "up"
error = channel.notify(dummy, is_test=True)
dummy_flip.old_status = "down"
dummy_flip.new_status = "up"
error = channel.notify(dummy_flip, is_test=True)
if error:
messages.warning(request, "Could not send a test notification. %s." % error)

4
templates/integrations/sms_message.html
View file

@ -1,6 +1,6 @@
{% load humanize %}
{% if check.status == "down" %}
{% if status == "down" %}
{{ site_name }}: The check "{{ check.name_then_code|safe }}" is DOWN.{% if check.last_ping %} Last ping was {{ check.last_ping|naturaltime }}.{% endif %}
{% else %}
{{ site_name }}: The check "{{ check.name_then_code|safe }}" is UP.
{% endif %}
{% endif %}