From 24d4d7a293d300632da93b3e73a56150607e640a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Helle=C5=9Ben?= Date: Sat, 7 Mar 2026 05:50:13 +0100 Subject: [PATCH] Add support for changing the update interval for feeds --- README.md | 1 + discord_rss_bot/feeds.py | 8 +- discord_rss_bot/git_backup.py | 16 +- discord_rss_bot/hoyolab_api.py | 6 +- discord_rss_bot/main.py | 243 +++++++++++++++++++++++- discord_rss_bot/settings.py | 10 +- discord_rss_bot/static/styles.css | 4 + discord_rss_bot/templates/feed.html | 219 +++++++++++++-------- discord_rss_bot/templates/nav.html | 4 + discord_rss_bot/templates/settings.html | 122 ++++++++++++ pyproject.toml | 8 - tests/test_blacklist.py | 8 +- tests/test_custom_filter.py | 4 +- tests/test_feeds.py | 21 +- tests/test_main.py | 150 +++++++++++++++ tests/test_search.py | 4 +- tests/test_settings.py | 4 +- tests/test_update_interval.py | 88 +++++++++ 18 files changed, 802 insertions(+), 118 deletions(-) create mode 100644 discord_rss_bot/templates/settings.html create mode 100644 tests/test_update_interval.py diff --git a/README.md b/README.md index 26a8ad8..09b6bbc 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ Discord: TheLovinator#9276 - Choose between Discord embed or plain text. - Regex filters for RSS feeds. - Blacklist/whitelist words in the title/description/author/etc. +- Set different update frequencies for each feed or use a global default. - Gets extra information from APIs if available, currently for: - [https://feeds.c3kay.de/](https://feeds.c3kay.de/) - Genshin Impact News diff --git a/discord_rss_bot/feeds.py b/discord_rss_bot/feeds.py index c01cecb..8e6ca63 100644 --- a/discord_rss_bot/feeds.py +++ b/discord_rss_bot/feeds.py @@ -98,7 +98,7 @@ def extract_domain(url: str) -> str: # noqa: PLR0911 return "Other" -def send_entry_to_discord(entry: Entry, custom_reader: Reader | None = None) -> str | None: # noqa: PLR0912 +def send_entry_to_discord(entry: Entry, custom_reader: Reader | None = None) -> str | None: # noqa: C901, PLR0912 """Send a single entry to Discord. Args: @@ -240,7 +240,7 @@ def set_title(custom_embed: CustomEmbed, discord_embed: DiscordEmbed) -> None: discord_embed.set_title(embed_title) if embed_title else None -def create_embed_webhook(webhook_url: str, entry: Entry) -> DiscordWebhook: +def create_embed_webhook(webhook_url: str, entry: Entry) -> DiscordWebhook: # noqa: C901 """Create a webhook with an embed. Args: @@ -341,7 +341,7 @@ def set_entry_as_read(reader: Reader, entry: Entry) -> None: logger.exception("Error setting entry to read: %s", entry.id) -def send_to_discord(custom_reader: Reader | None = None, feed: Feed | None = None, *, do_once: bool = False) -> None: # noqa: PLR0912 +def send_to_discord(custom_reader: Reader | None = None, feed: Feed | None = None, *, do_once: bool = False) -> None: # noqa: C901, PLR0912 """Send entries to Discord. If response was not ok, we will log the error and mark the entry as unread, so it will be sent again next time. @@ -520,7 +520,7 @@ def truncate_webhook_message(webhook_message: str) -> str: return webhook_message -def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: +def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: # noqa: C901 """Add a new feed, update it and mark every entry as read. Args: diff --git a/discord_rss_bot/git_backup.py b/discord_rss_bot/git_backup.py index 0277dca..4106d21 100644 --- a/discord_rss_bot/git_backup.py +++ b/discord_rss_bot/git_backup.py @@ -30,6 +30,8 @@ from pathlib import Path from typing import TYPE_CHECKING from typing import Any +from reader import TagNotFoundError + if TYPE_CHECKING: from reader import Reader @@ -66,6 +68,7 @@ _FEED_TAGS: tuple[str, ...] = ( "regex_whitelist_summary", "regex_whitelist_content", "regex_whitelist_author", + ".reader.update", ) @@ -178,10 +181,21 @@ def export_state(reader: Reader, backup_path: Path) -> None: webhooks: list[str | int | float | bool | dict[str, Any] | list[Any] | None] = list( reader.get_tag((), "webhooks", []) ) - except Exception: # noqa: BLE001 + except TagNotFoundError: webhooks = [] + # Export global update interval if set + global_update_interval: dict[str, Any] | None = None + try: + global_update_config = reader.get_tag((), ".reader.update", None) + if isinstance(global_update_config, dict): + global_update_interval = global_update_config + except TagNotFoundError: + pass + state: dict = {"feeds": feeds_state, "webhooks": webhooks} + if global_update_interval is not None: + state["global_update_interval"] = global_update_interval state_file: Path = backup_path / "state.json" state_file.write_text(json.dumps(state, indent=2, default=str), encoding="utf-8") diff --git a/discord_rss_bot/hoyolab_api.py b/discord_rss_bot/hoyolab_api.py index cb1ed71..227a413 100644 --- a/discord_rss_bot/hoyolab_api.py +++ b/discord_rss_bot/hoyolab_api.py @@ -4,10 +4,12 @@ import contextlib import json import logging import re -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING +from typing import Any import requests -from discord_webhook import DiscordEmbed, DiscordWebhook +from discord_webhook import DiscordEmbed +from discord_webhook import DiscordWebhook if TYPE_CHECKING: from reader import Entry diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py index 0e884bf..2e7af0f 100644 --- a/discord_rss_bot/main.py +++ b/discord_rss_bot/main.py @@ -100,6 +100,46 @@ logging.config.dictConfig(LOGGING_CONFIG) logger: logging.Logger = logging.getLogger(__name__) reader: Reader = get_reader() +# Time constants for relative time formatting +SECONDS_PER_MINUTE = 60 +SECONDS_PER_HOUR = 3600 +SECONDS_PER_DAY = 86400 + + +def relative_time(dt: datetime | None) -> str: + """Convert a datetime to a relative time string (e.g., '2 hours ago', 'in 5 minutes'). + + Args: + dt: The datetime to convert (should be timezone-aware). + + Returns: + A human-readable relative time string. + """ + if dt is None: + return "Never" + + now = datetime.now(tz=UTC) + diff = dt - now + seconds = int(abs(diff.total_seconds())) + is_future = diff.total_seconds() > 0 + + # Determine the appropriate unit and value + if seconds < SECONDS_PER_MINUTE: + value = seconds + unit = "s" + elif seconds < SECONDS_PER_HOUR: + value = seconds // SECONDS_PER_MINUTE + unit = "m" + elif seconds < SECONDS_PER_DAY: + value = seconds // SECONDS_PER_HOUR + unit = "h" + else: + value = seconds // SECONDS_PER_DAY + unit = "d" + + # Format based on future or past + return f"in {value}{unit}" if is_future else f"{value}{unit} ago" + @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None]: @@ -131,6 +171,7 @@ templates.env.filters["encode_url"] = lambda url: urllib.parse.quote(url) if url templates.env.filters["entry_is_whitelisted"] = entry_is_whitelisted templates.env.filters["entry_is_blacklisted"] = entry_is_blacklisted templates.env.filters["discord_markdown"] = markdownify +templates.env.filters["relative_time"] = relative_time templates.env.globals["get_backup_path"] = get_backup_path @@ -613,6 +654,102 @@ async def post_use_text(feed_url: Annotated[str, Form()]) -> RedirectResponse: return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303) +@app.post("/set_update_interval") +async def post_set_update_interval( + feed_url: Annotated[str, Form()], + interval_minutes: Annotated[int | None, Form()] = None, + redirect_to: Annotated[str, Form()] = "", +) -> RedirectResponse: + """Set the update interval for a feed. + + Args: + feed_url: The feed to change. + interval_minutes: The update interval in minutes (None to reset to global default). + redirect_to: Optional redirect URL (defaults to feed page). + + Returns: + RedirectResponse: Redirect to the specified page or feed page. + """ + clean_feed_url: str = feed_url.strip() + + # If no interval specified, reset to global default + if interval_minutes is None: + try: + reader.delete_tag(clean_feed_url, ".reader.update") + commit_state_change(reader, f"Reset update interval to default for {clean_feed_url}") + except TagNotFoundError: + pass + else: + # Validate interval (minimum 1 minute, no maximum) + interval_minutes = max(interval_minutes, 1) + reader.set_tag(clean_feed_url, ".reader.update", {"interval": interval_minutes}) # pyright: ignore[reportArgumentType] + commit_state_change(reader, f"Set update interval to {interval_minutes} minutes for {clean_feed_url}") + + # Update the feed immediately to recalculate update_after with the new interval + try: + reader.update_feed(clean_feed_url) + logger.info("Updated feed after interval change: %s", clean_feed_url) + except Exception: + logger.exception("Failed to update feed after interval change: %s", clean_feed_url) + + if redirect_to: + return RedirectResponse(url=redirect_to, status_code=303) + return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303) + + +@app.post("/reset_update_interval") +async def post_reset_update_interval( + feed_url: Annotated[str, Form()], + redirect_to: Annotated[str, Form()] = "", +) -> RedirectResponse: + """Reset the update interval for a feed to use the global default. + + Args: + feed_url: The feed to change. + redirect_to: Optional redirect URL (defaults to feed page). + + Returns: + RedirectResponse: Redirect to the specified page or feed page. + """ + clean_feed_url: str = feed_url.strip() + + try: + reader.delete_tag(clean_feed_url, ".reader.update") + commit_state_change(reader, f"Reset update interval to default for {clean_feed_url}") + except TagNotFoundError: + # Tag doesn't exist, which is fine + pass + + # Update the feed immediately to recalculate update_after with the new interval + try: + reader.update_feed(clean_feed_url) + logger.info("Updated feed after interval reset: %s", clean_feed_url) + except Exception: + logger.exception("Failed to update feed after interval reset: %s", clean_feed_url) + + if redirect_to: + return RedirectResponse(url=redirect_to, status_code=303) + return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303) + + +@app.post("/set_global_update_interval") +async def post_set_global_update_interval(interval_minutes: Annotated[int, Form()]) -> RedirectResponse: + """Set the global default update interval. + + Args: + interval_minutes: The update interval in minutes. + + Returns: + RedirectResponse: Redirect to the settings page. + """ + # Validate interval (minimum 1 minute, no maximum) + interval_minutes = max(interval_minutes, 1) + + reader.set_tag((), ".reader.update", {"interval": interval_minutes}) # pyright: ignore[reportArgumentType] + commit_state_change(reader, f"Set global update interval to {interval_minutes} minutes") + return RedirectResponse(url="/settings", status_code=303) + + @app.get("/add", response_class=HTMLResponse) def get_add(request: Request): """Page for adding a new feed. @@ -631,7 +768,7 @@ def get_add(request: Request): @app.get("/feed", response_class=HTMLResponse) -async def get_feed(feed_url: str, request: Request, starting_after: str = ""): +async def get_feed(feed_url: str, request: Request, starting_after: str = ""): # noqa: C901, PLR0912, PLR0914, PLR0915 """Get a feed by URL. Args: @@ -656,7 +793,7 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""): # Only show button if more than 10 entries. total_entries: int = reader.get_entry_counts(feed=feed).total or 0 - show_more_entires_button: bool = total_entries > entries_per_page + is_show_more_entries_button_visible: bool = total_entries > entries_per_page # Get entries from the feed. if starting_after: @@ -669,6 +806,27 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""): msg: str = f"{e}\n\n{[entry.id for entry in current_entries]}" html: str = create_html_for_feed(current_entries) + # Get feed and global intervals for error case too + feed_interval: int | None = None + try: + feed_update_config = reader.get_tag(feed, ".reader.update") + if isinstance(feed_update_config, dict) and "interval" in feed_update_config: + interval_value = feed_update_config["interval"] + if isinstance(interval_value, int): + feed_interval = interval_value + except TagNotFoundError: + pass + + global_interval: int = 60 + try: + global_update_config = reader.get_tag((), ".reader.update") + if isinstance(global_update_config, dict) and "interval" in global_update_config: + interval_value = global_update_config["interval"] + if isinstance(interval_value, int): + global_interval = interval_value + except TagNotFoundError: + pass + context = { "request": request, "feed": feed, @@ -678,8 +836,10 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""): "should_send_embed": False, "last_entry": None, "messages": msg, - "show_more_entires_button": show_more_entires_button, + "is_show_more_entries_button_visible": is_show_more_entries_button_visible, "total_entries": total_entries, + "feed_interval": feed_interval, + "global_interval": global_interval, } return templates.TemplateResponse(request=request, name="feed.html", context=context) @@ -708,6 +868,29 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""): add_missing_tags(reader) should_send_embed: bool = bool(reader.get_tag(feed, "should_send_embed")) + # Get the update interval for this feed + feed_interval: int | None = None + try: + feed_update_config = reader.get_tag(feed, ".reader.update") + if isinstance(feed_update_config, dict) and "interval" in feed_update_config: + interval_value = feed_update_config["interval"] + if isinstance(interval_value, int): + feed_interval = interval_value + except TagNotFoundError: + # No custom interval set for this feed, will use global default + pass + + # Get the global default update interval + global_interval: int = 60 # Default to 60 minutes if not set + try: + global_update_config = reader.get_tag((), ".reader.update") + if isinstance(global_update_config, dict) and "interval" in global_update_config: + interval_value = global_update_config["interval"] + if isinstance(interval_value, int): + global_interval = interval_value + except TagNotFoundError: + pass + context = { "request": request, "feed": feed, @@ -716,8 +899,10 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""): "html": html, "should_send_embed": should_send_embed, "last_entry": last_entry, - "show_more_entires_button": show_more_entires_button, + "is_show_more_entries_button_visible": is_show_more_entries_button_visible, "total_entries": total_entries, + "feed_interval": feed_interval, + "global_interval": global_interval, } return templates.TemplateResponse(request=request, name="feed.html", context=context) @@ -847,6 +1032,56 @@ def get_data_from_hook_url(hook_name: str, hook_url: str) -> WebhookInfo: return our_hook +@app.get("/settings", response_class=HTMLResponse) +async def get_settings(request: Request): + """Settings page. + + Args: + request: The request object. + + Returns: + HTMLResponse: The settings page. + """ + # Get the global default update interval + global_interval: int = 60 # Default to 60 minutes if not set + try: + global_update_config = reader.get_tag((), ".reader.update") + if isinstance(global_update_config, dict) and "interval" in global_update_config: + interval_value = global_update_config["interval"] + if isinstance(interval_value, int): + global_interval = interval_value + except TagNotFoundError: + pass + + # Get all feeds with their intervals + feeds: Iterable[Feed] = reader.get_feeds() + feed_intervals = [] + for feed in feeds: + feed_interval: int | None = None + try: + feed_update_config = reader.get_tag(feed, ".reader.update") + if isinstance(feed_update_config, dict) and "interval" in feed_update_config: + interval_value = feed_update_config["interval"] + if isinstance(interval_value, int): + feed_interval = interval_value + except TagNotFoundError: + pass + + feed_intervals.append({ + "feed": feed, + "interval": feed_interval, + "effective_interval": feed_interval or global_interval, + "domain": extract_domain(feed.url), + }) + + context = { + "request": request, + "global_interval": global_interval, + "feed_intervals": feed_intervals, + } + return templates.TemplateResponse(request=request, name="settings.html", context=context) + + @app.get("/webhooks", response_class=HTMLResponse) async def get_webhooks(request: Request): """Page for adding a new webhook. diff --git a/discord_rss_bot/settings.py b/discord_rss_bot/settings.py index 3ef7b1a..676a185 100644 --- a/discord_rss_bot/settings.py +++ b/discord_rss_bot/settings.py @@ -6,6 +6,7 @@ from pathlib import Path from platformdirs import user_data_dir from reader import Reader +from reader import TagNotFoundError from reader import make_reader if typing.TYPE_CHECKING: @@ -39,7 +40,12 @@ def get_reader(custom_location: Path | None = None) -> Reader: reader: Reader = make_reader(url=str(db_location)) # https://reader.readthedocs.io/en/latest/api.html#reader.types.UpdateConfig - # Set the update interval to 15 minutes - reader.set_tag((), ".reader.update", {"interval": 15}) + # Set the default update interval to 15 minutes if not already configured + # Users can change this via the Settings page or per-feed in the feed page + try: + reader.get_tag((), ".reader.update") + except TagNotFoundError: + # Set default + reader.set_tag((), ".reader.update", {"interval": 15}) return reader diff --git a/discord_rss_bot/static/styles.css b/discord_rss_bot/static/styles.css index db0cfba..5758237 100644 --- a/discord_rss_bot/static/styles.css +++ b/discord_rss_bot/static/styles.css @@ -13,3 +13,7 @@ body { .form-text { color: #acabab; } + +.interval-input { + max-width: 120px; +} \ No newline at end of file diff --git a/discord_rss_bot/templates/feed.html b/discord_rss_bot/templates/feed.html index 340a8a3..d58c714 100644 --- a/discord_rss_bot/templates/feed.html +++ b/discord_rss_bot/templates/feed.html @@ -1,90 +1,145 @@ {% extends "base.html" %} {% block title %} -| {{ feed.title }} + | {{ feed.title }} {% endblock title %} {% block content %} -
- -

- {{ feed.title }} ({{ total_entries }} entries) -

- {% if not feed.updates_enabled %} - Disabled - {% endif %} - - {% if feed.last_exception %} -
-
{{ feed.last_exception.type_name }}:
- {{ feed.last_exception.value_str }} - -
- 
{{ feed.last_exception.traceback_str }}
+
+ +

+ {{ feed.title }} ({{ total_entries }} entries) +

+ {% if not feed.updates_enabled %}Disabled{% endif %} + {% if feed.last_exception %} +
+
{{ feed.last_exception.type_name }}:
+ {{ feed.last_exception.value_str }} + +
+ 
{{ feed.last_exception.traceback_str }}
+
+
+ {% endif %} + +
+ Update +
+ +
+ {% if not feed.updates_enabled %} +
+ +
+ {% else %} +
+ +
+ {% endif %} + {% if not "youtube.com/feeds/videos.xml" in feed.url %} + {% if should_send_embed %} +
+ +
+ {% else %} +
+ +
+ {% endif %} + {% endif %} +
+ +
+ Whitelist + Blacklist + + Customize message + {% if not should_send_embed %}(Currently active){% endif %} + + {% if not "youtube.com/feeds/videos.xml" in feed.url %} + + Customize embed + {% if should_send_embed %}(Currently active){% endif %} + + {% endif %} +
+ +
+
Feed Information
+
+
+ Added: {{ feed.added | relative_time }} +
+
+ Last Updated: {{ feed.last_updated | relative_time }} +
+
+ Last Retrieved: {{ feed.last_retrieved | relative_time }} +
+
+ Next Update: {{ feed.update_after | relative_time }} +
+
+ Updates: {{ 'Enabled' if feed.updates_enabled else 'Disabled' }} +
+
+
+ +
+
Update Interval
+ {% if feed_interval %} +

+ Current: {{ feed_interval }} minutes + {% if feed_interval >= 60 %}({{ (feed_interval / 60) | round(1) }} hours){% endif %} + Custom +

+ {% else %} +

+ Current: {{ global_interval }} minutes + {% if global_interval >= 60 %}({{ (global_interval / 60) | round(1) }} hours){% endif %} + Using global default +

+ {% endif %} +
+ +
+ + +
+
+ {% if feed_interval %} +
+ + +
+ {% endif %}
+ {# Rendered HTML content #} + 
{{ html|safe }}
+ {% if is_show_more_entries_button_visible %} + + Show more entries + {% endif %} - - -
- Update - -
- -
- - {% if not feed.updates_enabled %} -
- -
- {% else %} -
- -
- {% endif %} - - {% if not "youtube.com/feeds/videos.xml" in feed.url %} - {% if should_send_embed %} -
- -
- {% else %} -
- -
- {% endif %} - {% endif %} -
- - -
- Whitelist - Blacklist - - Customize message {% if not should_send_embed %}(Currently active){% endif %} - - {% if not "youtube.com/feeds/videos.xml" in feed.url %} - - Customize embed {% if should_send_embed %}(Currently active){% endif %} - - {% endif %} -
-
- -{# Rendered HTML content #} -
{{ html|safe }}
- -{% if show_more_entires_button %} - - Show more entries - -{% endif %} - {% endblock content %} diff --git a/discord_rss_bot/templates/nav.html b/discord_rss_bot/templates/nav.html index 93f3f93..7442554 100644 --- a/discord_rss_bot/templates/nav.html +++ b/discord_rss_bot/templates/nav.html @@ -19,6 +19,10 @@
  • Webhooks
    • +
  • |
    • +
  • + Settings +
    • {% if get_backup_path() %}
  • |
  • diff --git a/discord_rss_bot/templates/settings.html b/discord_rss_bot/templates/settings.html new file mode 100644 index 0000000..4ea0023 --- /dev/null +++ b/discord_rss_bot/templates/settings.html @@ -0,0 +1,122 @@ +{% extends "base.html" %} +{% block title %} + | Settings +{% endblock title %} +{% block content %} +
    +
    +
    +

    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. +
    +
    +
    +
    +
    + +
    + + +
    +
    +
    +
    +
    +
    +
    +

    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 %} + + {{ item.feed.last_updated | relative_time }} + + {{ item.feed.update_after | relative_time }} + +
    + + + + +
    +     		+ {% if item.interval %} +
    + + + +
    + {% endif %} +
    +
    + {% else %} +

    No feeds added yet.

    + {% endif %} +
    +{% endblock content %} diff --git a/pyproject.toml b/pyproject.toml index 8e5721c..2bf6c57 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,15 +86,7 @@ lint.ignore = [ [tool.ruff.lint.per-file-ignores] "tests/*" = ["S101", "D103", "PLR2004"] -[tool.ruff.lint.mccabe] -max-complexity = 15 # Don't judge lol - [tool.pytest.ini_options] -python_files = ["test_*.py"] -log_cli = true -log_cli_level = "DEBUG" -log_cli_format = "%(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)" -log_cli_date_format = "%Y-%m-%d %H:%M:%S" filterwarnings = [ "ignore::bs4.GuessedAtParserWarning", "ignore:functools\\.partial will be a method descriptor in future Python versions; wrap it in staticmethod\\(\\) if you want to preserve the old behavior:FutureWarning", diff --git a/tests/test_blacklist.py b/tests/test_blacklist.py index d2a785b..354fdca 100644 --- a/tests/test_blacklist.py +++ b/tests/test_blacklist.py @@ -4,9 +4,13 @@ import tempfile from pathlib import Path from typing import TYPE_CHECKING -from reader import Entry, Feed, Reader, make_reader +from reader import Entry +from reader import Feed +from reader import Reader +from reader import make_reader -from discord_rss_bot.filter.blacklist import entry_should_be_skipped, feed_has_blacklist_tags +from discord_rss_bot.filter.blacklist import entry_should_be_skipped +from discord_rss_bot.filter.blacklist import feed_has_blacklist_tags if TYPE_CHECKING: from collections.abc import Iterable diff --git a/tests/test_custom_filter.py b/tests/test_custom_filter.py index 94ce18e..5538608 100644 --- a/tests/test_custom_filter.py +++ b/tests/test_custom_filter.py @@ -5,7 +5,9 @@ import tempfile from pathlib import Path from typing import TYPE_CHECKING -from discord_rss_bot.custom_filters import encode_url, entry_is_blacklisted, entry_is_whitelisted +from discord_rss_bot.custom_filters import encode_url +from discord_rss_bot.custom_filters import entry_is_blacklisted +from discord_rss_bot.custom_filters import entry_is_whitelisted from discord_rss_bot.settings import get_reader if TYPE_CHECKING: diff --git a/tests/test_feeds.py b/tests/test_feeds.py index 2b3a2b4..2ffe47e 100644 --- a/tests/test_feeds.py +++ b/tests/test_feeds.py @@ -4,19 +4,20 @@ import os import tempfile from pathlib import Path from typing import LiteralString -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock +from unittest.mock import patch import pytest -from reader import Feed, Reader, make_reader +from reader import Feed +from reader import Reader +from reader import make_reader -from discord_rss_bot.feeds import ( - extract_domain, - is_youtube_feed, - send_entry_to_discord, - send_to_discord, - should_send_embed_check, - truncate_webhook_message, -) +from discord_rss_bot.feeds import extract_domain +from discord_rss_bot.feeds import is_youtube_feed +from discord_rss_bot.feeds import send_entry_to_discord +from discord_rss_bot.feeds import send_to_discord +from discord_rss_bot.feeds import should_send_embed_check +from discord_rss_bot.feeds import truncate_webhook_message from discord_rss_bot.missing_tags import add_missing_tags diff --git a/tests/test_main.py b/tests/test_main.py index dadd0cc..dd21e57 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re import urllib.parse from typing import TYPE_CHECKING @@ -289,3 +290,152 @@ def test_backup_endpoint_returns_error_when_not_configured(monkeypatch: pytest.M assert "Git backup is not configured" in response.text or "GIT_BACKUP_PATH" in response.text, ( "Error message about backup not being configured should be shown" ) + + +def test_show_more_entries_button_visible_when_many_entries() -> None: + """Test that the 'Show more entries' button is visible when there are more than 20 entries.""" + # Add the webhook first + client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) + response: Response = client.post( + url="/add_webhook", + data={"webhook_name": webhook_name, "webhook_url": webhook_url}, + ) + assert response.status_code == 200, f"Failed to add webhook: {response.text}" + + # Remove the feed if it already exists + feeds: Response = client.get(url="/") + if feed_url in feeds.text: + client.post(url="/remove", data={"feed_url": feed_url}) + + # Add the feed + response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) + assert response.status_code == 200, f"Failed to add feed: {response.text}" + + # Get the feed page + response: Response = client.get(url="/feed", params={"feed_url": feed_url}) + assert response.status_code == 200, f"Failed to get /feed: {response.text}" + + # Check if the feed has more than 20 entries by looking at the response + # The button should be visible if there are more than 20 entries + # We check for both the button text and the link structure + if "Show more entries" in response.text: + # Button is visible - verify it has the correct structure + assert "starting_after=" in response.text, "Show more entries button should contain starting_after parameter" + # The button should be a link to the feed page with pagination + assert ( + f'href="/feed?feed_url={urllib.parse.quote(feed_url)}' in response.text + or f'href="/feed?feed_url={encoded_feed_url(feed_url)}' in response.text + ), "Show more entries button should link back to the feed page" + + +def test_show_more_entries_button_not_visible_when_few_entries() -> None: + """Test that the 'Show more entries' button is not visible when there are 20 or fewer entries.""" + # Use a feed with very few entries + small_feed_url = "https://lovinator.space/rss_test_small.xml" + + # Clean up if exists + client.post(url="/remove", data={"feed_url": small_feed_url}) + + # Add a small feed (this may not exist, so this test is conditional) + response: Response = client.post(url="/add", data={"feed_url": small_feed_url, "webhook_dropdown": webhook_name}) + + if response.status_code == 200: + # Get the feed page + response: Response = client.get(url="/feed", params={"feed_url": small_feed_url}) + assert response.status_code == 200, f"Failed to get /feed: {response.text}" + + # If the feed has 20 or fewer entries, the button should not be visible + # We check the total entry count in the page + if "0 entries" in response.text or " entries)" in response.text: + # Extract entry count and verify button visibility + + match: re.Match[str] | None = re.search(r"\((\d+) entries\)", response.text) + if match: + entry_count = int(match.group(1)) + if entry_count <= 20: + assert "Show more entries" not in response.text, ( + f"Show more entries button should not be visible when there are {entry_count} entries" + ) + + # Clean up + client.post(url="/remove", data={"feed_url": small_feed_url}) + + +def test_show_more_entries_pagination_works() -> None: + """Test that pagination with starting_after parameter works correctly.""" + # Add the webhook first + client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) + response: Response = client.post( + url="/add_webhook", + data={"webhook_name": webhook_name, "webhook_url": webhook_url}, + ) + assert response.status_code == 200, f"Failed to add webhook: {response.text}" + + # Remove the feed if it already exists + feeds: Response = client.get(url="/") + if feed_url in feeds.text: + client.post(url="/remove", data={"feed_url": feed_url}) + + # Add the feed + response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) + assert response.status_code == 200, f"Failed to add feed: {response.text}" + + # Get the first page + response: Response = client.get(url="/feed", params={"feed_url": feed_url}) + assert response.status_code == 200, f"Failed to get /feed: {response.text}" + + # Check if pagination is available + if "Show more entries" in response.text and "starting_after=" in response.text: + # Extract the starting_after parameter from the button link + match: re.Match[str] | None = re.search(r'starting_after=([^"&]+)', response.text) + if match: + starting_after_id: str = match.group(1) + + # Request the second page + response: Response = client.get( + url="/feed", params={"feed_url": feed_url, "starting_after": starting_after_id} + ) + assert response.status_code == 200, f"Failed to get paginated feed: {response.text}" + + # Verify we got a valid response (the page should contain entries) + assert "entries)" in response.text, "Paginated page should show entry count" + + +def test_show_more_entries_button_context_variable() -> None: + """Test that the button visibility variable is correctly passed to the template context.""" + # Add the webhook first + client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) + response: Response = client.post( + url="/add_webhook", + data={"webhook_name": webhook_name, "webhook_url": webhook_url}, + ) + assert response.status_code == 200, f"Failed to add webhook: {response.text}" + + # Remove the feed if it already exists + feeds: Response = client.get(url="/") + if feed_url in feeds.text: + client.post(url="/remove", data={"feed_url": feed_url}) + + # Add the feed + response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) + assert response.status_code == 200, f"Failed to add feed: {response.text}" + + # Get the feed page + response: Response = client.get(url="/feed", params={"feed_url": feed_url}) + assert response.status_code == 200, f"Failed to get /feed: {response.text}" + + # Extract the total entries count from the page + match: re.Match[str] | None = re.search(r"\((\d+) entries\)", response.text) + if match: + entry_count = int(match.group(1)) + + # If more than 20 entries, button should be visible + if entry_count > 20: + assert "Show more entries" in response.text, ( + f"Button should be visible when there are {entry_count} entries (more than 20)" + ) + # If 20 or fewer entries, button should not be visible + else: + assert "Show more entries" not in response.text, ( + f"Button should not be visible when there are {entry_count} entries (20 or fewer)" + ) diff --git a/tests/test_search.py b/tests/test_search.py index 2819734..ada1464 100644 --- a/tests/test_search.py +++ b/tests/test_search.py @@ -4,7 +4,9 @@ import tempfile from pathlib import Path from typing import TYPE_CHECKING -from reader import Feed, Reader, make_reader +from reader import Feed +from reader import Reader +from reader import make_reader from discord_rss_bot.search import create_search_context diff --git a/tests/test_settings.py b/tests/test_settings.py index dd5b44e..5a54094 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -6,7 +6,9 @@ from pathlib import Path from reader import Reader -from discord_rss_bot.settings import data_dir, default_custom_message, get_reader +from discord_rss_bot.settings import data_dir +from discord_rss_bot.settings import default_custom_message +from discord_rss_bot.settings import get_reader def test_reader() -> None: diff --git a/tests/test_update_interval.py b/tests/test_update_interval.py new file mode 100644 index 0000000..12e29a9 --- /dev/null +++ b/tests/test_update_interval.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import urllib.parse +from typing import TYPE_CHECKING + +from fastapi.testclient import TestClient + +from discord_rss_bot.main import app + +if TYPE_CHECKING: + from httpx import Response + +client: TestClient = TestClient(app) +webhook_name: str = "Test Webhook for Update Interval" +webhook_url: str = "https://discord.com/api/webhooks/1234567890/test_update_interval" +feed_url: str = "https://lovinator.space/rss_test.xml" + + +def test_global_update_interval() -> None: + """Test setting the global update interval.""" + # Set global update interval to 30 minutes + response: Response = client.post("/set_global_update_interval", data={"interval_minutes": "30"}) + assert response.status_code == 200, f"Failed to set global interval: {response.text}" + + # Check that the settings page shows the new interval + response = client.get("/settings") + assert response.status_code == 200, f"Failed to get settings page: {response.text}" + assert "30" in response.text, "Global interval not updated on settings page" + + +def test_per_feed_update_interval() -> None: + """Test setting per-feed update interval.""" + # Clean up any existing feed/webhook + client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) + client.post(url="/remove", data={"feed_url": feed_url}) + + # Add webhook and feed + response: Response = client.post( + url="/add_webhook", + data={"webhook_name": webhook_name, "webhook_url": webhook_url}, + ) + assert response.status_code == 200, f"Failed to add webhook: {response.text}" + + response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) + assert response.status_code == 200, f"Failed to add feed: {response.text}" + + # Set feed-specific update interval to 15 minutes + response = client.post("/set_update_interval", data={"feed_url": feed_url, "interval_minutes": "15"}) + assert response.status_code == 200, f"Failed to set feed interval: {response.text}" + + # Check that the feed page shows the custom interval + encoded_url = urllib.parse.quote(feed_url) + response = client.get(f"/feed?feed_url={encoded_url}") + assert response.status_code == 200, f"Failed to get feed page: {response.text}" + assert "15" in response.text, "Feed interval not displayed on feed page" + assert "Custom" in response.text, "Custom badge not shown for feed-specific interval" + + +def test_reset_feed_update_interval() -> None: + """Test resetting feed update interval to global default.""" + # First set a custom interval + response: Response = client.post("/set_update_interval", data={"feed_url": feed_url, "interval_minutes": "15"}) + assert response.status_code == 200, f"Failed to set feed interval: {response.text}" + + # Reset to global default + response = client.post("/reset_update_interval", data={"feed_url": feed_url}) + assert response.status_code == 200, f"Failed to reset feed interval: {response.text}" + + # Check that the feed page shows global default + encoded_url = urllib.parse.quote(feed_url) + response = client.get(f"/feed?feed_url={encoded_url}") + assert response.status_code == 200, f"Failed to get feed page: {response.text}" + assert "Using global default" in response.text, "Global default badge not shown after reset" + + +def test_update_interval_validation() -> None: + """Test that update interval validation works.""" + # Try to set an interval below minimum (should be clamped to 1) + response: Response = client.post("/set_global_update_interval", data={"interval_minutes": "0"}) + assert response.status_code == 200, f"Failed to handle minimum interval: {response.text}" + + # Try to set an interval above maximum (should be clamped to 10080) + response = client.post("/set_global_update_interval", data={"interval_minutes": "20000"}) + assert response.status_code == 200, f"Failed to handle maximum interval: {response.text}" + + # Clean up + client.post(url="/remove", data={"feed_url": feed_url}) + client.post(url="/delete_webhook", data={"webhook_url": webhook_url})