291 lines
10 KiB
Python
291 lines
10 KiB
Python
"""Tests for the ntfy.sh push notification sender."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import httpx
|
|
|
|
from cheddahbot.ntfy import NtfyChannel, NtfyNotifier
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NtfyChannel routing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNtfyChannel:
|
|
def test_accepts_matching_category_and_pattern(self):
|
|
ch = NtfyChannel(
|
|
name="human_action",
|
|
server="https://ntfy.sh",
|
|
topic="test-topic",
|
|
categories=["clickup", "autocora"],
|
|
include_patterns=["completed", "SUCCESS"],
|
|
)
|
|
assert ch.accepts("ClickUp task completed: **Acme PR**", "clickup") is True
|
|
assert ch.accepts("AutoCora SUCCESS: **keyword**", "autocora") is True
|
|
|
|
def test_rejects_wrong_category(self):
|
|
ch = NtfyChannel(
|
|
name="human_action",
|
|
server="https://ntfy.sh",
|
|
topic="test-topic",
|
|
categories=["clickup"],
|
|
include_patterns=["completed"],
|
|
)
|
|
assert ch.accepts("Some autocora message completed", "autocora") is False
|
|
|
|
def test_rejects_non_matching_pattern(self):
|
|
ch = NtfyChannel(
|
|
name="human_action",
|
|
server="https://ntfy.sh",
|
|
topic="test-topic",
|
|
categories=["clickup"],
|
|
include_patterns=["completed"],
|
|
)
|
|
assert ch.accepts("Executing ClickUp task: **Acme PR**", "clickup") is False
|
|
|
|
def test_no_include_patterns_accepts_all_in_category(self):
|
|
ch = NtfyChannel(
|
|
name="all_clickup",
|
|
server="https://ntfy.sh",
|
|
topic="test-topic",
|
|
categories=["clickup"],
|
|
)
|
|
assert ch.accepts("Any message at all", "clickup") is True
|
|
|
|
def test_exclude_patterns_take_priority(self):
|
|
ch = NtfyChannel(
|
|
name="test",
|
|
server="https://ntfy.sh",
|
|
topic="test-topic",
|
|
categories=["clickup"],
|
|
include_patterns=["task"],
|
|
exclude_patterns=["Executing"],
|
|
)
|
|
assert ch.accepts("Executing ClickUp task", "clickup") is False
|
|
assert ch.accepts("ClickUp task completed", "clickup") is True
|
|
|
|
def test_case_insensitive_patterns(self):
|
|
ch = NtfyChannel(
|
|
name="test",
|
|
server="https://ntfy.sh",
|
|
topic="test-topic",
|
|
categories=["autocora"],
|
|
include_patterns=["success"],
|
|
)
|
|
assert ch.accepts("AutoCora SUCCESS: **kw**", "autocora") is True
|
|
|
|
def test_empty_topic_filtered_by_notifier(self):
|
|
ch = NtfyChannel(
|
|
name="empty", server="https://ntfy.sh", topic="",
|
|
categories=["clickup"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
assert notifier.enabled is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NtfyNotifier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNtfyNotifier:
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_notify_posts_to_matching_channel(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch = NtfyChannel(
|
|
name="human_action",
|
|
server="https://ntfy.sh",
|
|
topic="my-topic",
|
|
categories=["clickup"],
|
|
include_patterns=["completed"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("ClickUp task completed: **Acme PR**", "clickup")
|
|
|
|
# Wait for daemon thread
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
mock_post.assert_called_once()
|
|
call_args = mock_post.call_args
|
|
assert call_args[0][0] == "https://ntfy.sh/my-topic"
|
|
assert call_args[1]["headers"]["Title"] == "CheddahBot [clickup]"
|
|
assert call_args[1]["headers"]["Priority"] == "high"
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_notify_skips_non_matching_channel(self, mock_post):
|
|
ch = NtfyChannel(
|
|
name="errors",
|
|
server="https://ntfy.sh",
|
|
topic="err-topic",
|
|
categories=["clickup"],
|
|
include_patterns=["failed"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("ClickUp task completed: **Acme PR**", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
mock_post.assert_not_called()
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_notify_routes_to_multiple_channels(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch1 = NtfyChannel(
|
|
name="all", server="https://ntfy.sh", topic="all-topic",
|
|
categories=["clickup"],
|
|
)
|
|
ch2 = NtfyChannel(
|
|
name="errors", server="https://ntfy.sh", topic="err-topic",
|
|
categories=["clickup"], include_patterns=["failed"],
|
|
)
|
|
notifier = NtfyNotifier([ch1, ch2])
|
|
notifier.notify("ClickUp task failed: **Acme**", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
assert mock_post.call_count == 2
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_webhook_error_is_swallowed(self, mock_post):
|
|
mock_post.side_effect = httpx.ConnectError("connection refused")
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
# Should not raise
|
|
notifier.notify("ClickUp task completed: **test**", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_4xx_is_logged_not_raised(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=400, text="Bad Request")
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("ClickUp task completed: **test**", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
def test_enabled_property(self):
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"],
|
|
)
|
|
assert NtfyNotifier([ch]).enabled is True
|
|
assert NtfyNotifier([]).enabled is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Post format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPostFormat:
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_includes_tags_header(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"], tags="white_check_mark",
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("task completed", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
headers = mock_post.call_args[1]["headers"]
|
|
assert headers["Tags"] == "white_check_mark"
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_omits_tags_header_when_empty(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"], tags="",
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("task completed", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
headers = mock_post.call_args[1]["headers"]
|
|
assert "Tags" not in headers
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_custom_server_url(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch = NtfyChannel(
|
|
name="test", server="https://my-ntfy.example.com",
|
|
topic="topic", categories=["clickup"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("task completed", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
assert mock_post.call_args[0][0] == "https://my-ntfy.example.com/topic"
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_message_sent_as_body(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"],
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("Hello **world**", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
assert mock_post.call_args[1]["content"] == b"Hello **world**"
|
|
|
|
@patch("cheddahbot.ntfy.httpx.post")
|
|
def test_priority_header(self, mock_post):
|
|
mock_post.return_value = MagicMock(status_code=200)
|
|
ch = NtfyChannel(
|
|
name="test", server="https://ntfy.sh", topic="topic",
|
|
categories=["clickup"], priority="urgent",
|
|
)
|
|
notifier = NtfyNotifier([ch])
|
|
notifier.notify("task completed", "clickup")
|
|
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.daemon and t.is_alive():
|
|
t.join(timeout=2)
|
|
|
|
assert mock_post.call_args[1]["headers"]["Priority"] == "urgent"
|