Files
shelfmark/tests/core/test_notifications.py
Alex 3a3a3ce449 Add new python tooling + apply ruff linter cleanup (#845)
- Adds `uv`, `ruff`, `pyright`, `vulture` and `pytest-xdist`
- Move project, lockfile, docker build etc to uv
- Align python tooling on 3.14
- Huge bulk of ruff linter fixes applied. Still in progress but all the
core types are now enforced
- Update CI and test helpers
2026-04-10 13:03:25 +01:00

600 lines
20 KiB
Python

"""Tests for core notification rendering and dispatch helpers."""
import logging
from shelfmark.core import notifications as notifications_module
class _FakeExecutor:
def __init__(self):
self.calls = []
def submit(self, fn, *args, **kwargs):
self.calls.append((fn, args, kwargs))
return object()
class _FakeNotifyType:
INFO = "INFO"
SUCCESS = "SUCCESS"
WARNING = "WARNING"
FAILURE = "FAILURE"
class _FakePlugin:
"""Fake Apprise plugin returned by instantiate()."""
def __init__(self, raw_url: str):
self._url = raw_url
self.app_id = "FakePlugin"
def url(self, privacy=False):
return self._url
class _FakeAppriseClient:
def __init__(self):
self.add_calls = []
self.instantiate_calls: list[dict[str, object | None]] = []
self.notify_calls = []
self.notify_result = True
self.notify_results_by_url: dict[str, bool] = {}
self.reject_urls: set[str] = set()
self.instantiate_exceptions_by_url: dict[str, Exception] = {}
self.notify_exceptions_by_url: dict[str, Exception] = {}
self.notify_warning_messages: list[str] = []
self.notify_info_messages: list[str] = []
self._active_url: str | None = None
def add(self, plugin):
url = getattr(plugin, "_url", str(plugin))
self.add_calls.append(url)
self._active_url = url
return True
def notify(self, **kwargs):
self.notify_calls.append(kwargs)
for message in self.notify_info_messages:
logging.getLogger("apprise.plugins.pushover").info(message)
for message in self.notify_warning_messages:
logging.getLogger("apprise.plugins.pushover").warning(message)
if self._active_url and self._active_url in self.notify_exceptions_by_url:
raise self.notify_exceptions_by_url[self._active_url]
if self._active_url and self._active_url in self.notify_results_by_url:
return self.notify_results_by_url[self._active_url]
return self.notify_result
class _FakeAppriseClass:
"""Fake for apprise.Apprise that acts as both constructor and has instantiate()."""
def __init__(self, module):
self._module = module
def __call__(self, *args, **kwargs):
self._module.apprise_kwargs = kwargs
asset = kwargs.get("asset")
self._module.asset_kwargs = getattr(asset, "kwargs", None)
self._module.client.asset = asset
return self._module.client
def instantiate(self, url, asset=None, tag=None, suppress_exceptions=True):
_ = (tag, suppress_exceptions)
client = self._module.client
client.instantiate_calls.append(
{
"url": url,
"asset_kwargs": getattr(asset, "kwargs", None),
}
)
if url in client.instantiate_exceptions_by_url:
raise client.instantiate_exceptions_by_url[url]
if url in client.reject_urls:
return None
return _FakePlugin(url)
class _FakeAppriseModule:
NotifyType = _FakeNotifyType
asset_kwargs: dict[str, str] | None = None
def __init__(self):
self.client = _FakeAppriseClient()
self.apprise_kwargs = {}
self.Apprise = _FakeAppriseClass(self)
class AppriseAsset:
def __init__(self, **kwargs):
self.kwargs = kwargs
def test_render_message_includes_admin_note_for_rejection():
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.REQUEST_REJECTED,
title="Example Book",
author="Example Author",
admin_note="Missing metadata",
)
title, body = notifications_module._render_message(context)
assert title == "Request Rejected"
assert "Missing metadata" in body
def test_render_message_includes_error_line_for_download_failure():
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.DOWNLOAD_FAILED,
title="Example Book",
author="Example Author",
error_message="Connection timeout",
)
title, body = notifications_module._render_message(context)
assert title == "Download Failed"
assert "Connection timeout" in body
def test_render_message_uses_request_approved_copy():
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.REQUEST_FULFILLED,
title="Example Book",
author="Example Author",
)
title, body = notifications_module._render_message(context)
assert title == "Request Approved"
assert "was approved." in body
def test_notify_admin_submits_non_blocking_when_route_matches_event(monkeypatch):
fake_executor = _FakeExecutor()
monkeypatch.setattr(notifications_module, "_executor", fake_executor)
monkeypatch.setattr(
notifications_module,
"_resolve_admin_routes",
lambda: [{"event": "request_created", "url": "discord://Webhook/Token"}],
)
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.REQUEST_CREATED,
title="Example Book",
author="Example Author",
username="reader",
)
notifications_module.notify_admin(
notifications_module.NotificationEvent.REQUEST_CREATED,
context,
)
assert len(fake_executor.calls) == 1
def test_notify_admin_skips_when_no_route_matches_event(monkeypatch):
fake_executor = _FakeExecutor()
monkeypatch.setattr(notifications_module, "_executor", fake_executor)
monkeypatch.setattr(
notifications_module,
"_resolve_admin_routes",
lambda: [{"event": "download_failed", "url": "discord://Webhook/Token"}],
)
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.REQUEST_CREATED,
title="Example Book",
author="Example Author",
)
notifications_module.notify_admin(
notifications_module.NotificationEvent.REQUEST_CREATED,
context,
)
assert fake_executor.calls == []
def test_send_admin_event_passes_expected_title_body_and_notify_type(monkeypatch):
fake_apprise = _FakeAppriseModule()
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.REQUEST_REJECTED,
title="Example Book",
author="Example Author",
admin_note="Rule blocked this source",
)
result = notifications_module._send_admin_event(
notifications_module.NotificationEvent.REQUEST_REJECTED,
context,
["discord://Webhook/Token"],
)
assert result["success"] is True
assert fake_apprise.client.notify_calls
notify_kwargs = fake_apprise.client.notify_calls[0]
assert notify_kwargs["title"] == "Request Rejected"
assert "Rule blocked this source" in notify_kwargs["body"]
assert notify_kwargs["notify_type"] == _FakeNotifyType.WARNING
def test_dispatch_to_apprise_uses_shelfmark_asset_defaults(monkeypatch):
fake_apprise = _FakeAppriseModule()
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
result = notifications_module._dispatch_to_apprise(
["ntfys://ntfy.sh/shelfmark"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is True
assert fake_apprise.asset_kwargs is not None
assert fake_apprise.asset_kwargs["app_id"] == "Shelfmark"
assert "logo.png" in fake_apprise.asset_kwargs["image_url_logo"]
def test_dispatch_to_apprise_passes_shelfmark_asset_to_instantiate(monkeypatch):
fake_apprise = _FakeAppriseModule()
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
result = notifications_module._dispatch_to_apprise(
["ntfys://ntfy.sh/shelfmark"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is True
assert fake_apprise.client.instantiate_calls
instantiate_call = fake_apprise.client.instantiate_calls[0]
asset_kwargs = instantiate_call["asset_kwargs"]
assert isinstance(asset_kwargs, dict)
assert asset_kwargs["app_id"] == "Shelfmark"
assert "logo.png" in asset_kwargs["image_url_logo"]
def test_dispatch_to_apprise_logs_captured_apprise_info_messages(monkeypatch):
fake_apprise = _FakeAppriseModule()
fake_apprise.client.notify_info_messages = ["Sent Pushover notification to ALL_DEVICES."]
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
info_messages: list[str] = []
def _fake_info(message, *args, **kwargs):
_ = kwargs
info_messages.append(message % args if args else str(message))
monkeypatch.setattr(notifications_module.logger, "info", _fake_info)
result = notifications_module._dispatch_to_apprise(
["ntfys://ntfy.sh/shelfmark"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is True
assert any(
"Apprise source [apprise.plugins.pushover]: Sent Pushover notification to ALL_DEVICES."
in message
for message in info_messages
)
def test_dispatch_to_apprise_notify_false_returns_generic_failure_and_logs(monkeypatch):
fake_apprise = _FakeAppriseModule()
fake_apprise.client.notify_result = False
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
warning_messages: list[str] = []
def _fake_warning(message, *args, **kwargs):
_ = kwargs
warning_messages.append(message % args if args else str(message))
monkeypatch.setattr(notifications_module.logger, "warning", _fake_warning)
result = notifications_module._dispatch_to_apprise(
["pover://user_key@app_token"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is False
assert result["message"] == "Notification delivery failed"
assert result["details"] == ["pover: delivery failed"]
assert any("scheme(s): pover" in message for message in warning_messages)
def test_dispatch_to_apprise_partial_success_returns_success(monkeypatch):
fake_apprise = _FakeAppriseModule()
fake_apprise.client.notify_results_by_url = {
"gotifys://gotify.example/token": False,
"ntfys://ntfy.sh/shelfmark": True,
}
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
result = notifications_module._dispatch_to_apprise(
["gotifys://gotify.example/token", "ntfys://ntfy.sh/shelfmark"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is True
assert result["message"] == "Notification sent to 1 URL(s) (1 URL(s) failed)"
assert result["details"] == ["gotifys: delivery failed"]
def test_dispatch_to_apprise_logs_captured_apprise_warning_messages(monkeypatch):
fake_apprise = _FakeAppriseModule()
fake_apprise.client.notify_result = False
fake_apprise.client.notify_warning_messages = [
"Failed to send Pushover notification to ALL_DEVICES: Unauthorized - Invalid Token., error=401."
]
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
warning_messages: list[str] = []
def _fake_warning(message, *args, **kwargs):
_ = kwargs
warning_messages.append(message % args if args else str(message))
monkeypatch.setattr(notifications_module.logger, "warning", _fake_warning)
result = notifications_module._dispatch_to_apprise(
["pover://user_key@app_token"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is False
assert any(
"pover: apprise.plugins.pushover: Failed to send Pushover notification" in detail
for detail in result.get("details", [])
)
assert any(
"Apprise source [apprise.plugins.pushover]: Failed to send Pushover notification" in msg
for msg in warning_messages
)
def test_dispatch_to_apprise_logs_add_exception_at_debug_with_trace(monkeypatch):
fake_apprise = _FakeAppriseModule()
fake_apprise.client.instantiate_exceptions_by_url = {
"ntfys://ntfy.sh/shelfmark": RuntimeError("add exploded"),
}
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
debug_calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
def _fake_debug(message, *args, **kwargs):
debug_calls.append((str(message), args, kwargs))
monkeypatch.setattr(notifications_module.logger, "debug", _fake_debug)
result = notifications_module._dispatch_to_apprise(
["ntfys://ntfy.sh/shelfmark"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is False
assert result["message"] == "No valid notification URLs configured"
assert result["details"] == ["ntfys: route registration failed (RuntimeError: add exploded)"]
assert any(
"Apprise route registration raised RuntimeError" in (message % args if args else message)
and kwargs.get("exc_info")
== (
RuntimeError,
fake_apprise.client.instantiate_exceptions_by_url["ntfys://ntfy.sh/shelfmark"],
fake_apprise.client.instantiate_exceptions_by_url[
"ntfys://ntfy.sh/shelfmark"
].__traceback__,
)
for message, args, kwargs in debug_calls
)
def test_dispatch_to_apprise_logs_notify_exception_at_debug_with_trace(monkeypatch):
fake_apprise = _FakeAppriseModule()
fake_apprise.client.notify_exceptions_by_url = {
"ntfys://ntfy.sh/shelfmark": RuntimeError("notify exploded"),
}
monkeypatch.setattr(notifications_module, "apprise", fake_apprise)
debug_calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
def _fake_debug(message, *args, **kwargs):
debug_calls.append((str(message), args, kwargs))
monkeypatch.setattr(notifications_module.logger, "debug", _fake_debug)
result = notifications_module._dispatch_to_apprise(
["ntfys://ntfy.sh/shelfmark"],
title="Test",
body="Body",
notify_type=_FakeNotifyType.INFO,
)
assert result["success"] is False
assert result["message"] == "Notification delivery failed"
assert result["details"] == ["ntfys: notify raised RuntimeError: notify exploded"]
assert any(
"Apprise notify raised RuntimeError" in (message % args if args else message)
and kwargs.get("exc_info")
== (
RuntimeError,
fake_apprise.client.notify_exceptions_by_url["ntfys://ntfy.sh/shelfmark"],
fake_apprise.client.notify_exceptions_by_url["ntfys://ntfy.sh/shelfmark"].__traceback__,
)
for message, args, kwargs in debug_calls
)
def test_resolve_admin_routes_returns_empty_when_no_routes(monkeypatch):
def _fake_get(key, default=None):
if key == "ADMIN_NOTIFICATION_ROUTES":
return []
return default
monkeypatch.setattr(notifications_module.app_config, "get", _fake_get)
routes = notifications_module._resolve_admin_routes()
assert routes == []
def test_resolve_user_routes_uses_user_overrides(monkeypatch):
def _fake_get(key, default=None, user_id=None):
if user_id != 42:
return default
values = {
"USER_NOTIFICATION_ROUTES": [
{"event": "all", "url": " ntfys://ntfy.sh/alice "},
{"event": "download_failed", "url": "ntfys://ntfy.sh/errors"},
{"event": "download_failed", "url": "ntfys://ntfy.sh/errors"},
],
}
return values.get(key, default)
monkeypatch.setattr(notifications_module.app_config, "get", _fake_get)
routes = notifications_module._resolve_user_routes(42)
assert routes == [
{"event": "all", "url": "ntfys://ntfy.sh/alice"},
{"event": "download_failed", "url": "ntfys://ntfy.sh/errors"},
]
def test_notify_user_submits_non_blocking_when_route_matches_event(monkeypatch):
fake_executor = _FakeExecutor()
monkeypatch.setattr(notifications_module, "_executor", fake_executor)
monkeypatch.setattr(
notifications_module,
"_resolve_user_routes",
lambda _user_id: [{"event": "download_failed", "url": "discord://Webhook/Token"}],
)
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.DOWNLOAD_FAILED,
title="Example Book",
author="Example Author",
username="reader",
)
notifications_module.notify_user(
7,
notifications_module.NotificationEvent.DOWNLOAD_FAILED,
context,
)
assert len(fake_executor.calls) == 1
def test_notify_user_skips_when_user_id_is_invalid(monkeypatch):
fake_executor = _FakeExecutor()
monkeypatch.setattr(notifications_module, "_executor", fake_executor)
context = notifications_module.NotificationContext(
event=notifications_module.NotificationEvent.DOWNLOAD_COMPLETE,
title="Example Book",
author="Example Author",
)
notifications_module.notify_user(
None,
notifications_module.NotificationEvent.DOWNLOAD_COMPLETE,
context,
)
assert fake_executor.calls == []
def test_resolve_route_urls_for_event_includes_all_and_specific_rows():
routes = [
{"event": "all", "url": "ntfys://ntfy.sh/all"},
{"event": "download_failed", "url": "ntfys://ntfy.sh/errors"},
{"event": "request_created", "url": "ntfys://ntfy.sh/requests"},
]
urls = notifications_module._resolve_route_urls_for_event(
routes,
notifications_module.NotificationEvent.DOWNLOAD_FAILED,
)
assert urls == [
"ntfys://ntfy.sh/all",
"ntfys://ntfy.sh/errors",
]
def test_resolve_route_urls_for_event_deduplicates_matching_urls():
routes = [
{"event": "all", "url": "ntfys://ntfy.sh/shared"},
{"event": "download_failed", "url": "ntfys://ntfy.sh/shared"},
{"event": "download_failed", "url": "ntfys://ntfy.sh/errors"},
]
urls = notifications_module._resolve_route_urls_for_event(
routes,
notifications_module.NotificationEvent.DOWNLOAD_FAILED,
)
assert urls == [
"ntfys://ntfy.sh/shared",
"ntfys://ntfy.sh/errors",
]
def test_resolve_admin_routes_expands_multiselect_event_rows(monkeypatch):
def _fake_get(key, default=None):
if key == "ADMIN_NOTIFICATION_ROUTES":
return [
{"event": ["request_created", "download_failed"], "url": "ntfys://ntfy.sh/multi"},
{"event": ["all", "download_complete"], "url": "ntfys://ntfy.sh/all"},
]
return default
monkeypatch.setattr(notifications_module.app_config, "get", _fake_get)
routes = notifications_module._resolve_admin_routes()
assert routes == [
{"event": "request_created", "url": "ntfys://ntfy.sh/multi"},
{"event": "download_failed", "url": "ntfys://ntfy.sh/multi"},
{"event": "all", "url": "ntfys://ntfy.sh/all"},
]
def test_resolve_user_routes_expands_multiselect_event_rows(monkeypatch):
def _fake_get(key, default=None, user_id=None):
if key != "USER_NOTIFICATION_ROUTES" or user_id != 7:
return default
return [
{
"event": ["download_complete", "request_fulfilled"],
"url": "ntfys://ntfy.sh/user-main",
},
{"event": ["all", "download_failed"], "url": "ntfys://ntfy.sh/user-all"},
]
monkeypatch.setattr(notifications_module.app_config, "get", _fake_get)
routes = notifications_module._resolve_user_routes(7)
assert routes == [
{"event": "download_complete", "url": "ntfys://ntfy.sh/user-main"},
{"event": "request_fulfilled", "url": "ntfys://ntfy.sh/user-main"},
{"event": "all", "url": "ntfys://ntfy.sh/user-all"},
]