Remove redundant url unquoting in route handlers

discord_rss_bot/main.py

@@ -709,7 +709,7 @@ async def post_attach_feed_webhook(
     Raises:
         HTTPException: If feed or webhook cannot be found.
     """
-    clean_feed_url: str = urllib.parse.unquote(feed_url.strip())
+    clean_feed_url: str = feed_url.strip()
     selected_webhook_name: str = webhook_dropdown.strip()
 
     try:
@@ -834,7 +834,8 @@ async def get_whitelist(
         HTMLResponse: The whitelist page.
     """
     clean_feed_url: str = feed_url.strip()
-    feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url))
+    feed: Feed = reader.get_feed(clean_feed_url)
+
     context = {
         "request": request,
         "feed": feed,
@@ -876,7 +877,7 @@ async def get_whitelist_preview(
     Returns:
         HTMLResponse: Rendered filter preview fragment.
     """
-    clean_feed_url: str = urllib.parse.unquote(feed_url.strip())
+    clean_feed_url: str = feed_url.strip()
     feed: Feed = reader.get_feed(clean_feed_url)
 
     form_values: dict[str, str] = {
@@ -963,7 +964,8 @@ async def get_blacklist(
     Returns:
         HTMLResponse: The blacklist page.
     """
-    feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url))
+    clean_feed_url: str = feed_url.strip()
+    feed: Feed = reader.get_feed(clean_feed_url)
 
     context = {
         "request": request,
@@ -1006,8 +1008,9 @@ async def get_blacklist_preview(
     Returns:
         HTMLResponse: Rendered filter preview fragment.
     """
-    clean_feed_url: str = urllib.parse.unquote(feed_url.strip())
+    clean_feed_url: str = feed_url.strip()
     feed: Feed = reader.get_feed(clean_feed_url)
+
     form_values: dict[str, str] = {
         "blacklist_title": blacklist_title,
         "blacklist_summary": blacklist_summary,
@@ -1384,7 +1387,7 @@ async def post_set_custom(
     reader.set_tag(feed_url, "message_avatar_url", typing.cast("JSONType", message_avatar_url.strip()))
 
     clean_feed_url: str = feed_url.strip()
-    feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url))
+    feed: Feed = reader.get_feed(clean_feed_url)
 
     stored_custom_message: str = get_custom_message(reader, feed)
     if our_custom_message != stored_custom_message:
@@ -1410,7 +1413,8 @@ async def get_custom(
     Returns:
         HTMLResponse: The custom message page.
     """
-    feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url.strip()))
+    clean_feed_url: str = feed_url.strip()
+    feed: Feed = reader.get_feed(clean_feed_url)
 
     # Collect extension variable names for the template.
     ext_registry: dict[str, type] = get_extension_registry()
@@ -1466,7 +1470,8 @@ async def get_embed_page(
     Returns:
         HTMLResponse: The custom message page.
     """
-    feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url.strip()))
+    clean_feed_url: str = feed_url.strip()
+    feed: Feed = reader.get_feed(clean_feed_url)
 
     embed: CustomEmbed = get_embed(reader, feed)
 
@@ -1562,7 +1567,7 @@ async def post_embed(  # ruff:ignore[complex-structure, too-many-branches]
         RedirectResponse: Redirect to the embed page.
     """
     clean_feed_url: str = feed_url.strip()
-    feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url))
+    feed: Feed = reader.get_feed(clean_feed_url)
 
     custom_embed: CustomEmbed = get_embed(reader, feed)
 
@@ -1619,7 +1624,9 @@ async def get_extensions(
     Returns:
         HTMLResponse: The extensions configuration page.
     """
-    feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url.strip()))
+    clean_feed_url: str = feed_url.strip()
+    feed: Feed = reader.get_feed(clean_feed_url)
+
     registry: dict[str, type] = get_extension_registry()
     enabled: list[str] = get_enabled_extensions(reader, feed.url)
 
@@ -2128,7 +2135,7 @@ async def get_feed(  # ruff:ignore[complex-structure, too-many-branches, too-man
     """
     entries_per_page: int = 20
 
-    clean_feed_url: str = urllib.parse.unquote(feed_url.strip())
+    clean_feed_url: str = feed_url.strip()
 
     try:
         feed: Feed = reader.get_feed(clean_feed_url)
@@ -2557,8 +2564,8 @@ async def get_sent_webhooks(
     Returns:
         sent_webhooks.html HTML
     """
-    clean_feed_url: str = urllib.parse.unquote(feed_url.strip())
-    clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip())
+    clean_feed_url: str = feed_url.strip()
+    clean_webhook_url: str = webhook_url.strip()
 
     records: list[SentWebhookRecord] = get_sent_webhook_records(reader)
     if clean_feed_url:
@@ -2673,12 +2680,13 @@ async def remove_feed(
     Raises:
         HTTPException: Feed not found
     """
+    clean_feed_url: str = feed_url.strip()
     try:
-        reader.delete_feed(urllib.parse.unquote(feed_url))
+        reader.delete_feed(clean_feed_url)
     except FeedNotFoundError as e:
         raise HTTPException(status_code=404, detail="Feed not found") from e
 
-    commit_state_change(reader, f"Remove feed {urllib.parse.unquote(feed_url)}")
+    commit_state_change(reader, f"Remove feed {clean_feed_url}")
 
     return RedirectResponse(url="/", status_code=303)
 
@@ -2703,7 +2711,7 @@ async def update_feed(
         HTTPException: If the feed is not found.
     """
     try:
-        clean_feed_url: str = urllib.parse.unquote(feed_url)
+        clean_feed_url: str = feed_url.strip()
         modified_entries: list[tuple[str, str]] = update_feed_and_collect_modified_entries(reader, clean_feed_url)
     except FeedNotFoundError as e:
         raise HTTPException(status_code=404, detail="Feed not found") from e
@@ -3239,19 +3247,18 @@ async def post_entry(
     Returns:
         RedirectResponse: Redirect to the feed page.
     """
-    unquoted_entry_id: str = urllib.parse.unquote(entry_id)
-    clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) if feed_url else ""
+    clean_feed_url: str = feed_url.strip() if feed_url else ""
 
     # Prefer feed-scoped lookup when feed_url is provided. This avoids ambiguity when
     # multiple feeds contain entries with the same ID.
     entry: Entry | None = None
     if clean_feed_url:
         entry = next(
-            (entry for entry in reader.get_entries(feed=clean_feed_url) if entry.id == unquoted_entry_id),
+            (entry for entry in reader.get_entries(feed=clean_feed_url) if entry.id == entry_id),
             None,
         )
     else:
-        entry = next((entry for entry in reader.get_entries() if entry.id == unquoted_entry_id), None)
+        entry = next((entry for entry in reader.get_entries() if entry.id == entry_id), None)
 
     if entry is None:
         return HTMLResponse(status_code=404, content=f"Entry '{entry_id}' not found.")
@@ -3511,8 +3518,9 @@ async def get_webhook_entries_mass_update_preview(
     Returns:
         HTMLResponse: Rendered partial template containing summary + preview table.
     """
-    clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip())
+    clean_webhook_url: str = webhook_url.strip()
     all_feeds: list[Feed] = list(reader.get_feeds())
+
     webhook_feeds: list[Feed] = [
         feed for feed in all_feeds if str(reader.get_tag(feed.url, "webhook", "")) == clean_webhook_url
     ]
@@ -3564,7 +3572,7 @@ async def get_webhook_entries(  # ruff:ignore[complex-structure, too-many-locals
         HTTPException: If no feeds are found for this webhook or webhook doesn't exist.
     """
     entries_per_page: int = 20
-    clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip())
+    clean_webhook_url: str = webhook_url.strip()
 
     # Get the webhook name from the webhooks list
     webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
@@ -3656,7 +3664,7 @@ async def get_webhook_entries(  # ruff:ignore[complex-structure, too-many-locals
         "is_show_more_entries_button_visible": is_show_more_entries_button_visible,
         "total_entries": total_entries,
         "feeds_count": len(webhook_feeds),
-        "message": urllib.parse.unquote(message) if message else "",
+        "message": message,
         **mass_update_context,
     }
     return templates.TemplateResponse(request=request, name="webhook_entries.html", context=context)
@@ -3687,7 +3695,7 @@ async def post_bulk_change_feed_urls(  # ruff:ignore[complex-structure, too-many
     Raises:
         HTTPException: If webhook is missing or replace_from is empty.
     """
-    clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip())
+    clean_webhook_url: str = webhook_url.strip()
     clean_replace_from: str = replace_from.strip()
     clean_replace_to: str = replace_to.strip()
 

tests/test_main.py

@@ -71,7 +71,6 @@ def test_search() -> None:
     feeds: Response = client.get("/")
     if feed_url in feeds.text:
         client.post(url="/remove", data={"feed_url": feed_url})
-        client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)})
 
     # Delete the webhook if it already exists before we run the test.
     response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
@@ -181,7 +180,6 @@ def test_create_feed() -> None:
     feeds: Response = client.get(url="/")
     if feed_url in feeds.text:
         client.post(url="/remove", data={"feed_url": feed_url})
-        client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)})
 
     # Add the feed.
     response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
@@ -255,7 +253,6 @@ def test_get() -> None:
     feeds: Response = client.get("/")
     if feed_url in feeds.text:
         client.post(url="/remove", data={"feed_url": feed_url})
-        client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)})
 
     # Add the feed.
     response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
@@ -272,16 +269,16 @@ def test_get() -> None:
     response: Response = client.get(url="/add_webhook")
     assert response.status_code == 200, f"/add_webhook failed: {response.text}"
 
-    response: Response = client.get(url="/blacklist", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/blacklist", params={"feed_url": feed_url})
     assert response.status_code == 200, f"/blacklist failed: {response.text}"
 
-    response: Response = client.get(url="/custom", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/custom", params={"feed_url": feed_url})
     assert response.status_code == 200, f"/custom failed: {response.text}"
 
-    response: Response = client.get(url="/embed", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/embed", params={"feed_url": feed_url})
     assert response.status_code == 200, f"/embed failed: {response.text}"
 
-    response: Response = client.get(url="/feed", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/feed", params={"feed_url": feed_url})
     assert response.status_code == 200, f"/feed failed: {response.text}"
 
     response: Response = client.get(url="/")
@@ -293,7 +290,7 @@ def test_get() -> None:
     response = client.get(url="/webhook_entries", params={"webhook_url": webhook_url})
     assert response.status_code == 200, f"/webhook_entries failed: {response.text}"
 
-    response: Response = client.get(url="/whitelist", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/whitelist", params={"feed_url": feed_url})
     assert response.status_code == 200, f"/whitelist failed: {response.text}"
 
 
@@ -351,10 +348,98 @@ def test_feed_page_shows_steam_thumbnail_hint_for_steam_feeds() -> None:
         app.dependency_overrides = {}
 
 
+@pytest.mark.parametrize(
+    "encoded_url",
+    [
+        pytest.param(
+            "https://gazellegames.net/feeds.php?feed=test&name=Halo%3A+Campaign+Evolved",
+            id="percent-encoded-colon",
+        ),
+        pytest.param(
+            "https://example.com/path%2Fwith%2Fslashes?q=val",
+            id="percent-encoded-slash",
+        ),
+        pytest.param(
+            "https://example.com/feed?title=A%23B%26C%3D",
+            id="percent-encoded-special-chars",
+        ),
+        pytest.param(
+            "https://example.com/feed?name=with+plus+signs",
+            id="literal-plus-signs",
+        ),
+        pytest.param(
+            "https://example.com/feed?filter=%7B%22key%22%3A+%22val%22%7D",
+            id="percent-encoded-json",
+        ),
+        pytest.param(
+            "https://example.com/feed?redirect=https%3A%2F%2Fother.com%2F",
+            id="percent-encoded-nested-url",
+        ),
+    ],
+)
+def test_feed_page_with_percent_encoded_url(encoded_url: str) -> None:
+    """Feed lookups with percent-encoded URLs should not double-decode them.
+
+    FastAPI already decodes query parameters. Previously, get_feed() called
+    urllib.parse.unquote() on the already-decoded value, which caused %3A to
+    become : and broke the lookup. This test verifies that all common
+    percent-encoded characters survive the query parameter round-trip.
+    """
+
+    @dataclass(slots=True)
+    class DummyFeed:
+        url: str
+        title: str
+        updates_enabled: bool = True
+        last_exception: None = None
+        added: None = None
+        last_updated: None = None
+        last_retrieved: None = None
+        update_after: None = None
+
+    class StubReader:
+        def __init__(self) -> None:
+            self.feed = DummyFeed(url=encoded_url, title="Test Feed")
+
+        def get_feed(self, feed_url: str) -> DummyFeed:
+            assert feed_url == self.feed.url, f"Expected {self.feed.url!r}, got {feed_url!r}"
+            return self.feed
+
+        def get_tag(self, _resource: str | DummyFeed, key: str, default: TestTagValue = None) -> TestTagValue:
+            return {
+                "webhooks": [],
+                "delivery_mode": "embed",
+                "screenshot_layout": "desktop",
+                "media_gallery_image_limit": 0,
+                "webhook_text_length_limit": 4000,
+                "save_sent_webhooks": True,
+            }.get(key, default)
+
+        def get_entry_counts(self, **_kwargs: TestKwargValue) -> SimpleNamespace:
+            return SimpleNamespace(total=0)
+
+        def get_entries(self, **_kwargs: TestKwargValue) -> list[Entry]:
+            return []
+
+        def get_feed_counts(self, **_kwargs: TestKwargValue) -> SimpleNamespace:
+            return SimpleNamespace()
+
+    stub = StubReader()
+    app.dependency_overrides[get_reader_dependency] = lambda: stub
+
+    try:
+        with patch("discord_rss_bot.main.create_html_for_feed", return_value=""):
+            response: Response = client.get(url="/feed", params={"feed_url": encoded_url})
+
+        assert response.status_code == 200, f"/feed failed: {response.text}"
+    finally:
+        app.dependency_overrides = {}
+
+
 def test_blacklist_page_uses_live_preview_layout() -> None:
     ensure_preview_feed_exists()
 
-    response: Response = client.get(url="/blacklist", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/blacklist", params={"feed_url": feed_url})
 
     assert response.status_code == 200, f"/blacklist failed: {response.text}"
     assert 'hx-get="/blacklist_preview"' in response.text
@@ -365,7 +450,7 @@ def test_blacklist_page_uses_live_preview_layout() -> None:
 def test_whitelist_page_uses_live_preview_layout() -> None:
     ensure_preview_feed_exists()
 
-    response: Response = client.get(url="/whitelist", params={"feed_url": encoded_feed_url(feed_url)})
+    response: Response = client.get(url="/whitelist", params={"feed_url": feed_url})
 
     assert response.status_code == 200, f"/whitelist failed: {response.text}"
     assert 'hx-get="/whitelist_preview"' in response.text
@@ -919,7 +1004,6 @@ def test_pause_feed() -> None:
     feeds: Response = client.get(url="/")
     if feed_url in feeds.text:
         client.post(url="/remove", data={"feed_url": feed_url})
-        client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)})
 
     # Add the feed.
     response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
@@ -955,7 +1039,6 @@ def test_unpause_feed() -> None:
     feeds: Response = client.get("/")
     if feed_url in feeds.text:
         client.post(url="/remove", data={"feed_url": feed_url})
-        client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)})
 
     # Add the feed.
     response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
@@ -991,7 +1074,6 @@ def test_remove_feed() -> None:
     feeds: Response = client.get(url="/")
     if feed_url in feeds.text:
         client.post(url="/remove", data={"feed_url": feed_url})
-        client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)})
 
     # Add the feed.
     response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
@@ -1317,7 +1399,7 @@ def test_update_feed_not_found() -> None:
     nonexistent_feed_url = "https://nonexistent-feed.example.com/rss.xml"
 
     # Try to update the non-existent feed
-    response: Response = client.get(url="/update", params={"feed_url": urllib.parse.quote(nonexistent_feed_url)})
+    response: Response = client.get(url="/update", params={"feed_url": nonexistent_feed_url})
 
     # Check that it returns a 404 status code
     assert response.status_code == 404, f"Expected 404 for non-existent feed, got: {response.status_code}"
@@ -1376,15 +1458,13 @@ def test_post_entry_send_to_discord() -> None:
     entries: list[Entry] = list(reader.get_entries(feed=feed_url, limit=1))
     assert entries, "Feed should have at least one entry to send"
     entry_to_send: main_module.Entry = entries[0]
-    encoded_id: str = urllib.parse.quote(entry_to_send.id)
-
     no_redirect_client = TestClient(app, follow_redirects=False)
 
     # Patch execute_webhook so no real HTTP requests are made to Discord.
     with patch("discord_rss_bot.feeds.execute_webhook") as mock_execute:
         response = no_redirect_client.get(
             url="/post_entry",
-            params={"entry_id": encoded_id, "feed_url": urllib.parse.quote(feed_url)},
+            params={"entry_id": entry_to_send.id, "feed_url": feed_url},
         )
 
     assert response.status_code == 303, f"Expected redirect after sending, got {response.status_code}: {response.text}"
@@ -1445,7 +1525,7 @@ def test_post_entry_uses_feed_url_to_disambiguate_duplicate_ids() -> None:
         with patch("discord_rss_bot.main.send_entry_to_discord", side_effect=fake_send_entry_to_discord):
             response: Response = no_redirect_client.get(
                 url="/post_entry",
-                params={"entry_id": urllib.parse.quote(shared_id), "feed_url": urllib.parse.quote(feed_b)},
+                params={"entry_id": shared_id, "feed_url": feed_b},
             )
 
         assert response.status_code == 303, f"Expected redirect after sending, got {response.status_code}"
@@ -2075,10 +2155,9 @@ def test_webhook_entries_url_encoding() -> None:
     assert response.status_code == 200, f"Failed to add feed: {response.text}"
 
     # Get webhook_entries with URL-encoded webhook URL
-    encoded_webhook_url = urllib.parse.quote(webhook_url)
     response = client.get(
         url="/webhook_entries",
-        params={"webhook_url": encoded_webhook_url},
+        params={"webhook_url": webhook_url},
     )
 
     assert response.status_code == 200, f"Failed to get /webhook_entries with encoded URL: {response.text}"