Randomize test order

discord_rss_bot/filter/blacklist.py

@@ -2,10 +2,13 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from discord_rss_bot.filter.utils import is_regex_match, is_word_in_text
+from discord_rss_bot.filter.utils import is_regex_match
+from discord_rss_bot.filter.utils import is_word_in_text
 
 if TYPE_CHECKING:
-    from reader import Entry, Feed, Reader
+    from reader import Entry
+    from reader import Feed
+    from reader import Reader
 
 
 def feed_has_blacklist_tags(custom_reader: Reader, feed: Feed) -> bool:

discord_rss_bot/filter/whitelist.py

@@ -2,10 +2,13 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from discord_rss_bot.filter.utils import is_regex_match, is_word_in_text
+from discord_rss_bot.filter.utils import is_regex_match
+from discord_rss_bot.filter.utils import is_word_in_text
 
 if TYPE_CHECKING:
-    from reader import Entry, Feed, Reader
+    from reader import Entry
+    from reader import Feed
+    from reader import Reader
 
 
 def has_white_tags(custom_reader: Reader, feed: Feed) -> bool:

discord_rss_bot/git_backup.py

@@ -44,7 +44,6 @@ type TAG_VALUE = (
     | list[str | int | float | bool | dict[str, Any] | list[Any] | None]
     | None
 )
-"""Type alias for the value of a feed tag, which can be a nested structure of dicts and lists, or None."""
 
 # Tags that are exported per-feed (empty values are omitted).
 _FEED_TAGS: tuple[str, ...] = (
@@ -179,7 +178,7 @@ def export_state(reader: Reader, backup_path: Path) -> None:
 
     try:
         webhooks: list[str | int | float | bool | dict[str, Any] | list[Any] | None] = list(
-            reader.get_tag((), "webhooks", [])
+            reader.get_tag((), "webhooks", []),
         )
     except TagNotFoundError:
         webhooks = []

discord_rss_bot/main.py

@@ -31,8 +31,10 @@ from markdownify import markdownify
 from reader import Entry
 from reader import EntryNotFoundError
 from reader import Feed
+from reader import FeedExistsError
 from reader import FeedNotFoundError
 from reader import Reader
+from reader import ReaderError
 from reader import TagNotFoundError
 from starlette.responses import RedirectResponse
 
@@ -697,6 +699,45 @@ async def post_set_update_interval(
     return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303)
 
 
+@app.post("/change_feed_url")
+async def post_change_feed_url(
+    old_feed_url: Annotated[str, Form()],
+    new_feed_url: Annotated[str, Form()],
+) -> RedirectResponse:
+    """Change the URL for an existing feed.
+
+    Args:
+        old_feed_url: Current feed URL.
+        new_feed_url: New feed URL to change to.
+
+    Returns:
+        RedirectResponse: Redirect to the feed page for the resulting URL.
+
+    Raises:
+        HTTPException: If the old feed is not found, the new URL already exists, or change fails.
+    """
+    clean_old_feed_url: str = old_feed_url.strip()
+    clean_new_feed_url: str = new_feed_url.strip()
+
+    if not clean_old_feed_url or not clean_new_feed_url:
+        raise HTTPException(status_code=400, detail="Feed URLs cannot be empty")
+
+    if clean_old_feed_url == clean_new_feed_url:
+        return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_old_feed_url)}", status_code=303)
+
+    try:
+        reader.change_feed_url(clean_old_feed_url, clean_new_feed_url)
+    except FeedNotFoundError as e:
+        raise HTTPException(status_code=404, detail=f"Feed not found: {clean_old_feed_url}") from e
+    except FeedExistsError as e:
+        raise HTTPException(status_code=409, detail=f"Feed already exists: {clean_new_feed_url}") from e
+    except ReaderError as e:
+        raise HTTPException(status_code=400, detail=f"Failed to change feed URL: {e}") from e
+
+    commit_state_change(reader, f"Change feed URL from {clean_old_feed_url} to {clean_new_feed_url}")
+    return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_new_feed_url)}", status_code=303)
+
+
 @app.post("/reset_update_interval")
 async def post_reset_update_interval(
     feed_url: Annotated[str, Form()],
@@ -804,7 +845,7 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
         except EntryNotFoundError as e:
             current_entries = list(reader.get_entries(feed=clean_feed_url))
             msg: str = f"{e}\n\n{[entry.id for entry in current_entries]}"
-            html: str = create_html_for_feed(current_entries)
+            html: str = create_html_for_feed(current_entries, clean_feed_url)
 
             # Get feed and global intervals for error case too
             feed_interval: int | None = None
@@ -860,7 +901,7 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
         last_entry = entries[-1]
 
     # Create the html for the entries.
-    html: str = create_html_for_feed(entries)
+    html: str = create_html_for_feed(entries, clean_feed_url)
 
     try:
         should_send_embed: bool = bool(reader.get_tag(feed, "should_send_embed"))
@@ -907,11 +948,12 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
     return templates.TemplateResponse(request=request, name="feed.html", context=context)
 
 
-def create_html_for_feed(entries: Iterable[Entry]) -> str:
+def create_html_for_feed(entries: Iterable[Entry], current_feed_url: str = "") -> str:  # noqa: PLR0914
     """Create HTML for the search results.
 
     Args:
         entries: The entries to create HTML for.
+        current_feed_url: The feed URL currently being viewed in /feed.
 
     Returns:
         str: The HTML for the search results.
@@ -940,6 +982,12 @@ def create_html_for_feed(entries: Iterable[Entry]) -> str:
         if entry_is_whitelisted(entry):
             whitelisted = "<span class='badge bg-success'>Whitelisted</span>"
 
+        source_feed_url: str = getattr(entry, "original_feed_url", None) or entry.feed.url
+
+        from_another_feed: str = ""
+        if current_feed_url and source_feed_url != current_feed_url:
+            from_another_feed = f"<span class='badge bg-warning text-dark'>From another feed: {source_feed_url}</span>"
+
         entry_id: str = urllib.parse.quote(entry.id)
         to_discord_html: str = f"<a class='text-muted' href='/post_entry?entry_id={entry_id}'>Send to Discord</a>"
 
@@ -966,14 +1014,14 @@ def create_html_for_feed(entries: Iterable[Entry]) -> str:
         image_html: str = f"<img src='{first_image}' class='img-fluid'>" if first_image else ""
 
         html += f"""<div class="p-2 mb-2 border border-dark">
-{blacklisted}{whitelisted}<a class="text-muted text-decoration-none" href="{entry.link}"><h2>{entry.title}</h2></a>
+{blacklisted}{whitelisted}{from_another_feed}<a class="text-muted text-decoration-none" href="{entry.link}"><h2>{entry.title}</h2></a>
 {f"By {entry.author} @" if entry.author else ""}{published} - {to_discord_html}
 
 {text}
 {video_embed_html}
 {image_html}
 </div>
-"""
+"""  # noqa: E501
     return html.strip()
 
 

discord_rss_bot/settings.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import typing
 from functools import lru_cache
 from pathlib import Path
@@ -12,7 +13,12 @@ from reader import make_reader
 if typing.TYPE_CHECKING:
     from reader.types import JSONType
 
-data_dir: str = user_data_dir(appname="discord_rss_bot", appauthor="TheLovinator", roaming=True, ensure_exists=True)
+data_dir: str = os.getenv("DISCORD_RSS_BOT_DATA_DIR", "").strip() or user_data_dir(
+    appname="discord_rss_bot",
+    appauthor="TheLovinator",
+    roaming=True,
+    ensure_exists=True,
+)
 
 
 # TODO(TheLovinator): Add default things to the database and make the edible.

discord_rss_bot/templates/feed.html

@@ -76,6 +76,22 @@
                 </a>
             {% endif %}
         </div>
+        <!-- Feed URL Configuration -->
+        <div class="mt-4 border-top border-secondary pt-3">
+            <h5 class="mb-3">Feed URL</h5>
+            <p class="text-muted mb-2">Change the URL for this feed. This can be useful if a feed has moved.</p>
+            <form action="/change_feed_url" method="post" class="mb-2">
+                <input type="hidden" name="old_feed_url" value="{{ feed.url }}" />
+                <div class="input-group input-group-sm mb-2">
+                    <input type="url"
+                           class="form-control form-control-sm"
+                           name="new_feed_url"
+                           value="{{ feed.url }}"
+                           required />
+                    <button class="btn btn-warning" type="submit">Update URL</button>
+                </div>
+            </form>
+        </div>
         <!-- Feed Metadata -->
         <div class="mt-4 border-top border-secondary pt-3">
             <h5 class="mb-3">Feed Information</h5>

pyproject.toml

@@ -22,7 +22,7 @@ dependencies = [
 ]
 
 [dependency-groups]
-dev = ["djlint", "pytest"]
+dev = ["djlint", "pytest", "pytest-randomly", "pytest-xdist"]
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]
@@ -87,6 +87,7 @@ lint.ignore = [
 "tests/*" = ["S101", "D103", "PLR2004"]
 
 [tool.pytest.ini_options]
+addopts = "-n 5 --dist loadfile"
 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",

tests/conftest.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import os
+import shutil
+import sys
+import tempfile
+from contextlib import suppress
+from pathlib import Path
+from typing import Any
+
+
+def pytest_configure() -> None:
+    """Isolate persistent app state per xdist worker to avoid cross-worker test interference."""
+    worker_id: str = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
+    worker_data_dir: Path = Path(tempfile.gettempdir()) / "discord-rss-bot-tests" / worker_id
+
+    # Start each worker from a clean state.
+    shutil.rmtree(worker_data_dir, ignore_errors=True)
+    worker_data_dir.mkdir(parents=True, exist_ok=True)
+
+    os.environ["DISCORD_RSS_BOT_DATA_DIR"] = str(worker_data_dir)
+
+    # If modules were imported before this hook (unlikely), force them to use
+    # the worker-specific location.
+    settings_module: Any = sys.modules.get("discord_rss_bot.settings")
+    if settings_module is not None:
+        settings_module.data_dir = str(worker_data_dir)
+        get_reader: Any = getattr(settings_module, "get_reader", None)
+        if get_reader is not None and hasattr(get_reader, "cache_clear"):
+            get_reader.cache_clear()
+
+    main_module: Any = sys.modules.get("discord_rss_bot.main")
+    if main_module is not None and settings_module is not None:
+        with suppress(Exception):
+            current_reader = getattr(main_module, "reader", None)
+            if current_reader is not None:
+                current_reader.close()
+        get_reader: Any = getattr(settings_module, "get_reader", None)
+        if callable(get_reader):
+            main_module.reader = get_reader()

tests/test_git_backup.py

@@ -25,7 +25,8 @@ if TYPE_CHECKING:
 
 
 SKIP_IF_NO_GIT: pytest.MarkDecorator = pytest.mark.skipif(
-    shutil.which("git") is None, reason="git executable not found"
+    shutil.which("git") is None,
+    reason="git executable not found",
 )
 
 

tests/test_main.py

@@ -2,17 +2,22 @@ from __future__ import annotations
 
 import re
 import urllib.parse
+from dataclasses import dataclass
+from dataclasses import field
 from typing import TYPE_CHECKING
+from typing import cast
 
 from fastapi.testclient import TestClient
 
 from discord_rss_bot.main import app
+from discord_rss_bot.main import create_html_for_feed
 
 if TYPE_CHECKING:
     from pathlib import Path
 
     import pytest
     from httpx import Response
+    from reader import Entry
 
 client: TestClient = TestClient(app)
 webhook_name: str = "Hello, I am a webhook!"
@@ -76,6 +81,14 @@ def test_add_webhook() -> None:
 
 def test_create_feed() -> None:
     """Test the /create_feed page."""
+    # Ensure webhook exists for this test regardless of test order.
+    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 before we run the test.
     feeds: Response = client.get(url="/")
     if feed_url in feeds.text:
@@ -94,6 +107,14 @@ def test_create_feed() -> None:
 
 def test_get() -> None:
     """Test the /create_feed page."""
+    # Ensure webhook exists for this test regardless of test order.
+    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 before we run the test.
     feeds: Response = client.get("/")
     if feed_url in feeds.text:
@@ -139,6 +160,14 @@ def test_get() -> None:
 
 def test_pause_feed() -> None:
     """Test the /pause_feed page."""
+    # Ensure webhook exists for this test regardless of test order.
+    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 before we run the test.
     feeds: Response = client.get(url="/")
     if feed_url in feeds.text:
@@ -147,6 +176,7 @@ def test_pause_feed() -> None:
 
     # 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}"
 
     # Unpause the feed if it is paused.
     feeds: Response = client.get(url="/")
@@ -166,6 +196,14 @@ def test_pause_feed() -> None:
 
 def test_unpause_feed() -> None:
     """Test the /unpause_feed page."""
+    # Ensure webhook exists for this test regardless of test order.
+    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 before we run the test.
     feeds: Response = client.get("/")
     if feed_url in feeds.text:
@@ -174,6 +212,7 @@ def test_unpause_feed() -> None:
 
     # 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}"
 
     # Pause the feed if it is unpaused.
     feeds: Response = client.get(url="/")
@@ -193,6 +232,14 @@ def test_unpause_feed() -> None:
 
 def test_remove_feed() -> None:
     """Test the /remove page."""
+    # Ensure webhook exists for this test regardless of test order.
+    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 before we run the test.
     feeds: Response = client.get(url="/")
     if feed_url in feeds.text:
@@ -201,6 +248,7 @@ def test_remove_feed() -> None:
 
     # 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}"
 
     # Remove the feed.
     response: Response = client.post(url="/remove", data={"feed_url": feed_url})
@@ -212,6 +260,45 @@ def test_remove_feed() -> None:
     assert feed_url not in response.text, f"Feed found in /: {response.text}"
 
 
+def test_change_feed_url() -> None:
+    """Test changing a feed URL from the feed page endpoint."""
+    new_feed_url = "https://lovinator.space/rss_test_small.xml"
+
+    # Ensure test feeds do not already exist.
+    client.post(url="/remove", data={"feed_url": feed_url})
+    client.post(url="/remove", data={"feed_url": new_feed_url})
+
+    # Ensure webhook exists.
+    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}"
+
+    # Add the original feed.
+    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}"
+
+    # Change feed URL.
+    response = client.post(
+        url="/change_feed_url",
+        data={"old_feed_url": feed_url, "new_feed_url": new_feed_url},
+    )
+    assert response.status_code == 200, f"Failed to change feed URL: {response.text}"
+
+    # New feed should be accessible.
+    response = client.get(url="/feed", params={"feed_url": new_feed_url})
+    assert response.status_code == 200, f"New feed URL is not accessible: {response.text}"
+
+    # Old feed should no longer be accessible.
+    response = client.get(url="/feed", params={"feed_url": feed_url})
+    assert response.status_code == 404, "Old feed URL should no longer exist"
+
+    # Cleanup.
+    client.post(url="/remove", data={"feed_url": new_feed_url})
+
+
 def test_delete_webhook() -> None:
     """Test the /delete_webhook page."""
     # Remove the feed if it already exists before we run the test.
@@ -330,6 +417,14 @@ def test_show_more_entries_button_visible_when_many_entries() -> None:
 
 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."""
+    # Ensure webhook exists for this test regardless of test order.
+    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}"
+
     # Use a feed with very few entries
     small_feed_url = "https://lovinator.space/rss_test_small.xml"
 
@@ -393,7 +488,8 @@ def test_show_more_entries_pagination_works() -> None:
 
             # Request the second page
             response: Response = client.get(
-                url="/feed", params={"feed_url": feed_url, "starting_after": starting_after_id}
+                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}"
 
@@ -439,3 +535,49 @@ def test_show_more_entries_button_context_variable() -> None:
             assert "Show more entries" not in response.text, (
                 f"Button should not be visible when there are {entry_count} entries (20 or fewer)"
             )
+
+
+def test_create_html_marks_entries_from_another_feed(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Entries from another feed should be marked in /feed html output."""
+
+    @dataclass(slots=True)
+    class DummyContent:
+        value: str
+
+    @dataclass(slots=True)
+    class DummyFeed:
+        url: str
+
+    @dataclass(slots=True)
+    class DummyEntry:
+        feed: DummyFeed
+        id: str
+        original_feed_url: str | None = None
+        link: str = "https://example.com/post"
+        title: str = "Example title"
+        author: str = "Author"
+        summary: str = "Summary"
+        content: list[DummyContent] = field(default_factory=lambda: [DummyContent("Content")])
+        published: None = None
+
+        def __post_init__(self) -> None:
+            if self.original_feed_url is None:
+                self.original_feed_url = self.feed.url
+
+    selected_feed_url = "https://example.com/feed-a.xml"
+    same_feed_entry = DummyEntry(DummyFeed(selected_feed_url), "same")
+    # feed.url matches selected feed, but original_feed_url differs; marker should still show.
+    other_feed_entry = DummyEntry(
+        DummyFeed(selected_feed_url),
+        "other",
+        original_feed_url="https://example.com/feed-b.xml",
+    )
+
+    monkeypatch.setattr("discord_rss_bot.main.replace_tags_in_text_message", lambda _entry: "Rendered content")
+    monkeypatch.setattr("discord_rss_bot.main.entry_is_blacklisted", lambda _entry: False)
+    monkeypatch.setattr("discord_rss_bot.main.entry_is_whitelisted", lambda _entry: False)
+
+    html = create_html_for_feed(cast("list[Entry]", [same_feed_entry, other_feed_entry]), selected_feed_url)
+
+    assert "From another feed: https://example.com/feed-b.xml" in html
+    assert "From another feed: https://example.com/feed-a.xml" not in html

tests/test_update_interval.py

@@ -58,8 +58,21 @@ def test_per_feed_update_interval() -> None:
 
 def test_reset_feed_update_interval() -> None:
     """Test resetting feed update interval to global default."""
+    # Ensure feed/webhook setup exists regardless of test order
+    client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
+    client.post(url="/remove", data={"feed_url": feed_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}"
+
+    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}"
+
     # First set a custom interval
-    response: Response = client.post("/set_update_interval", data={"feed_url": feed_url, "interval_minutes": "15"})
+    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