diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py
index e59d6f7..6693c82 100644
--- a/discord_rss_bot/main.py
+++ b/discord_rss_bot/main.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import concurrent.futures
import io
import json
import logging
@@ -15,6 +16,7 @@ from datetime import datetime
from functools import lru_cache
from html import escape
from html import unescape
+from pathlib import Path
from typing import TYPE_CHECKING
from typing import Annotated
from typing import TypedDict
@@ -61,6 +63,7 @@ from discord_rss_bot.custom_message import get_message_username
from discord_rss_bot.custom_message import replace_tags_in_text_message
from discord_rss_bot.custom_message import save_embed
from discord_rss_bot.feeds import FeedUpdateError
+from discord_rss_bot.feeds import JsonValue
from discord_rss_bot.feeds import SentWebhookRecord
from discord_rss_bot.feeds import coerce_media_gallery_image_limit
from discord_rss_bot.feeds import coerce_webhook_text_length_limit
@@ -91,12 +94,15 @@ from discord_rss_bot.git_backup import commit_state_change
from discord_rss_bot.git_backup import get_backup_path
from discord_rss_bot.is_url_valid import is_url_valid
from discord_rss_bot.search import create_search_context
+from discord_rss_bot.settings import data_dir
+from discord_rss_bot.settings import default_custom_embed
+from discord_rss_bot.settings import default_custom_message
from discord_rss_bot.settings import get_reader
+from discord_rss_bot.settings import make_app_reader
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
from collections.abc import Iterable
- from pathlib import Path
from reader.types import JSONType
@@ -2649,12 +2655,439 @@ async def manual_backup(
return RedirectResponse(url=f"/?message={urllib.parse.quote(message)}", status_code=303)
+def _get_grouped_feeds(reader: Reader) -> list[dict[str, typing.Any]]:
+ """Build a list of webhook groups with pre-computed indices for template use.
+
+ Each group dict contains:
+ name: The webhook name (or "Orphaned (no webhook)").
+ group_idx: 1-based index for the group.
+ feeds: List of dicts with:
+ feed: The Feed object.
+ feed_idx: 1-based index within the group.
+
+ Returns:
+ list[dict]: Grouped feeds with pre-computed indices for template rendering.
+ """
+ hooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+
+ feeds_by_webhook: dict[str, list[Feed]] = {}
+ orphaned: list[Feed] = []
+
+ for feed in reader.get_feeds():
+ feed_webhook: str = str(reader.get_tag(feed.url, "webhook", ""))
+ hook_name: str = ""
+ for hook in hooks:
+ if hook["url"] == feed_webhook:
+ hook_name = hook["name"]
+ break
+ if hook_name:
+ feeds_by_webhook.setdefault(hook_name, []).append(feed)
+ else:
+ orphaned.append(feed)
+
+ grouped: list[dict[str, typing.Any]] = []
+ for group_idx, (name, feed_list) in enumerate(feeds_by_webhook.items(), start=1):
+ grouped.append({
+ "name": name,
+ "group_idx": group_idx,
+ "feeds": [{"feed": f, "feed_idx": idx} for idx, f in enumerate(feed_list, start=1)],
+ })
+ if orphaned:
+ grouped.append({
+ "name": "Orphaned (no webhook)",
+ "group_idx": len(grouped) + 1,
+ "feeds": [{"feed": f, "feed_idx": idx} for idx, f in enumerate(orphaned, start=1)],
+ })
+
+ return grouped
+
+
+@app.get("/mass", response_class=HTMLResponse)
+async def get_mass(
+ request: Request,
+ reader: Annotated[Reader, Depends(get_reader_dependency)],
+ active_tab: str = "create",
+) -> HTMLResponse:
+ """Mass operations page: create, delete, or modify feeds in bulk.
+
+ Args:
+ request: The request object.
+ reader: The Reader instance.
+ active_tab: The active tab (create, delete, modify).
+
+ Returns:
+ HTMLResponse: The mass operations page.
+ """
+ hooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+
+ context: dict[str, typing.Any] = {
+ "request": request,
+ "webhooks": hooks,
+ "all_feeds_grouped": _get_grouped_feeds(reader),
+ "active_tab": active_tab,
+ }
+ return templates.TemplateResponse(request=request, name="mass.html", context=context)
+
+
+def _create_and_tag_feed(reader: Reader, feed_url: str, webhook_url: str) -> str | None:
+ """Add a feed and set its tags, without updating.
+
+ Returns:
+ The feed URL on success, or None if adding failed.
+ """
+ clean_url: str = feed_url.strip()
+ try:
+ reader.add_feed(clean_url)
+ except FeedExistsError:
+ pass
+ except ReaderError:
+ return None
+
+ reader.set_tag(clean_url, "webhook", webhook_url) # pyright: ignore[reportArgumentType]
+ reader.set_tag(clean_url, "save_sent_webhooks", True) # pyright: ignore[reportArgumentType]
+ reader.set_tag(clean_url, "media_gallery_image_limit", cast("JSONType", 1))
+
+ global_webhook_text_length_limit: int = coerce_webhook_text_length_limit(
+ cast("JsonValue", reader.get_tag((), "webhook_text_length_limit", 4000)), # pyright: ignore[reportArgumentType]
+ )
+ reader.set_tag(clean_url, "webhook_text_length_limit", cast("JSONType", global_webhook_text_length_limit))
+ reader.set_tag(clean_url, "custom_message", default_custom_message) # pyright: ignore[reportArgumentType]
+
+ global_screenshot_layout: str = str(reader.get_tag((), "screenshot_layout", "desktop")).strip().lower()
+ if global_screenshot_layout not in {"desktop", "mobile"}:
+ global_screenshot_layout = "desktop"
+ reader.set_tag(clean_url, "screenshot_layout", global_screenshot_layout) # pyright: ignore[reportArgumentType]
+
+ global_delivery_mode: str = str(reader.get_tag((), "delivery_mode", "embed")).strip().lower()
+ if global_delivery_mode not in {"embed", "text"}:
+ global_delivery_mode = "embed"
+ reader.set_tag(clean_url, "delivery_mode", global_delivery_mode) # pyright: ignore[reportArgumentType]
+ reader.set_tag(clean_url, "should_send_embed", global_delivery_mode == "embed") # pyright: ignore[reportArgumentType]
+ reader.set_tag(clean_url, "embed", json.dumps(default_custom_embed)) # pyright: ignore[reportArgumentType]
+
+ return clean_url
+
+
+def _modify_single_feed( # ruff:ignore[complex-structure, too-many-branches, too-many-statements]
+ reader: Reader,
+ url: str,
+ modify_action: str,
+ modify_value: str,
+) -> dict[str, typing.Any]:
+ """Apply a modification action to a single feed and return the result.
+
+ Args:
+ reader: The Reader instance.
+ url: The feed URL to modify.
+ modify_action: The action to perform.
+ modify_value: The value for the action.
+
+ Returns:
+ dict: Result with url, success, error, and action_taken keys.
+ """
+ result: dict[str, typing.Any] = {
+ "url": url,
+ "success": False,
+ "error": "",
+ "action_taken": "",
+ }
+
+ try:
+ reader.get_feed(url)
+ except FeedNotFoundError:
+ result["error"] = "Feed not found"
+ return result
+
+ if modify_action == "pause":
+ reader.disable_feed_updates(url)
+ result["success"] = True
+ result["action_taken"] = "Paused"
+ elif modify_action == "unpause":
+ reader.enable_feed_updates(url)
+ result["success"] = True
+ result["action_taken"] = "Unpaused"
+ elif modify_action == "change_webhook":
+ webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+ webhook_url: str = ""
+ for hook in webhooks:
+ if hook["name"] == modify_value:
+ webhook_url = hook["url"]
+ break
+ if webhook_url:
+ reader.set_tag(url, "webhook", webhook_url) # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = f"Webhook changed to {modify_value}"
+ else:
+ result["error"] = f"Webhook '{modify_value}' not found"
+ elif modify_action == "delivery_mode":
+ if modify_value == "embed":
+ reader.set_tag(url, "delivery_mode", "embed") # pyright: ignore[reportArgumentType]
+ reader.set_tag(url, "should_send_embed", True) # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = "Delivery mode set to embed"
+ elif modify_value == "text":
+ reader.set_tag(url, "delivery_mode", "text") # pyright: ignore[reportArgumentType]
+ reader.set_tag(url, "should_send_embed", False) # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = "Delivery mode set to text"
+ elif modify_value == "screenshot_desktop":
+ reader.set_tag(url, "delivery_mode", "screenshot") # pyright: ignore[reportArgumentType]
+ reader.set_tag(url, "screenshot_layout", "desktop") # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = "Delivery mode set to screenshot (desktop)"
+ elif modify_value == "screenshot_mobile":
+ reader.set_tag(url, "delivery_mode", "screenshot") # pyright: ignore[reportArgumentType]
+ reader.set_tag(url, "screenshot_layout", "mobile") # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = "Delivery mode set to screenshot (mobile)"
+ else:
+ result["error"] = f"Unknown delivery mode: {modify_value}"
+ elif modify_action == "screenshot_layout":
+ if modify_value in {"desktop", "mobile"}:
+ reader.set_tag(url, "screenshot_layout", modify_value) # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = f"Screenshot layout set to {modify_value}"
+ else:
+ result["error"] = f"Unknown layout: {modify_value}"
+ elif modify_action == "update_interval":
+ try:
+ interval: int = int(modify_value)
+ except ValueError as e:
+ result["error"] = str(e)
+ return result
+ if interval < 1:
+ result["error"] = "Interval must be at least 1 minute"
+ else:
+ reader.set_tag(url, ".reader.update", {"interval": interval}) # pyright: ignore[reportArgumentType]
+ result["success"] = True
+ result["action_taken"] = f"Update interval set to {interval} minute(s)" # ruff:ignore[hardcoded-sql-expression]
+ else:
+ result["error"] = f"Unknown action: {modify_action}"
+
+ return result
+
+
+def _update_and_mark_read(db_path: Path, feed_url: str) -> tuple[str, bool, str]:
+ """Update a feed and mark entries as read, in its own reader instance.
+
+ Called from worker threads to parallelize HTTP fetches.
+
+ Returns:
+ (feed_url, success, error_message)
+ """
+ worker_reader: Reader = make_app_reader(db_path)
+ try:
+ worker_reader.update_feed(feed_url)
+ for entry in worker_reader.get_entries(feed=feed_url, read=False):
+ worker_reader.set_entry_read(entry, True)
+ except ReaderError as e:
+ logger.warning("Failed to update feed %s: %s", feed_url, e)
+ return feed_url, False, str(e)[:200]
+ except Exception as e:
+ logger.exception("Unexpected error updating feed %s", feed_url)
+ return feed_url, False, str(e)[:200]
+ finally:
+ worker_reader.close()
+ return feed_url, True, ""
+
+
+@app.post("/mass/create", response_class=HTMLResponse)
+async def post_mass_create( # ruff:ignore[complex-structure]
+ request: Request,
+ feed_urls: Annotated[str, Form()],
+ webhook_dropdown: Annotated[str, Form()],
+ reader: Annotated[Reader, Depends(get_reader_dependency)],
+) -> HTMLResponse:
+ """Create multiple feeds at once.
+
+ Phase 1: Add feeds and set tags (sequential, fast, no HTTP).
+ Phase 2: Update feeds in parallel via a thread pool.
+
+ Args:
+ request: The request object.
+ feed_urls: Feed URLs (one per line).
+ webhook_dropdown: The webhook to attach feeds to.
+ reader: The Reader instance.
+
+ Returns:
+ HTMLResponse: The mass operations page with results.
+
+ Raises:
+ HTTPException: If the selected webhook is not found.
+ """
+ urls: list[str] = [url.strip() for url in feed_urls.strip().split("\n") if url.strip()]
+ results: list[dict[str, typing.Any]] = []
+
+ webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+
+ # Resolve webhook name to URL once for all feeds
+ webhook_url: str = ""
+ for hook in webhooks:
+ if hook["name"] == webhook_dropdown:
+ webhook_url = hook["url"]
+ break
+
+ if not webhook_url:
+ raise HTTPException(status_code=404, detail="Webhook not found")
+
+ # Phase 1: Add feeds and set tags (sequential, fast)
+ urls_to_update: list[str] = []
+ for url in urls:
+ added_url: str | None = _create_and_tag_feed(reader, url, webhook_url)
+ if added_url:
+ urls_to_update.append(added_url)
+ results.append({"url": url, "success": False, "feed_url": None, "error": ""})
+ continue
+ results.append({"url": url, "success": False, "feed_url": None, "error": "Failed to add feed"})
+
+ # Phase 2: Update feeds in parallel
+ if urls_to_update:
+ db_path: Path = Path(data_dir) / "db.sqlite"
+ max_workers: int = min(10, len(urls_to_update))
+ with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+ future_to_url = {executor.submit(_update_and_mark_read, db_path, u): u for u in urls_to_update}
+ for future in concurrent.futures.as_completed(future_to_url):
+ original_url: str = future_to_url[future]
+ feed_url, update_success, error_msg = future.result()
+ # Find and update the matching result entry
+ for result in results:
+ if result["url"] == original_url:
+ result["success"] = update_success
+ result["feed_url"] = feed_url if update_success else None
+ result["error"] = error_msg
+ break
+
+ reader.update_search()
+
+ success_count: int = sum(1 for r in results if r["success"])
+ if success_count > 0:
+ commit_state_change(reader, f"Mass create {success_count} feed(s)")
+
+ context: dict[str, typing.Any] = {
+ "request": request,
+ "webhooks": webhooks,
+ "all_feeds_grouped": _get_grouped_feeds(reader),
+ "active_tab": "create",
+ "feed_urls": feed_urls,
+ "selected_webhook": webhook_dropdown,
+ "create_results": results,
+ "message": f"Created {success_count} of {len(results)} feed(s).",
+ }
+ return templates.TemplateResponse(request=request, name="mass.html", context=context)
+
+
+@app.post("/mass/delete", response_class=HTMLResponse)
+async def post_mass_delete(
+ request: Request,
+ reader: Annotated[Reader, Depends(get_reader_dependency)],
+ feed_urls: Annotated[list[str] | None, Form()] = None,
+) -> HTMLResponse:
+ """Delete multiple feeds at once.
+
+ Args:
+ request: The request object.
+ reader: The Reader instance.
+ feed_urls: List of feed URLs to delete.
+
+ Returns:
+ HTMLResponse: The mass operations page with results.
+ """
+ if feed_urls is None:
+ feed_urls = []
+ results: list[dict[str, typing.Any]] = []
+
+ for url in feed_urls:
+ result: dict[str, typing.Any] = {"url": url, "success": False, "error": ""}
+ try:
+ reader.delete_feed(url)
+ result["success"] = True
+ except FeedNotFoundError:
+ result["error"] = "Feed not found"
+ except Exception as e: # ruff:ignore[blind-except]
+ result["error"] = str(e)[:200]
+ results.append(result)
+
+ deleted_count: int = sum(1 for r in results if r["success"])
+ if deleted_count > 0:
+ commit_state_change(reader, f"Mass delete {deleted_count} feed(s)")
+
+ webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+
+ failed_count: int = len(results) - deleted_count
+ context: dict[str, typing.Any] = {
+ "request": request,
+ "webhooks": webhooks,
+ "all_feeds_grouped": _get_grouped_feeds(reader),
+ "active_tab": "delete",
+ "delete_results": results,
+ "delete_summary": {"deleted": deleted_count, "failed": failed_count},
+ "message": f"Deleted {deleted_count} feed(s). {failed_count} failed.",
+ }
+ return templates.TemplateResponse(request=request, name="mass.html", context=context)
+
+
+@app.post("/mass/modify", response_class=HTMLResponse)
+async def post_mass_modify(
+ request: Request,
+ reader: Annotated[Reader, Depends(get_reader_dependency)],
+ feed_urls: Annotated[list[str] | None, Form()] = None,
+ modify_action: Annotated[str, Form()] = "",
+ modify_value: Annotated[str, Form()] = "",
+) -> HTMLResponse:
+ """Modify multiple feeds at once.
+
+ Args:
+ request: The request object.
+ reader: The Reader instance.
+ feed_urls: List of feed URLs to modify.
+ modify_action: The action to perform (pause, unpause, change_webhook, etc.).
+ modify_value: The value for the action.
+
+ Returns:
+ HTMLResponse: The mass operations page with results.
+ """
+ feed_urls_list: list[str] = feed_urls if feed_urls is not None else []
+ results: list[dict[str, typing.Any]] = []
+
+ for url in feed_urls_list:
+ try:
+ result = _modify_single_feed(reader, url, modify_action, modify_value)
+ except Exception as e:
+ logger.exception("Failed to modify feed %s", url)
+ result = {"url": url, "success": False, "error": str(e)[:200], "action_taken": ""}
+ results.append(result)
+
+ modified_count: int = sum(1 for r in results if r["success"])
+ failed_count: int = sum(1 for r in results if not r["success"])
+ if modified_count > 0:
+ commit_state_change(reader, f"Mass modify {modified_count} feed(s)")
+
+ webhooks = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+
+ context: dict[str, typing.Any] = {
+ "request": request,
+ "webhooks": webhooks,
+ "all_feeds_grouped": _get_grouped_feeds(reader),
+ "active_tab": "modify",
+ "modify_results": results,
+ "modify_action": modify_action,
+ "modify_value": modify_value,
+ "modify_summary": {
+ "modified": modified_count,
+ "failed": failed_count,
+ "skipped": 0,
+ },
+ "message": f"Modified {modified_count} feed(s). {failed_count} failed.",
+ }
+ return templates.TemplateResponse(request=request, name="mass.html", context=context)
+
+
@app.get("/search", response_class=HTMLResponse)
async def search(
request: Request,
query: str,
reader: Annotated[Reader, Depends(get_reader_dependency)],
-):
+) -> HTMLResponse:
"""Get entries matching a full-text search query.
Args:
diff --git a/discord_rss_bot/templates/mass.html b/discord_rss_bot/templates/mass.html
new file mode 100644
index 0000000..88a6af8
--- /dev/null
+++ b/discord_rss_bot/templates/mass.html
@@ -0,0 +1,355 @@
+{% extends "base.html" %}
+{% block title %}
+ Mass operations | discord-rss-bot
+{% endblock title %}
+{% block description %}
+ Mass create, delete, or modify RSS feeds in bulk.
+{% endblock description %}
+{% block content %}
+ {% if message %}
+
+ {% endif %}
+
+ -
+
+
+ -
+
+
+ -
+
+
+
+
+ {# ===== Mass Create ===== #}
+
+
+
Create multiple feeds
+
Enter one feed URL per line. All feeds will be attached to the selected webhook.
+
+
+ {% if create_results %}
+
+
Results
+
+ {% for result in create_results %}
+ -
+ {{ result.url }}
+ {% if result.success %}
+ ✓ Added
+ {% if result.feed_url %}
+ - View feed
+ {% endif %}
+ {% else %}
+ ✗ {{ result.error }}
+ {% endif %}
+
+ {% endfor %}
+
+
+ {% endif %}
+
+ {# ===== Mass Delete ===== #}
+
+
+
Delete multiple feeds
+
Select feeds to delete. This action cannot be undone.
+ {% if all_feeds_grouped %}
+
+ {% else %}
+
No feeds found.
+ {% endif %}
+
+ {% if delete_results %}
+
+
Results
+
+ {% for result in delete_results %}
+ -
+ {{ result.url }}
+ {% if result.success %}
+ ✓ Deleted
+ {% else %}
+ ✗ {{ result.error }}
+ {% endif %}
+
+ {% endfor %}
+ - {{ delete_summary.deleted }} deleted, {{ delete_summary.failed }} failed
+
+
+ {% endif %}
+
+ {# ===== Mass Modify ===== #}
+
+
+
Modify multiple feeds
+
Select feeds and choose a setting to update.
+ {% if all_feeds_grouped %}
+
+ {% else %}
+
No feeds found.
+ {% endif %}
+
+ {% if modify_results %}
+
+
Results
+
+ {% for result in modify_results %}
+ -
+ {{ result.url }}
+ {% if result.success %}
+ ✓ {{ result.action_taken }}
+ {% else %}
+ ✗ {{ result.error }}
+ {% endif %}
+
+ {% endfor %}
+ -
+ {{ modify_summary.modified }} modified, {{ modify_summary.failed }} failed, {{ modify_summary.skipped }} skipped
+
+
+
+ {% endif %}
+
+
+{% endblock content %}
+{% block head %}
+
+{% endblock head %}
diff --git a/discord_rss_bot/templates/nav.html b/discord_rss_bot/templates/nav.html
index 1fb1728..d0a5cdf 100644
--- a/discord_rss_bot/templates/nav.html
+++ b/discord_rss_bot/templates/nav.html
@@ -18,6 +18,10 @@
|
{% endif %}
+
+ Mass ops
+
+ |
Webhooks
diff --git a/tests/test_main.py b/tests/test_main.py
index b69f06d..ee1b12a 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -14,8 +14,10 @@ from typing import cast
from unittest.mock import MagicMock
from unittest.mock import patch
+import pytest
from fastapi.testclient import TestClient
from reader import FeedExistsError
+from reader import FeedNotFoundError
import discord_rss_bot.main as main_module
from discord_rss_bot import feeds
@@ -26,7 +28,6 @@ from discord_rss_bot.main import get_reader_dependency
if TYPE_CHECKING:
from pathlib import Path
- import pytest
from httpx2 import Response
from reader import Entry
from reader import Reader
@@ -3355,3 +3356,532 @@ def test_post_set_custom_clears_username_and_avatar() -> None:
assert not stub.set_tag.call_args_list[1].args[2]
finally:
app.dependency_overrides = {}
+
+
+# ---------------------------------------------------------------------------
+# Mass operations tests
+# ---------------------------------------------------------------------------
+
+
+class _StubFeedForMass:
+ """Minimal feed stub for mass operations tests."""
+
+ def __init__(self, url: str, title: str | None = None, updates_enabled: bool = True) -> None: # ruff:ignore[boolean-default-value-positional-argument, boolean-type-hint-positional-argument]
+ self.url: str = url
+ self.title: str | None = title
+ self.updates_enabled: bool = updates_enabled
+ self.last_exception: str | None = None
+
+
+class _StubReaderForMass:
+ """Stub reader with feeds grouped under a webhook, used by mass operations tests."""
+
+ def __init__(
+ self,
+ feeds: list[_StubFeedForMass] | None = None,
+ *,
+ webhook_name: str = "",
+ webhook_url: str = "",
+ delete_fail: bool = False,
+ ) -> None:
+ self._feeds: list[_StubFeedForMass] = feeds or []
+ self._webhook_name: str = webhook_name
+ self._webhook_url: str = webhook_url
+ self._delete_fail: bool = delete_fail
+ self.set_tag_calls: list[tuple[str, str, object]] = []
+ self.disabled_feeds: set[str] = set()
+ self.enabled_feeds: set[str] = set()
+
+ def get_feeds(self) -> list[_StubFeedForMass]:
+ """Return the list of feeds."""
+ return self._feeds
+
+ def get_tag(self, resource: tuple[()] | str, key: str, default: TestTagValue = None) -> TestTagValue: # ruff:ignore[too-many-return-statements]
+ """Return stored tag values."""
+ if resource == () and key == "webhooks":
+ if self._webhook_name:
+ return [{"name": self._webhook_name, "url": self._webhook_url}]
+ return []
+ if resource == () and key == "delivery_mode":
+ return "embed"
+ if resource == () and key == "webhook_text_length_limit":
+ return 4000
+ if resource == () and key == "screenshot_layout":
+ return "desktop"
+ if isinstance(resource, str) and key == "webhook":
+ return self._webhook_url
+ return default
+
+ def set_tag(self, resource: str, key: str, value: object) -> None: # pyright: ignore[reportArgumentType]
+ """Record tag set calls."""
+ self.set_tag_calls.append((resource, key, value))
+
+ def update_search(self) -> None:
+ """No-op search index update."""
+
+ def delete_feed(self, url: str) -> None:
+ """Delete a feed by URL.
+
+ Raises:
+ RuntimeError: If delete_fail was set in the constructor.
+ """
+ if self._delete_fail:
+ msg = "Delete failed"
+ raise RuntimeError(msg)
+ self._feeds = [f for f in self._feeds if f.url != url]
+
+ def get_feed(self, url: str) -> _StubFeedForMass:
+ """Get a feed by URL.
+
+ Returns:
+ The matching feed.
+
+ Raises:
+ FeedNotFoundError: If no feed matches the URL.
+ """
+ for feed in self._feeds:
+ if feed.url == url:
+ return feed
+ raise FeedNotFoundError(url)
+
+ def disable_feed_updates(self, url: str) -> None:
+ """Disable updates for a feed."""
+ self.disabled_feeds.add(url)
+ self.enabled_feeds.discard(url)
+
+ def enable_feed_updates(self, url: str) -> None:
+ """Enable updates for a feed."""
+ self.enabled_feeds.add(url)
+ self.disabled_feeds.discard(url)
+
+ def add_feed(self, url: str) -> None:
+ """Add a feed to the stub."""
+ self._feeds.append(_StubFeedForMass(url=url))
+
+
+MASS_FEED_1_URL: str = "https://example.com/feed1.xml"
+MASS_FEED_2_URL: str = "https://example.com/feed2.xml"
+MASS_WEBHOOK_NAME: str = "Mass test webhook"
+MASS_WEBHOOK_URL: str = "https://discord.com/api/webhooks/mass/abc"
+
+
+@pytest.fixture
+def mass_stub_reader() -> _StubReaderForMass:
+ """Create a stub reader with feeds for mass operations tests.
+
+ Returns:
+ A configured stub reader instance.
+ """
+ return _StubReaderForMass(
+ feeds=[
+ _StubFeedForMass(url=MASS_FEED_1_URL, title="Feed One"),
+ _StubFeedForMass(url=MASS_FEED_2_URL, title="Feed Two", updates_enabled=False),
+ ],
+ webhook_name=MASS_WEBHOOK_NAME,
+ webhook_url=MASS_WEBHOOK_URL,
+ )
+
+
+class TestMassPage:
+ """Tests for the mass operations page."""
+
+ def test_mass_page_loads(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """GET /mass should render all three tabs."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.get("/mass")
+ assert response.status_code == 200
+ assert "Mass Create" in response.text
+ assert "Mass Delete" in response.text
+ assert "Mass Modify" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_mass_page_shows_feeds_when_webhook_exists(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """GET /mass should list feeds grouped by webhook."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.get("/mass")
+ assert response.status_code == 200
+ assert MASS_WEBHOOK_NAME in response.text
+ assert "Feed One" in response.text
+ assert "Feed Two" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_mass_page_no_feeds_shows_empty_state(self) -> None:
+ """GET /mass with no feeds should show 'No feeds found'."""
+ stub = _StubReaderForMass(webhook_name=MASS_WEBHOOK_NAME, webhook_url=MASS_WEBHOOK_URL)
+ app.dependency_overrides[get_reader_dependency] = lambda: stub
+ try:
+ response: Response = client.get("/mass")
+ assert response.status_code == 200
+ assert "No feeds found" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_mass_page_defaults_to_create_tab(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """GET /mass without active_tab should default to create tab."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.get("/mass")
+ assert response.status_code == 200
+ assert 'id="create-tab"' in response.text
+ assert 'aria-selected="true"' in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_mass_page_respects_active_tab_param(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """GET /mass?active_tab=delete should show delete tab active."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.get("/mass?active_tab=delete")
+ assert response.status_code == 200
+ assert "Delete multiple feeds" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_mass_page_shows_webhooks_in_create_form(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """The mass create tab should list webhooks in the dropdown."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.get("/mass")
+ assert response.status_code == 200
+ assert MASS_WEBHOOK_NAME in response.text
+ assert "Create all feeds" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+
+class TestMassDelete:
+ """Tests for POST /mass/delete."""
+
+ def test_delete_selected_feeds(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/delete with feed URLs should delete them."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/delete",
+ data={"feed_urls": [MASS_FEED_1_URL, MASS_FEED_2_URL]},
+ )
+ assert response.status_code == 200
+ assert "Deleted 2 feed(s)" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_delete_no_selection(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/delete with no feed URLs should not crash."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.post("/mass/delete")
+ assert response.status_code == 200
+ assert "Deleted 0 feed(s)" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_delete_feed_error_does_not_crash(self) -> None:
+ """POST /mass/delete should handle reader errors gracefully."""
+ stub = _StubReaderForMass(
+ feeds=[_StubFeedForMass(url=MASS_FEED_1_URL)],
+ delete_fail=True,
+ )
+ app.dependency_overrides[get_reader_dependency] = lambda: stub
+ try:
+ response: Response = client.post("/mass/delete", data={"feed_urls": [MASS_FEED_1_URL]})
+ assert response.status_code == 200
+ assert "0 deleted, 1 failed" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+
+class TestMassModify:
+ """Tests for POST /mass/modify."""
+
+ def test_modify_pause_feed(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with pause action should disable feed updates."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "pause",
+ "modify_value": "",
+ },
+ )
+ assert response.status_code == 200
+ assert MASS_FEED_1_URL in response.text
+ assert "Modified 1 feed(s)" in response.text
+ assert MASS_FEED_1_URL in mass_stub_reader.disabled_feeds
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_unpause_feed(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with unpause action should enable feed updates."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_2_URL],
+ "modify_action": "unpause",
+ "modify_value": "",
+ },
+ )
+ assert response.status_code == 200
+ assert "Modified 1 feed(s)" in response.text
+ assert MASS_FEED_2_URL in mass_stub_reader.enabled_feeds
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_nonexistent_feed(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify for a non-existent feed should report error."""
+ nonexistent = "https://example.com/nonexistent.xml"
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [nonexistent],
+ "modify_action": "pause",
+ "modify_value": "",
+ },
+ )
+ assert response.status_code == 200
+ assert "0 modified, 1 failed" in response.text
+ assert "Feed not found" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_unknown_action(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with an unknown action should report error."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "nonexistent_action",
+ "modify_value": "",
+ },
+ )
+ assert response.status_code == 200
+ assert "0 modified, 1 failed" in response.text
+ assert "Unknown action" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_change_webhook_updates_tag(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with change_webhook should call set_tag with the new webhook URL."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "change_webhook",
+ "modify_value": MASS_WEBHOOK_NAME,
+ },
+ )
+ assert response.status_code == 200
+ assert "Modified 1 feed(s)" in response.text
+ # Check that set_tag was called with the webhook tag
+ webhook_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "webhook"]
+ assert len(webhook_calls) == 1
+ assert webhook_calls[0][2] == MASS_WEBHOOK_URL
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_unknown_webhook_reports_error(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with a non-existent webhook name should report error."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "change_webhook",
+ "modify_value": "Nonexistent Hook",
+ },
+ )
+ assert response.status_code == 200
+ assert "0 modified, 1 failed" in response.text
+ assert "not found" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_delivery_mode_embed(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with delivery_mode=embed should set delivery_mode and should_send_embed."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "delivery_mode",
+ "modify_value": "embed",
+ },
+ )
+ assert response.status_code == 200
+ assert "Modified 1 feed(s)" in response.text
+ delivery_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "delivery_mode"]
+ embed_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "should_send_embed"]
+ assert len(delivery_calls) == 1
+ assert delivery_calls[0][2] == "embed"
+ assert len(embed_calls) == 1
+ assert embed_calls[0][2] is True
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_delivery_mode_text(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with delivery_mode=text should set should_send_embed to False."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "delivery_mode",
+ "modify_value": "text",
+ },
+ )
+ assert response.status_code == 200
+ assert "Modified 1 feed(s)" in response.text
+ embed_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "should_send_embed"]
+ assert len(embed_calls) == 1
+ assert embed_calls[0][2] is False
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_update_interval(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with update_interval should store the interval."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ with patch("discord_rss_bot.main.commit_state_change"):
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "update_interval",
+ "modify_value": "30",
+ },
+ )
+ assert response.status_code == 200
+ assert "Modified 1 feed(s)" in response.text
+ update_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == ".reader.update"]
+ assert len(update_calls) == 1
+ assert update_calls[0][2] == {"interval": 30}
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_update_interval_invalid(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with non-numeric update_interval should report error."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.post(
+ "/mass/modify",
+ data={
+ "feed_urls": [MASS_FEED_1_URL],
+ "modify_action": "update_interval",
+ "modify_value": "not-a-number",
+ },
+ )
+ assert response.status_code == 200
+ assert "0 modified, 1 failed" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_modify_no_selection(self, mass_stub_reader: _StubReaderForMass) -> None:
+ """POST /mass/modify with no feed URLs should not crash."""
+ app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
+ try:
+ response: Response = client.post(
+ "/mass/modify",
+ data={"modify_action": "pause", "modify_value": ""},
+ )
+ assert response.status_code == 200
+ assert "Modified 0 feed(s)" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+
+class TestMassCreate:
+ """Tests for POST /mass/create."""
+
+ def test_create_rejects_missing_webhook(self) -> None:
+ """POST /mass/create without a valid webhook should return 404."""
+ stub = _StubReaderForMass()
+ app.dependency_overrides[get_reader_dependency] = lambda: stub
+ try:
+ response: Response = client.post(
+ "/mass/create",
+ data={
+ "feed_urls": "https://example.com/feed.xml",
+ "webhook_dropdown": "nonexistent",
+ },
+ )
+ assert response.status_code == 404
+ assert "Webhook not found" in response.text
+ finally:
+ app.dependency_overrides = {}
+
+ def test_create_single_url(self) -> None:
+ """POST /mass/create with a single URL should attempt to create it."""
+ stub = _StubReaderForMass(webhook_name=MASS_WEBHOOK_NAME, webhook_url=MASS_WEBHOOK_URL)
+ app.dependency_overrides[get_reader_dependency] = lambda: stub
+ try:
+ with (
+ patch("discord_rss_bot.main._update_and_mark_read") as mock_update,
+ patch("discord_rss_bot.main.commit_state_change"),
+ ):
+ mock_update.return_value = ("https://example.com/new-feed.xml", True, "")
+ response: Response = client.post(
+ "/mass/create",
+ data={
+ "feed_urls": "https://example.com/new-feed.xml",
+ "webhook_dropdown": MASS_WEBHOOK_NAME,
+ },
+ )
+ assert response.status_code == 200
+ finally:
+ app.dependency_overrides = {}
+
+ def test_create_multiple_urls(self) -> None:
+ """POST /mass/create with multiple URLs should attempt to create them."""
+ stub = _StubReaderForMass(webhook_name=MASS_WEBHOOK_NAME, webhook_url=MASS_WEBHOOK_URL)
+ app.dependency_overrides[get_reader_dependency] = lambda: stub
+ try:
+ with (
+ patch("discord_rss_bot.main._update_and_mark_read") as mock_update,
+ patch("discord_rss_bot.main.commit_state_change"),
+ ):
+ mock_update.side_effect = [
+ ("https://example.com/feed1.xml", True, ""),
+ ("https://example.com/feed2.xml", True, ""),
+ ]
+ response: Response = client.post(
+ "/mass/create",
+ data={
+ "feed_urls": "https://example.com/feed1.xml\nhttps://example.com/feed2.xml",
+ "webhook_dropdown": MASS_WEBHOOK_NAME,
+ },
+ )
+ assert response.status_code == 200
+ finally:
+ app.dependency_overrides = {}
+
+
+class TestMassNav:
+ """Tests for the mass operations navigation link."""
+
+ def test_mass_link_in_navbar(self) -> None:
+ """The navbar should contain a link to the mass operations page."""
+ response: Response = client.get("/")
+ assert response.status_code == 200
+ assert "/mass" in response.text, f"Expected /mass link in navbar: {response.text}"