Add webhook management to feed detail page and corresponding tests

discord_rss_bot/main.py

@@ -1112,6 +1112,14 @@ async def get_feed(  # noqa: C901, PLR0912, PLR0914, PLR0915
     except FeedNotFoundError as e:
         raise HTTPException(status_code=404, detail=f"Feed '{clean_feed_url}' not found.\n\n{e}") from e
 
+    webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
+    current_webhook_url: str = str(reader.get_tag(feed.url, "webhook", "")).strip()
+    current_webhook_name: str = ""
+    for hook in webhooks:
+        if hook.get("url", "").strip() == current_webhook_url:
+            current_webhook_name = hook.get("name", "").strip()
+            break
+
     # Only show button if more than 10 entries.
     total_entries: int = reader.get_entry_counts(feed=feed).total or 0
     is_show_more_entries_button_visible: bool = total_entries > entries_per_page
@@ -1157,6 +1165,9 @@ async def get_feed(  # noqa: C901, PLR0912, PLR0914, PLR0915
                 "total_entries": total_entries,
                 "feed_interval": feed_interval,
                 "global_interval": global_interval,
+                "webhooks": webhooks,
+                "current_webhook_url": current_webhook_url,
+                "current_webhook_name": current_webhook_name,
             }
             return templates.TemplateResponse(request=request, name="feed.html", context=context)
 
@@ -1213,6 +1224,9 @@ async def get_feed(  # noqa: C901, PLR0912, PLR0914, PLR0915
         "total_entries": total_entries,
         "feed_interval": feed_interval,
         "global_interval": global_interval,
+        "webhooks": webhooks,
+        "current_webhook_url": current_webhook_url,
+        "current_webhook_name": current_webhook_name,
     }
     return templates.TemplateResponse(request=request, name="feed.html", context=context)
 

discord_rss_bot/templates/feed.html

@@ -171,6 +171,47 @@
                             </div>
                         </form>
                     </section>
+                    <section class="mt-4 pt-3 border-top border-secondary-subtle">
+                        <h3 class="h6 text-uppercase text-muted mb-3">Webhook</h3>
+                        {% if current_webhook_name %}
+                            <p class="text-muted mb-3">
+                                Current webhook:
+                                <strong>{{ current_webhook_name }}</strong>
+                            </p>
+                        {% elif current_webhook_url %}
+                            <p class="text-warning mb-3">This feed references a missing webhook. Choose a webhook below to reattach it.</p>
+                        {% else %}
+                            <p class="text-warning mb-3">No webhook is attached to this feed yet.</p>
+                        {% endif %}
+                        {% if webhooks %}
+                            <form action="/attach_feed_webhook"
+                                  method="post"
+                                  class="d-flex flex-wrap align-items-center gap-2 mb-0">
+                                <input type="hidden" name="feed_url" value="{{ feed.url }}" />
+                                <input type="hidden"
+                                       name="redirect_to"
+                                       value="/feed?feed_url={{ feed.url|encode_url }}" />
+                                <select name="webhook_dropdown"
+                                        class="form-select form-select-sm bg-dark border-dark text-muted"
+                                        required>
+                                    <option value=""
+                                            disabled
+                                            {% if not current_webhook_name %}selected{% endif %}>
+                                        Select webhook...
+                                    </option>
+                                    {% for hook in webhooks %}
+                                        <option value="{{ hook.name }}"
+                                                {% if hook.name == current_webhook_name %}selected{% endif %}>
+                                            {{ hook.name }}
+                                        </option>
+                                    {% endfor %}
+                                </select>
+                                <button class="btn btn-outline-light btn-sm" type="submit">Save webhook</button>
+                            </form>
+                        {% else %}
+                            <p class="text-muted mb-0">Add a webhook first to attach this feed.</p>
+                        {% endif %}
+                    </section>
                     <section class="mt-4 pt-3 border-top border-secondary-subtle">
                         <h3 class="h6 text-uppercase text-muted mb-3">Feed Information</h3>
                         <div class="row g-2 text-muted small">

tests/test_main.py

@@ -613,6 +613,60 @@ def test_attach_feed_webhook_from_index() -> None:
     client.post(url="/delete_webhook", data={"webhook_url": replacement_webhook_url})
 
 
+def test_attach_feed_webhook_from_feed_page() -> None:
+    """Feed detail page should allow attaching/replacing webhook directly."""
+    original_webhook_name = "feed-page-original-webhook"
+    original_webhook_url = "https://discord.com/api/webhooks/333/original"
+    replacement_webhook_name = "feed-page-replacement-webhook"
+    replacement_webhook_url = "https://discord.com/api/webhooks/444/replacement"
+
+    # Start clean.
+    client.post(url="/remove", data={"feed_url": feed_url})
+    client.post(url="/delete_webhook", data={"webhook_url": original_webhook_url})
+    client.post(url="/delete_webhook", data={"webhook_url": replacement_webhook_url})
+
+    # Create two webhooks and attach feed to original.
+    response = client.post(
+        url="/add_webhook",
+        data={"webhook_name": original_webhook_name, "webhook_url": original_webhook_url},
+    )
+    assert response.status_code == 200, f"Failed to add original webhook: {response.text}"
+
+    response = client.post(
+        url="/add_webhook",
+        data={"webhook_name": replacement_webhook_name, "webhook_url": replacement_webhook_url},
+    )
+    assert response.status_code == 200, f"Failed to add replacement webhook: {response.text}"
+
+    response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": original_webhook_name})
+    assert response.status_code == 200, f"Failed to add feed: {response.text}"
+
+    # Feed page should show the webhook form and current webhook label.
+    response = client.get(url="/feed", params={"feed_url": feed_url})
+    assert response.status_code == 200, f"Failed to get /feed: {response.text}"
+    assert "Current webhook:" in response.text
+    assert "/attach_feed_webhook" in response.text
+
+    # Reattach to replacement webhook via endpoint used by feed page form.
+    response = client.post(
+        url="/attach_feed_webhook",
+        data={
+            "feed_url": feed_url,
+            "webhook_dropdown": replacement_webhook_name,
+            "redirect_to": f"/feed?feed_url={urllib.parse.quote(feed_url)}",
+        },
+    )
+    assert response.status_code == 200, f"Failed to reattach feed webhook: {response.text}"
+
+    reader = get_reader_dependency()
+    assert reader.get_tag(feed_url, "webhook", "") == replacement_webhook_url
+
+    # Cleanup.
+    client.post(url="/remove", data={"feed_url": feed_url})
+    client.post(url="/delete_webhook", data={"webhook_url": original_webhook_url})
+    client.post(url="/delete_webhook", data={"webhook_url": replacement_webhook_url})
+
+
 def test_update_feed_not_found() -> None:
     """Test updating a non-existent feed."""
     # Generate a feed URL that does not exist