Extract admin page helpers from app.main

app/admin.py

@@ -5,8 +5,10 @@
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.ledger.service import create_bounty
 from app.models import WebhookEvent
 
+ADMIN_WEBHOOK_LIMIT_OPTIONS = [10, 25, 50, 100]
 WEBHOOK_OUTCOME_SCAN_ORDER = {
     "missing_submitter": 0,
     "bounty_not_found": 1,
@@ -74,3 +76,47 @@ def webhook_status_summary(session: Session) -> list[dict[str, Any]]:
             str(item["processed_status"]),
         ),
     )
+
+
+def admin_page_context(
+    session: Session,
+    *,
+    login: str,
+    csrf_token: str,
+    webhook_status: str | None,
+    webhook_limit: int,
+) -> dict[str, Any]:
+    normalized_status = normalize_webhook_status_filter(webhook_status) or ""
+    return {
+        "login": login,
+        "csrf_token": csrf_token,
+        "webhook_events": list_webhook_events(session, normalized_status, webhook_limit),
+        "webhook_status_summary": webhook_status_summary(session),
+        "webhook_limit": webhook_limit,
+        "webhook_limit_options": ADMIN_WEBHOOK_LIMIT_OPTIONS,
+        "webhook_status": normalized_status,
+    }
+
+
+def create_admin_bounty_from_form(
+    session: Session,
+    *,
+    repo: str,
+    issue_number: int,
+    issue_url: str,
+    title: str,
+    reward_mrwk: str,
+    max_awards: int,
+    acceptance: str,
+) -> int:
+    bounty = create_bounty(
+        session,
+        repo=repo,
+        issue_number=issue_number,
+        issue_url=issue_url,
+        title=title,
+        reward_mrwk=reward_mrwk,
+        max_awards=max_awards,
+        acceptance=acceptance,
+    )
+    return int(bounty.id)

app/main.py

@@ -9,7 +9,7 @@
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Annotated, Any
-from urllib.parse import urlencode, urlsplit, urlunsplit
+from urllib.parse import unquote, urlencode, urlsplit, urlunsplit
 
 import httpx
 from fastapi import Depends, FastAPI, Form, HTTPException, Query, Request
@@ -21,10 +21,10 @@
 from sqlalchemy.orm import Session
 
 from app.admin import (
+    admin_page_context,
+    create_admin_bounty_from_form,
     list_webhook_events,
-    normalize_webhook_status_filter,
     webhook_events_to_dict,
-    webhook_status_summary,
 )
 from app.config import Settings, get_settings
 from app.db import create_schema, session_scope
@@ -278,13 +278,17 @@ def _oauth_configured(settings: Settings) -> bool:
 
 
 def _safe_next_path(next_path: str | None) -> str:
+    decoded_next_path = unquote(next_path) if next_path else ""
     if (
         not next_path
         or not next_path.startswith("/")
         or next_path.startswith("//")
         or len(next_path) > 2048
         or "\\" in next_path
+        or decoded_next_path.startswith("//")
+        or "\\" in decoded_next_path
         or any(ord(char) < 32 or 127 <= ord(char) < 160 for char in next_path)
+        or any(ord(char) < 32 or 127 <= ord(char) < 160 for char in decoded_next_path)
     ):
         return "/me"
     return next_path
@@ -1407,22 +1411,18 @@ def admin_page(
             if _oauth_configured(settings):
                 return RedirectResponse("/auth/github/login?next=/admin", status_code=302)
             raise HTTPException(status_code=503, detail="GitHub OAuth is not configured")
-        normalized_status = normalize_webhook_status_filter(webhook_status) or ""
         with session_scope(db_url) as session:
-            webhook_events = list_webhook_events(session, normalized_status, webhook_limit)
-            webhook_summary = webhook_status_summary(session)
+            context = admin_page_context(
+                session,
+                login=login,
+                csrf_token=_csrf_token("admin-bounty", login, settings.cookie_secret),
+                webhook_status=webhook_status,
+                webhook_limit=webhook_limit,
+            )
         return templates.TemplateResponse(
             request,
             "admin.html",
-            {
-                "login": login,
-                "csrf_token": _csrf_token("admin-bounty", login, settings.cookie_secret),
-                "webhook_events": webhook_events,
-                "webhook_status_summary": webhook_summary,
-                "webhook_limit": webhook_limit,
-                "webhook_limit_options": [10, 25, 50, 100],
-                "webhook_status": normalized_status,
-            },
+            context,
         )
 
     @app.post("/admin/bounties")
@@ -1448,7 +1448,7 @@ def admin_create_bounty(
             raise HTTPException(status_code=403, detail="invalid CSRF token")
         with session_scope(db_url) as session:
             try:
-                bounty = create_bounty(
+                bounty_id = create_admin_bounty_from_form(
                     session,
                     repo=repo,
                     issue_number=issue_number,
@@ -1460,7 +1460,6 @@ def admin_create_bounty(
                 )
             except LedgerError as exc:
                 raise HTTPException(status_code=400, detail=str(exc)) from exc
-            bounty_id = bounty.id
         return RedirectResponse(f"/bounties/{bounty_id}", status_code=303)
 
     return app

tests/test_admin_helpers.py

@@ -3,13 +3,16 @@
 from datetime import UTC, datetime, timedelta
 
 from app.admin import (
+    ADMIN_WEBHOOK_LIMIT_OPTIONS,
+    admin_page_context,
+    create_admin_bounty_from_form,
     list_webhook_events,
     normalize_webhook_status_filter,
     webhook_events_to_dict,
     webhook_status_summary,
 )
 from app.db import create_schema, session_scope
-from app.models import WebhookEvent
+from app.models import Bounty, WebhookEvent
 
 
 def _event(
@@ -93,3 +96,53 @@ def test_webhook_status_summary_uses_admin_scan_order(sqlite_url: str) -> None:
         {"processed_status": "paid", "count": 2},
         {"processed_status": "custom_status", "count": 3},
     ]
+
+
+def test_admin_page_context_builds_webhook_dashboard_context(sqlite_url: str) -> None:
+    create_schema(sqlite_url)
+    base_time = datetime(2026, 5, 25, 12, 0, tzinfo=UTC)
+    with session_scope(sqlite_url) as session:
+        session.add(_event("delivery-paid", "paid", base_time))
+        session.add(_event("delivery-missing", "Missing_Submitter", base_time + timedelta(hours=1)))
+
+    with session_scope(sqlite_url) as session:
+        context = admin_page_context(
+            session,
+            login="maintainer",
+            csrf_token="csrf-token",
+            webhook_status=" Missing_Submitter ",
+            webhook_limit=10,
+        )
+
+    assert context["login"] == "maintainer"
+    assert context["csrf_token"] == "csrf-token"
+    assert context["webhook_status"] == "missing_submitter"
+    assert context["webhook_limit"] == 10
+    assert context["webhook_limit_options"] == ADMIN_WEBHOOK_LIMIT_OPTIONS
+    assert [event.delivery_id for event in context["webhook_events"]] == ["delivery-missing"]
+    assert context["webhook_status_summary"] == [
+        {"processed_status": "missing_submitter", "count": 1},
+        {"processed_status": "paid", "count": 1},
+    ]
+
+
+def test_create_admin_bounty_from_form_returns_created_bounty_id(sqlite_url: str) -> None:
+    create_schema(sqlite_url)
+
+    with session_scope(sqlite_url) as session:
+        bounty_id = create_admin_bounty_from_form(
+            session,
+            repo="RAmimbo/MergeWork",
+            issue_number=321,
+            issue_url="https://github.com/ramimbo/mergework/issues/321",
+            title="Admin helper bounty",
+            reward_mrwk="25",
+            max_awards=2,
+            acceptance="Admin page form creates this bounty.",
+        )
+        bounty = session.get(Bounty, bounty_id)
+
+    assert bounty is not None
+    assert bounty.repo == "ramimbo/mergework"
+    assert bounty.issue_number == 321
+    assert bounty.max_awards == 2