diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py index 6693c82..70b8de6 100644 --- a/discord_rss_bot/main.py +++ b/discord_rss_bot/main.py @@ -1,12 +1,9 @@ from __future__ import annotations -import concurrent.futures -import io import json import logging import logging.config import re -import tempfile import typing import urllib.parse from contextlib import asynccontextmanager @@ -16,7 +13,6 @@ 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 @@ -28,11 +24,9 @@ import uvicorn from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import Depends from fastapi import FastAPI -from fastapi import File from fastapi import Form from fastapi import HTTPException from fastapi import Request -from fastapi import UploadFile from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates @@ -43,14 +37,11 @@ from reader import Entry from reader import EntryNotFoundError from reader import Feed from reader import FeedExistsError -from reader import FeedImportError from reader import FeedNotFoundError from reader import Reader from reader import ReaderError from reader import TagNotFoundError -from reader import opml from starlette.responses import RedirectResponse -from starlette.responses import Response as StarletteResponse from discord_rss_bot.custom_filters import entry_is_blacklisted from discord_rss_bot.custom_filters import entry_is_whitelisted @@ -63,7 +54,6 @@ 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 @@ -94,15 +84,12 @@ 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 @@ -287,196 +274,6 @@ templates.env.globals["get_backup_path"] = get_backup_path # pyright: ignore[re templates.env.globals["has_webhooks"] = has_webhooks # pyright: ignore[reportArgumentType] -@app.get("/export_opml") -def export_opml( - reader: Annotated[Reader, Depends(get_reader_dependency)], -) -> StarletteResponse: - """Export all feeds as an OPML subscription list. - - Args: - reader: The Reader instance. - - Returns: - StarletteResponse: The OPML file for download. - """ - export = reader.export_feeds() - return StarletteResponse( - content=export.content, - status_code=200, - headers={ - "Content-Type": "application/xml", - "Content-Disposition": f'attachment; filename="{export.filename}"', - }, - ) - - -@app.post("/import_opml", response_model=None) -async def import_opml( - request: Request, - file: Annotated[UploadFile, File()], - reader: Annotated[Reader, Depends(get_reader_dependency)], -): - """Upload an OPML file and show a preview of feeds to import. - - Args: - request: The request object. - file: The uploaded OPML file. - reader: The Reader instance. - - Returns: - HTMLResponse: The OPML import preview page. - RedirectResponse: Redirect to settings on error. - """ - if not file.filename or not file.filename.lower().endswith(".opml"): - return RedirectResponse( - url=f"/settings?message={urllib.parse.quote('Please upload a file with a .opml extension.')}", - status_code=303, - ) - - try: - content: bytes = await file.read() - feeds_to_import = opml.parse(io.BytesIO(content)) - except FeedImportError as e: - return RedirectResponse( - url=f"/settings?message={urllib.parse.quote(f'Failed to parse OPML file: {e}')}", - status_code=303, - ) - - # Check which feeds already exist - existing_urls: set[str] = {feed.url for feed in reader.get_feeds()} - feed_list = [ - { - "url": feed.url, - "title": feed.title or feed.url, - "already_exists": feed.url in existing_urls, - } - for feed in feeds_to_import - ] - - context = { - "request": request, - "feeds": feed_list, - "total": len(feed_list), - "new_count": sum(1 for f in feed_list if not f["already_exists"]), - "existing_count": sum(1 for f in feed_list if f["already_exists"]), - "webhooks": reader.get_tag((), "webhooks", []), - } - return templates.TemplateResponse(request=request, name="import_opml_preview.html", context=context) - - -@app.post("/import_opml_confirm") -async def import_opml_confirm( - request: Request, - reader: Annotated[Reader, Depends(get_reader_dependency)], - feed_urls: Annotated[list[str] | None, Form()] = None, - webhook_name: Annotated[str | None, Form()] = None, -) -> RedirectResponse: - """Import the selected feeds from the OPML preview. - - Args: - request: The request object. - reader: The Reader instance. - feed_urls: The selected feed URLs to import. - webhook_name: Optional webhook name to attach imported feeds to. - - Returns: - RedirectResponse: Redirect to the settings page with a status message. - """ - if feed_urls is None: - feed_urls = [] - if not feed_urls: - return RedirectResponse( - url="/settings?message=No%20feeds%20were%20selected%20for%20import.", - status_code=303, - ) - - webhook_url = _resolve_webhook_url(reader, webhook_name) - - imported, updated_webhook, errors = _import_opml_feeds(reader, feed_urls, webhook_url) - message = _summarize_opml_import(imported, updated_webhook, errors) - - logger.info("OPML import complete: %s", message) - commit_state_change(reader, f"OPML import: {imported} feeds") - - return RedirectResponse(url=f"/settings?message={urllib.parse.quote(message)}", status_code=303) - - -def _resolve_webhook_url(reader: Reader, webhook_name: str | None) -> str: - """Resolve a webhook name to its URL from reader storage. - - Args: - reader: The Reader instance. - webhook_name: The webhook name to look up. - - Returns: - The webhook URL, or empty string if not found or no name given. - """ - if not webhook_name: - return "" - hooks = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", []))) - for hook in hooks: - if hook.get("name") == webhook_name.strip(): - return hook.get("url", "").strip() - return "" - - -def _import_opml_feeds( - reader: Reader, - feed_urls: list[str], - webhook_url: str, -) -> tuple[int, int, list[str]]: - """Add feeds from an OPML import, optionally setting webhooks. - - Args: - reader: The Reader instance. - feed_urls: The feed URLs to add. - webhook_url: Webhook URL to attach, or empty string. - - Returns: - A tuple of (imported_count, updated_webhook_count, error_messages). - """ - imported: int = 0 - updated_webhook: int = 0 - errors: list[str] = [] - - for feed_url in feed_urls: - try: - reader.add_feed(feed_url) - if webhook_url: - reader.set_tag(feed_url, "webhook", webhook_url) # pyright: ignore[reportArgumentType] - imported += 1 - except FeedExistsError: - if webhook_url: - reader.set_tag(feed_url, "webhook", webhook_url) # pyright: ignore[reportArgumentType] - updated_webhook += 1 - except Exception as e: - errors.append(f"{feed_url}: {e}") - logger.exception("Failed to import feed: %s", feed_url) - - return imported, updated_webhook, errors - - -def _summarize_opml_import(imported: int, updated_webhook: int, errors: list[str]) -> str: - """Build a human-readable summary of an OPML import result. - - Args: - imported: Number of newly imported feeds. - updated_webhook: Number of existing feeds whose webhook was updated. - errors: List of error strings. - - Returns: - A summary string. - """ - parts: list[str] = [] - if imported: - parts.append(f"Successfully imported {imported} feed{'s' if imported != 1 else ''}") - if updated_webhook: - parts.append(f"Updated webhook for {updated_webhook} existing feed{'s' if updated_webhook != 1 else ''}") - if errors: - parts.append(f"{len(errors)} error{'s' if len(errors) != 1 else ''}") - return ". ".join(parts) + "." - - def get_global_delivery_mode(reader: Reader) -> str: """Return the normalized default delivery mode for new feeds. @@ -2332,14 +2129,12 @@ def get_data_from_hook_url(hook_name: str, hook_url: str) -> WebhookInfo: async def get_settings( request: Request, reader: Annotated[Reader, Depends(get_reader_dependency)], - message: str = "", ): """Settings page. Args: request: The request object. reader: The Reader instance. - message: Optional message to display to the user. Returns: HTMLResponse: The settings page. @@ -2391,7 +2186,6 @@ async def get_settings( "max_webhook_text_length_limit": 4000, "feed_intervals": feed_intervals, "chromium_installed": is_chromium_installed(), - "messages": message or None, } return templates.TemplateResponse(request=request, name="settings.html", context=context) @@ -2597,33 +2391,6 @@ async def update_feed( return RedirectResponse(url="/feed?feed_url=" + urllib.parse.quote(feed_url), status_code=303) -@app.get("/export") -def export_database( - reader: Annotated[Reader, Depends(get_reader_dependency)], -) -> StarletteResponse: - """Export the entire database as a compressed SQLite file. - - Args: - reader: The Reader instance. - - Returns: - StarletteResponse: The exported database file for download. - """ - with tempfile.TemporaryDirectory() as tmpdir: - export_path: Path = reader._storage.export(tmpdir, "discord-rss-bot-export") # ruff:ignore[private-member-access] - filename: str = export_path.name - file_bytes: bytes = export_path.read_bytes() - - return StarletteResponse( - content=file_bytes, - status_code=200, - headers={ - "Content-Type": "application/gzip", - "Content-Disposition": f'attachment; filename="{filename}"', - }, - ) - - @app.post("/backup") async def manual_backup( request: Request, @@ -2655,439 +2422,12 @@ 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/import_opml_preview.html b/discord_rss_bot/templates/import_opml_preview.html deleted file mode 100644 index ef901db..0000000 --- a/discord_rss_bot/templates/import_opml_preview.html +++ /dev/null @@ -1,117 +0,0 @@ -{% extends "base.html" %} -{% block title %} - Import OPML Feeds | discord-rss-bot -{% endblock title %} -{% block description %} - Select which feeds to import from your OPML subscription list. -{% endblock description %} -{% block content %} -
-
-
-

Import OPML Feeds

-
-

- Found {{ total }} feed{{ 's' if total != 1 else '' }} in the OPML file. - {% if new_count %} - {{ new_count }} new - {% endif %} - {% if existing_count %} - {{ existing_count }} already exist{{ 's' if existing_count == 1 else '' }} - {% endif %} -

-
- {% if feeds %} -
- - {% if webhooks %} -
- - -
- New feeds will be added with this webhook. Existing feeds will have their webhook overwritten. You can change this later per-feed. -
-
- {% else %} - - {% endif %} -
- - - - - - - - - - - {% for feed in feeds %} - - - - - - - {% endfor %} - -
- - FeedURLStatus
- - - {{ feed.title }} - - {{ feed.url }} - - {% if feed.already_exists %} - Already exists - {% else %} - New - {% endif %} -
-
-
- - Cancel -
-
- {% else %} -

No feeds found in the OPML file.

- Back to Settings - {% endif %} -
-{% endblock content %} -{% block head %} - -{% endblock head %} diff --git a/discord_rss_bot/templates/mass.html b/discord_rss_bot/templates/mass.html deleted file mode 100644 index 88a6af8..0000000 --- a/discord_rss_bot/templates/mass.html +++ /dev/null @@ -1,355 +0,0 @@ -{% 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 d0a5cdf..1fb1728 100644 --- a/discord_rss_bot/templates/nav.html +++ b/discord_rss_bot/templates/nav.html @@ -18,10 +18,6 @@ {% endif %} - - diff --git a/discord_rss_bot/templates/settings.html b/discord_rss_bot/templates/settings.html index 32437ea..df4b5ee 100644 --- a/discord_rss_bot/templates/settings.html +++ b/discord_rss_bot/templates/settings.html @@ -6,262 +6,191 @@ Adjust default update intervals, delivery modes, screenshot layout, and webhook text limits for feeds managed by your bot. {% endblock description %} {% block content %} -
- -
-
-
-
-
-

Global Defaults

-
+
+
+
+

Global Settings

+
+

+ Set a default interval for all feeds. Individual feeds can still override this value. +

+
+
+ Current default is {{ global_interval }} min. + Even though we check ETags and Last-Modified headers, choosing a very low interval may cause issues with some feeds or cause excessive load on the server hosting the feed. Remember to be kind. +
+
+
+
+
+ +
+ + +
+
+
+
+
+ +
+ + +
+
+ New feeds inherit this value. Existing feeds keep their current delivery mode. +
+
+
+
+
+ +
+ + +
+
+ New feeds inherit this value. Existing feeds keep their current screenshot layout. +
+ {% if not chromium_installed %} +
+ Screenshot mode requires Chromium to be installed for Playwright. + Run uv run playwright install chromium once on this machine.
-

- These defaults apply to newly added feeds. Existing feeds keep their own per-feed settings. -

-
- -
- - -
- - -
-
- Currently {{ global_interval }} min. Low intervals may cause issues with some feeds. -
- -
- -
-
- -
- - -
-
How new feeds send messages to Discord.
-
-
- -
-
- -
- - -
-
- Default screenshot viewport for new feeds. - {% if not chromium_installed %} - - Requires Chromium. Run uv run playwright install chromium. - + {% endif %} +
+
+
+
+ +
+ + +
+
+ New feeds inherit this value. Existing feeds keep their current + per-feed text limit. Text mode allows values up to 4000 characters. Embeds are capped at 2000 characters. +
+
+
+
+
+
+
+

Feed Update Intervals

+
+

+ Customize the update interval for individual feeds. Leave empty or reset to use the global default. +

+
+ {% if feed_intervals %} +
+ + + + + + + + + + + + + + + {% for item in feed_intervals %} + + + + + + + + + + + {% endfor %} + +
FeedDomainStatusIntervalLast UpdatedNext UpdateSet Interval (min)Actions
+ {{ item.feed.title }} + + {{ item.domain }} + + + {{ 'Enabled' if item.feed.updates_enabled else 'Disabled' }} + + + {{ item.effective_interval }} min + {% if item.interval %} + Custom + {% else %} + Global {% endif %} - - - - -
-
- -
- - -
-
Max characters for text mode (4000). Embeds are capped at 2000.
-
-
- - - - - -
-
-
-
-
-

Data Management

-
-
-

Import and export your feeds and database.

-
- -
-
-
OPML Export
-
-

- Download all feeds as an OPML subscription list. - Supports title, links, and description. -

- Export OPML -
-
-
- -
-
-
OPML Import
-
-

Upload an OPML file to preview and select which feeds to import.

-
-
- - -
+
+ {{ item.feed.last_updated | relative_time }} + + {{ item.feed.update_after | relative_time }} + + + + + + - - - - -
-
-
- Database Export -
-
-

Download a compressed SQLite dump for backup or migration.

- Download Export -
-
-
- - - - - -
-
-
-
-
-

Feed Update Intervals

-
-
-

- Customize the update interval for individual feeds. Leave empty or reset to use the global default of {{ global_interval }} min. -

- {% if feed_intervals %} -
- - - - - - - - - - - - - - - {% for item in feed_intervals %} - - - - - - - - - - - {% endfor %} - -
FeedDomainStatusIntervalLast Updated - Next - Update - Set Interval (min)Actions
- {{ item.feed.title }} - - {{ item.domain }} - - - {{ 'Enabled' if item.feed.updates_enabled else 'Disabled' }} - - - {{ item.effective_interval }} min - {% if item.interval %} - Custom - {% else %} - Global - {% endif %} - - {{ item.feed.last_updated | relative_time }} - - {{ item.feed.update_after | relative_time }} - -
- - - - -
-
- {% if item.interval %} -
- - - -
- {% endif %} -
-
- {% else %} -

No feeds added yet.

- {% endif %} -
-
-
- +
+ {% if item.interval %} +
+ + + +
+ {% endif %} +
+
+ {% else %} +

No feeds added yet.

+ {% endif %} +
{% endblock content %} diff --git a/tests/test_main.py b/tests/test_main.py index ee1b12a..9c28d64 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -14,10 +14,7 @@ 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 @@ -28,6 +25,7 @@ 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 @@ -648,38 +646,44 @@ def test_author_templates_render_authors_str() -> None: assert "Legacy Entry Author" not in filter_preview_html -def test_settings_page_loads() -> None: - """The settings page should render without errors.""" +def test_settings_page_shows_screenshot_layout_setting() -> None: response: Response = client.get(url="/settings") assert response.status_code == 200, f"/settings failed: {response.text}" + assert "Default delivery mode for new feeds" in response.text + assert "Default screenshot layout for new feeds" in response.text + assert "2000 characters" in response.text + assert 'id="global_text_length_limit"' in response.text + assert "uv run playwright install chromium" in response.text -def test_set_global_delivery_mode_stores_value() -> None: - """POST /set_global_delivery_mode should persist the delivery mode in reader tags.""" +def test_set_global_delivery_mode() -> None: response: Response = client.post(url="/set_global_delivery_mode", data={"delivery_mode": "text"}) assert response.status_code == 200, f"Failed to set global delivery mode: {response.text}" - reader: Reader = get_reader_dependency() - assert reader.get_tag((), "delivery_mode", "") == "text" + + response = client.get(url="/settings") + assert response.status_code == 200, f"/settings failed after setting delivery mode: {response.text}" + assert re.search(r"]*\bselected\b", response.text) -def test_set_global_webhook_text_length_limit_stores_value() -> None: - """POST /set_global_webhook_text_length_limit should persist the limit in reader tags.""" +def test_set_global_webhook_text_length_limit() -> None: response: Response = client.post( url="/set_global_webhook_text_length_limit", data={"text_length_limit": "2500"}, ) assert response.status_code == 200, f"Failed to set global webhook text length limit: {response.text}" - reader: Reader = get_reader_dependency() - assert reader.get_tag((), "webhook_text_length_limit", 0) == 2500 + + response = client.get(url="/settings") + assert response.status_code == 200, f"/settings failed after setting webhook text length limit: {response.text}" + assert 'value="2500"' in response.text -def test_set_global_delivery_mode_affects_add_page() -> None: - """Setting the delivery mode should be reflected in the response from /add.""" - reader: Reader = get_reader_dependency() - reader.set_tag((), "delivery_mode", "text") # pyright: ignore[reportArgumentType] +def test_add_page_shows_global_default_delivery_mode_hint() -> None: + response: Response = client.post(url="/set_global_delivery_mode", data={"delivery_mode": "text"}) + assert response.status_code == 200, f"Failed to set global delivery mode: {response.text}" - response: Response = client.get(url="/add") + response = client.get(url="/add") assert response.status_code == 200, f"/add failed: {response.text}" + assert "text" in response.text def test_navbar_add_feed_visible_only_when_webhooks_exist() -> None: @@ -895,12 +899,13 @@ def test_sent_webhooks_view_shows_saved_records() -> None: app.dependency_overrides = {} -def test_set_global_screenshot_layout_stores_value() -> None: - """POST /set_global_screenshot_layout should persist the layout in reader tags.""" +def test_set_global_screenshot_layout() -> None: response: Response = client.post(url="/set_global_screenshot_layout", data={"screenshot_layout": "mobile"}) assert response.status_code == 200, f"Failed to set global screenshot layout: {response.text}" - reader: Reader = get_reader_dependency() - assert reader.get_tag((), "screenshot_layout", "") == "mobile" + + response = client.get(url="/settings") + assert response.status_code == 200, f"/settings failed after setting layout: {response.text}" + assert re.search(r"]*\bselected\b", response.text) def test_pause_feed() -> None: @@ -2549,268 +2554,10 @@ def test_reader_dependency_override_is_used() -> None: # --------------------------------------------------------------------------- -# Tests for OPML import / export +# Tests for post_embed — saving embed fields (including clearing to "") # --------------------------------------------------------------------------- -def test_export_opml_returns_opml_content() -> None: - """GET /export_opml should return an OPML XML file.""" - reader: Reader = get_reader_dependency() - with contextlib.suppress(Exception): - reader.add_feed(feed_url) - - response: Response = client.get("/export_opml") - assert response.status_code == 200, f"/export_opml failed: {response.text}" - assert response.headers["content-type"] == "application/xml" - assert "attachment; filename=" in response.headers["content-disposition"] - assert response.headers["content-disposition"].startswith("attachment") - assert ".opml" in response.headers["content-disposition"] - assert b"" in response.content - assert feed_url.encode() in response.content or b"lovinator.space" in response.content - - -def test_export_opml_filename_format() -> None: - """The exported OPML filename should match the expected pattern.""" - response: Response = client.get("/export_opml") - assert response.status_code == 200 - disposition: str = response.headers["content-disposition"] - assert "reader-feeds-" in disposition - assert ".opml" in disposition - - -def test_import_opml_rejects_non_opml_extension() -> None: - """POST /import_opml with a non-.opml file should redirect with an error.""" - response: Response = client.post( - url="/import_opml", - files={"file": ("feeds.txt", b"not opml", "text/plain")}, - follow_redirects=False, - ) - assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}" - assert ".opml" in response.headers.get("location", "").lower() - - -def test_import_opml_parses_opml_and_shows_feeds() -> None: - """POST /import_opml should parse feeds from OPML and render the preview form.""" - opml_content: bytes = b""" - - test - - - - -""" - - response: Response = client.post( - url="/import_opml", - files={"file": ("feeds.opml", opml_content, "application/xml")}, - follow_redirects=False, - ) - - assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}" - assert "https://example.com/feed.xml" in response.text - assert 'action="/import_opml_confirm"' in response.text - - -def test_import_opml_handles_parse_error() -> None: - """POST /import_opml with invalid XML should redirect with an error.""" - response: Response = client.post( - url="/import_opml", - files={"file": ("bad.opml", b"not valid opml", "application/xml")}, - follow_redirects=False, - ) - - assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}" - location = response.headers.get("location", "") - assert "Failed to parse OPML file" in urllib.parse.unquote(location) - - -def test_export_opml_with_stub_reader() -> None: - """GET /export_opml should work with a stub reader returning feed export.""" - - @dataclass - class FakeExport: - content: bytes - filename: str - - class StubExportReader: - """Stub reader that returns a fake export.""" - - def export_feeds(self) -> FakeExport: - return FakeExport( - content=( - b'' - b'reader feeds' - ), - filename="reader-feeds-2026-07-18-12-00-00.opml", - ) - - def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object: - return default - - def get_feeds(self) -> list: - return [] - - stub = StubExportReader() - app.dependency_overrides[get_reader_dependency] = lambda: stub - - try: - response: Response = client.get("/export_opml") - assert response.status_code == 200, f"/export_opml with stub failed: {response.text}" - assert response.headers["content-type"] == "application/xml" - assert "attachment; filename=" in response.headers["content-disposition"] - assert ".opml" in response.headers["content-disposition"] - finally: - app.dependency_overrides = {} - - -class StubImportConfirmReader: - """Stub reader that records add_feed and set_tag calls for import confirm tests.""" - - def __init__(self) -> None: - """Initialize the stub with empty tracking lists.""" - self.added_urls: list[str] = [] - self.tags: dict[tuple[str, str], str] = {} - - def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object: - """Stub get_tag. - - Returns: - The default value. - """ - return default - - def get_feeds(self) -> list: - """Stub get_feeds. - - Returns: - An empty list. - """ - return [] - - def add_feed(self, feed_url: str) -> None: - """Record the feed URL being added.""" - self.added_urls.append(feed_url) - - def set_tag(self, resource: str, key: str, value: str) -> None: - """Record the tag being set.""" - self.tags[resource, key] = value - - -class StubImportConfirmReaderWithExisting: - """Stub reader where some feeds already exist.""" - - def __init__(self) -> None: - """Initialize the stub with empty tracking lists.""" - self.added_urls: list[str] = [] - self.tags: dict[tuple[str, str], str] = {} - - def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object: - """Stub get_tag. - - Returns: - The default value. - """ - if _key == "webhooks": - return [{"name": "Test Webhook", "url": "https://discord.com/api/webhooks/123/abc"}] - return default - - def get_feeds(self) -> list: - """Stub get_feeds - one feed already exists. - - Returns: - A list with one existing feed. - """ - return [type("FakeFeed", (), {"url": "https://example.com/existing.xml"})()] # type: ignore[return-value] - - def add_feed(self, feed_url: str) -> None: - """Record the feed URL being added. - - Raises: - FeedExistsError: If the feed URL already exists. - """ - if feed_url == "https://example.com/existing.xml": - raise FeedExistsError(feed_url) - self.added_urls.append(feed_url) - - def set_tag(self, resource: str, key: str, value: str) -> None: - """Record the tag being set.""" - self.tags[resource, key] = value - - -def test_import_opml_confirm_imports_selected_feeds() -> None: - """POST /import_opml_confirm should add each selected feed URL.""" - stub = StubImportConfirmReader() - app.dependency_overrides[get_reader_dependency] = lambda: stub - - try: - with patch("discord_rss_bot.main.commit_state_change"): - response: Response = client.post( - url="/import_opml_confirm", - data={"feed_urls": ["https://example.com/feed1.xml", "https://example.com/feed2.xml"]}, - follow_redirects=False, - ) - - assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}" - assert stub.added_urls == ["https://example.com/feed1.xml", "https://example.com/feed2.xml"] - location = response.headers.get("location", "") - assert "Successfully imported 2 feeds" in urllib.parse.unquote(location) - finally: - app.dependency_overrides = {} - - -def test_import_opml_confirm_no_selection() -> None: - """POST /import_opml_confirm with no selection should redirect with a message.""" - stub = StubImportConfirmReader() - app.dependency_overrides[get_reader_dependency] = lambda: stub - - try: - response: Response = client.post( - url="/import_opml_confirm", - data={}, - follow_redirects=False, - ) - - assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}" - location = response.headers.get("location", "") - assert "No feeds were selected" in urllib.parse.unquote(location) - assert stub.added_urls == [] - finally: - app.dependency_overrides = {} - - -def test_import_opml_confirm_updates_webhook_on_existing_feeds() -> None: - """POST /import_opml_confirm should update webhook on feeds that already exist.""" - stub = StubImportConfirmReaderWithExisting() - app.dependency_overrides[get_reader_dependency] = lambda: stub - - try: - with patch("discord_rss_bot.main.commit_state_change"): - response: Response = client.post( - url="/import_opml_confirm", - data={ - "feed_urls": ["https://example.com/existing.xml", "https://example.com/new.xml"], - "webhook_name": "Test Webhook", - }, - follow_redirects=False, - ) - - assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}" - assert stub.added_urls == ["https://example.com/new.xml"] - # Existing feed should have its webhook updated - assert ( - stub.tags.get(("https://example.com/existing.xml", "webhook")) == "https://discord.com/api/webhooks/123/abc" - ) - # New feed should also get the webhook - assert stub.tags.get(("https://example.com/new.xml", "webhook")) == "https://discord.com/api/webhooks/123/abc" - location = response.headers.get("location", "") - decoded = urllib.parse.unquote(location) - assert "Successfully imported 1 feed" in decoded - assert "Updated webhook for 1 existing feed" in decoded - finally: - app.dependency_overrides = {} - - def _make_stub_reader_for_embed( *, stored_embed: str | None = None, @@ -3356,532 +3103,3 @@ 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}"