From 64a116d9478643141c6738c55361f8964bc83d52 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Sat, 18 Jul 2026 19:29:42 +0200 Subject: [PATCH] Add mass editor --- discord_rss_bot/main.py | 437 ++++++++++++++++++++++- discord_rss_bot/templates/mass.html | 355 +++++++++++++++++++ discord_rss_bot/templates/nav.html | 4 + tests/test_main.py | 532 +++++++++++++++++++++++++++- 4 files changed, 1325 insertions(+), 3 deletions(-) create mode 100644 discord_rss_bot/templates/mass.html 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 %} +
+ {% for group in all_feeds_grouped %} +
+
+

{{ group.name }} ({{ group.feeds|length }})

+
+ + +
+
+
    + {% for item in group.feeds %} + {% set feed = item.feed %} +
  • + + +
    • + {% endfor %} +
+
+ {% endfor %} + +
+ {% 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 %} +
+ {% for group in all_feeds_grouped %} +
+
+

{{ group.name }} ({{ group.feeds|length }})

+
+ + +
+
+
    + {% for item in group.feeds %} + {% set feed = item.feed %} +
  • + + +
    • + {% endfor %} +
+
+ {% endfor %} +
+
+
+ + +
+
+ +
+ +
+
+
+ +
+ {% 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}"