mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-10-05 22:05:50 +01:00
Feature: Notification support + Enhanced request management (#618)
- Added notification support via Apprise dependency - Notifications can be configured globally or per user, with full customization of events and notification type. - Added expanded ActivityCard for increased detail of each request, file info, and managing the attached file. - Enhanced tests
This commit is contained in:
Alex
@@ -491,6 +491,42 @@ class TestAdminUserUpdateEndpoint:
|
||||
settings = user_db.get_user_settings(user["id"])
|
||||
assert settings["DESTINATION_AUDIOBOOK"] == "/audiobooks/alice"
|
||||
|
||||
def test_update_user_settings_accepts_notification_overrides(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
resp = admin_client.put(
|
||||
f"/api/admin/users/{user['id']}",
|
||||
json={
|
||||
"settings": {
|
||||
"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"},
|
||||
],
|
||||
}
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
settings = user_db.get_user_settings(user["id"])
|
||||
assert settings["USER_NOTIFICATION_ROUTES"] == [
|
||||
{"event": ["all"], "url": "ntfys://ntfy.sh/alice"},
|
||||
{"event": ["download_failed"], "url": "ntfys://ntfy.sh/errors"},
|
||||
]
|
||||
|
||||
def test_update_user_settings_rejects_invalid_notification_url(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
resp = admin_client.put(
|
||||
f"/api/admin/users/{user['id']}",
|
||||
json={"settings": {"USER_NOTIFICATION_ROUTES": [{"event": "all", "url": "not-a-valid-url"}]}},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json["error"] == "Invalid settings payload"
|
||||
assert any(
|
||||
"Invalid value for USER_NOTIFICATION_ROUTES" in msg
|
||||
for msg in resp.json["details"]
|
||||
)
|
||||
|
||||
def test_update_user_settings_accepts_valid_request_policy_rule(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
@@ -1115,6 +1151,186 @@ class TestAdminDeliveryPreferences:
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/admin/users/<id>/notification-preferences
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAdminNotificationPreferences:
|
||||
"""Tests for GET /api/admin/users/<id>/notification-preferences."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_config(self, tmp_path, monkeypatch):
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
config_dir = str(tmp_path)
|
||||
monkeypatch.setenv("CONFIG_DIR", config_dir)
|
||||
monkeypatch.setattr("shelfmark.config.env.CONFIG_DIR", Path(config_dir))
|
||||
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
notifications_config = {
|
||||
"ADMIN_NOTIFICATION_ROUTES": [
|
||||
{"event": "all", "url": "ntfys://ntfy.sh/admin"},
|
||||
{"event": "download_failed", "url": "ntfys://ntfy.sh/admin-errors"},
|
||||
],
|
||||
"USER_NOTIFICATION_ROUTES": [
|
||||
{"event": "all", "url": "ntfys://ntfy.sh/default-user"},
|
||||
],
|
||||
}
|
||||
(plugins_dir / "notifications.json").write_text(json.dumps(notifications_config))
|
||||
|
||||
from shelfmark.core.config import config as app_config
|
||||
app_config.refresh()
|
||||
|
||||
def test_returns_curated_fields_and_effective_values(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
user_db.set_user_settings(
|
||||
user["id"],
|
||||
{
|
||||
"USER_NOTIFICATION_ROUTES": [
|
||||
{"event": "all", "url": "ntfys://ntfy.sh/alice"},
|
||||
{"event": "download_failed", "url": "ntfys://ntfy.sh/alice-errors"},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
resp = admin_client.get(f"/api/admin/users/{user['id']}/notification-preferences")
|
||||
assert resp.status_code == 200
|
||||
|
||||
data = resp.json
|
||||
assert data["tab"] == "notifications"
|
||||
assert data["keys"] == [
|
||||
"USER_NOTIFICATION_ROUTES",
|
||||
]
|
||||
|
||||
field_keys = [field["key"] for field in data["fields"]]
|
||||
assert set(field_keys) == set(data["keys"])
|
||||
|
||||
assert data["userOverrides"]["USER_NOTIFICATION_ROUTES"] == [
|
||||
{"event": "all", "url": "ntfys://ntfy.sh/alice"},
|
||||
{"event": "download_failed", "url": "ntfys://ntfy.sh/alice-errors"},
|
||||
]
|
||||
|
||||
assert data["effective"]["USER_NOTIFICATION_ROUTES"]["source"] == "user_override"
|
||||
|
||||
def test_returns_404_for_unknown_user(self, admin_client):
|
||||
resp = admin_client.get("/api/admin/users/9999/notification-preferences")
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_requires_admin(self, regular_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
resp = regular_client.get(f"/api/admin/users/{user['id']}/notification-preferences")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/admin/users/<id>/notification-preferences/test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAdminNotificationPreferencesTestAction:
|
||||
"""Tests for POST /api/admin/users/<id>/notification-preferences/test."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_config(self, tmp_path, monkeypatch):
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
config_dir = str(tmp_path)
|
||||
monkeypatch.setenv("CONFIG_DIR", config_dir)
|
||||
monkeypatch.setattr("shelfmark.config.env.CONFIG_DIR", Path(config_dir))
|
||||
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
notifications_config = {
|
||||
"ADMIN_NOTIFICATION_ROUTES": [
|
||||
{"event": "all", "url": "ntfys://ntfy.sh/admin"},
|
||||
],
|
||||
"USER_NOTIFICATION_ROUTES": [
|
||||
{"event": "all", "url": "ntfys://ntfy.sh/default-user"},
|
||||
],
|
||||
}
|
||||
(plugins_dir / "notifications.json").write_text(json.dumps(notifications_config))
|
||||
|
||||
from shelfmark.core.config import config as app_config
|
||||
app_config.refresh()
|
||||
|
||||
def test_requires_admin(self, regular_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
resp = regular_client.post(
|
||||
f"/api/admin/users/{user['id']}/notification-preferences/test",
|
||||
json={"USER_NOTIFICATION_ROUTES": [{"event": "all", "url": "ntfys://ntfy.sh/alice"}]},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_returns_404_for_unknown_user(self, admin_client):
|
||||
resp = admin_client.post("/api/admin/users/9999/notification-preferences/test", json={})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_uses_payload_routes_when_provided(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
with patch(
|
||||
"shelfmark.config.notifications_settings.send_test_notification",
|
||||
return_value={"success": True, "message": "ok"},
|
||||
) as mock_send:
|
||||
resp = admin_client.post(
|
||||
f"/api/admin/users/{user['id']}/notification-preferences/test",
|
||||
json={
|
||||
"USER_NOTIFICATION_ROUTES": [
|
||||
{"event": "all", "url": " ntfys://ntfy.sh/alice "},
|
||||
{"event": "download_failed", "url": "ntfys://ntfy.sh/alice-errors"},
|
||||
{"event": "download_failed", "url": "ntfys://ntfy.sh/alice-errors"},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json["success"] is True
|
||||
mock_send.assert_called_once_with(
|
||||
["ntfys://ntfy.sh/alice", "ntfys://ntfy.sh/alice-errors"]
|
||||
)
|
||||
|
||||
def test_uses_effective_routes_when_payload_missing(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
with patch(
|
||||
"shelfmark.config.notifications_settings.send_test_notification",
|
||||
return_value={"success": True, "message": "ok"},
|
||||
) as mock_send:
|
||||
resp = admin_client.post(
|
||||
f"/api/admin/users/{user['id']}/notification-preferences/test",
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json["success"] is True
|
||||
mock_send.assert_called_once_with(["ntfys://ntfy.sh/default-user"])
|
||||
|
||||
def test_rejects_invalid_urls(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
resp = admin_client.post(
|
||||
f"/api/admin/users/{user['id']}/notification-preferences/test",
|
||||
json={"USER_NOTIFICATION_ROUTES": [{"event": "all", "url": "not-a-valid-url"}]},
|
||||
)
|
||||
|
||||
assert resp.status_code == 400
|
||||
assert "invalid personal notification URL" in resp.json["message"]
|
||||
|
||||
def test_requires_at_least_one_url(self, admin_client, user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
|
||||
resp = admin_client.post(
|
||||
f"/api/admin/users/{user['id']}/notification-preferences/test",
|
||||
json={"USER_NOTIFICATION_ROUTES": [{"event": "all", "url": ""}]},
|
||||
)
|
||||
|
||||
assert resp.status_code == 400
|
||||
assert "Add at least one personal notification URL route first." in resp.json["message"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/admin/settings/overrides-summary
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user