feat(notifications): add Test button for ntfy and RSS with inline result
- POST /api/notifications/test/ntfy — sends a real push using current form values (not saved settings) so auth can be verified before saving; returns status + HTTP detail on success or error message on failure - POST /api/notifications/test/rss — confirms the feed token exists and returns event count; no bill FK required - NtfyTestRequest + NotificationTestResult schemas added - Frontend: Test button next to Save on both ntfy and RSS sections; result shown inline as a green/red pill; uses current form state for ntfy so the user can test before committing All future notification types should follow the same test-before-save pattern. Authored-By: Jack Levy
This commit is contained in:
@@ -1,9 +1,11 @@
|
||||
"""
|
||||
Notifications API — user notification settings and per-user RSS feed.
|
||||
"""
|
||||
import base64
|
||||
import secrets
|
||||
from xml.etree.ElementTree import Element, SubElement, tostring
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from fastapi.responses import Response
|
||||
from sqlalchemy import select
|
||||
@@ -13,7 +15,12 @@ from app.core.dependencies import get_current_user
|
||||
from app.database import get_db
|
||||
from app.models.notification import NotificationEvent
|
||||
from app.models.user import User
|
||||
from app.schemas.schemas import NotificationSettingsResponse, NotificationSettingsUpdate
|
||||
from app.schemas.schemas import (
|
||||
NotificationSettingsResponse,
|
||||
NotificationSettingsUpdate,
|
||||
NotificationTestResult,
|
||||
NtfyTestRequest,
|
||||
)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@@ -98,6 +105,66 @@ async def reset_rss_token(
|
||||
return _prefs_to_response(user.notification_prefs or {}, user.rss_token)
|
||||
|
||||
|
||||
@router.post("/test/ntfy", response_model=NotificationTestResult)
|
||||
async def test_ntfy(
|
||||
body: NtfyTestRequest,
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Send a test push notification to verify ntfy settings."""
|
||||
url = body.ntfy_topic_url.strip()
|
||||
if not url:
|
||||
return NotificationTestResult(status="error", detail="Topic URL is required")
|
||||
|
||||
headers: dict[str, str] = {
|
||||
"Title": "PocketVeto — Test Notification",
|
||||
"Priority": "default",
|
||||
"Tags": "white_check_mark",
|
||||
}
|
||||
if body.ntfy_auth_method == "token" and body.ntfy_token.strip():
|
||||
headers["Authorization"] = f"Bearer {body.ntfy_token.strip()}"
|
||||
elif body.ntfy_auth_method == "basic" and body.ntfy_username.strip():
|
||||
creds = base64.b64encode(
|
||||
f"{body.ntfy_username.strip()}:{body.ntfy_password}".encode()
|
||||
).decode()
|
||||
headers["Authorization"] = f"Basic {creds}"
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
resp = await client.post(
|
||||
url,
|
||||
content="Your PocketVeto notification settings are working correctly.".encode("utf-8"),
|
||||
headers=headers,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return NotificationTestResult(status="ok", detail=f"Test notification sent (HTTP {resp.status_code})")
|
||||
except httpx.HTTPStatusError as e:
|
||||
return NotificationTestResult(status="error", detail=f"HTTP {e.response.status_code}: {e.response.text[:200]}")
|
||||
except httpx.RequestError as e:
|
||||
return NotificationTestResult(status="error", detail=f"Connection error: {e}")
|
||||
|
||||
|
||||
@router.post("/test/rss", response_model=NotificationTestResult)
|
||||
async def test_rss(
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Verify the user's RSS feed is reachable and return its event count."""
|
||||
user = await db.get(User, current_user.id)
|
||||
if not user.rss_token:
|
||||
return NotificationTestResult(status="error", detail="RSS token not generated — save settings first")
|
||||
|
||||
count_result = await db.execute(
|
||||
select(NotificationEvent).where(NotificationEvent.user_id == user.id)
|
||||
)
|
||||
event_count = len(count_result.scalars().all())
|
||||
|
||||
return NotificationTestResult(
|
||||
status="ok",
|
||||
detail=f"RSS feed is active with {event_count} event{'s' if event_count != 1 else ''}. Subscribe to the URL shown above.",
|
||||
event_count=event_count,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/feed/{rss_token}.xml", include_in_schema=False)
|
||||
async def rss_feed(rss_token: str, db: AsyncSession = Depends(get_db)):
|
||||
"""Public tokenized RSS feed — no auth required."""
|
||||
|
||||
@@ -28,6 +28,20 @@ class NotificationSettingsUpdate(BaseModel):
|
||||
ntfy_enabled: Optional[bool] = None
|
||||
rss_enabled: Optional[bool] = None
|
||||
|
||||
|
||||
class NtfyTestRequest(BaseModel):
|
||||
ntfy_topic_url: str
|
||||
ntfy_auth_method: str = "none"
|
||||
ntfy_token: str = ""
|
||||
ntfy_username: str = ""
|
||||
ntfy_password: str = ""
|
||||
|
||||
|
||||
class NotificationTestResult(BaseModel):
|
||||
status: str # "ok" | "error"
|
||||
detail: str
|
||||
event_count: Optional[int] = None # RSS only
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user