Add ntfy push notifications with action buttons for meal reminders
CI / test-and-package (push) Successful in 37s

- ntfy.py: send_meal_reminder() with Home/Out HTTP action buttons, test helper
- models.py: new columns ntfy_url, ntfy_token, callback_token on users table
- meals.py: /api/ntfy-callback (public, token-authenticated), /ntfy-settings, /ntfy-test
- app.py: flask send-reminders CLI command (cron-schedulable)
- Dashboard: ntfy config form in Account Settings
- i18n: translations for all ntfy strings (en/fr/de)
- Dockerfile: include ntfy.py in COPY
- tests: 15 new tests covering callback, settings, and auth
- docs: updated AGENTS.md and doc/index.md
This commit is contained in:
agentbox
2026-08-01 06:17:26 +00:00
parent 7bf00ee2e3
commit 3681f9dfce
10 changed files with 650 additions and 4 deletions
+4 -2
View File
@@ -14,9 +14,10 @@ Flask web app for tracking household meal attendance. Users register into househ
|---|---| |---|---|
| `app.py` | Flask factory (`create_app()`), registers blueprints and i18n | | `app.py` | Flask factory (`create_app()`), registers blueprints and i18n |
| `auth.py` | Auth blueprint: `/login`, `/register`, `/logout`. Session-based. | | `auth.py` | Auth blueprint: `/login`, `/register`, `/logout`. Session-based. |
| `meals.py` | Meals blueprint: `/dashboard`, `/respond`, `/history`, `/delete-account`, `/admin/*` | | `meals.py` | Meals blueprint: `/dashboard`, `/respond`, `/history`, `/delete-account`, `/admin/*`, `/ntfy-settings`, `/ntfy-test`, `/api/ntfy-callback` |
| `models.py` | All DB access. SQLite via `sqlite3`. 4 tables: `households`, `users`, `meal_periods`, `responses`. | | `models.py` | All DB access. SQLite via `sqlite3`. 4 tables: `households`, `users`, `meal_periods`, `responses`. Ntfy helpers in `# Ntfy / notifications` section. |
| `i18n.py` | Translation module. `t(key)` helper for en/fr/de. Detects `Accept-Language`, session override via `?lang=`. All user-visible strings flow through `t()`. | | `i18n.py` | Translation module. `t(key)` helper for en/fr/de. Detects `Accept-Language`, session override via `?lang=`. All user-visible strings flow through `t()`. |
| `ntfy.py` | Ntfy notification helpers. `send_meal_reminder()` sends push with Home/Out HTTP action buttons. `send_test_notification()` verifies config. |
| `templates/` | Jinja2: `base.html` (nav, lang footer), `login.html`, `register.html`, `dashboard.html` | | `templates/` | Jinja2: `base.html` (nav, lang footer), `login.html`, `register.html`, `dashboard.html` |
| `static/style.css` | All CSS (no framework) | | `static/style.css` | All CSS (no framework) |
| `tests/` | Pytest suite (47 tests): `conftest.py` sets up in-memory DB per test | | `tests/` | Pytest suite (47 tests): `conftest.py` sets up in-memory DB per test |
@@ -33,6 +34,7 @@ Flask web app for tracking household meal attendance. Users register into househ
- **i18n:** Every user-visible string in templates uses `{{ t('English text') }}`. In Python, `flash(t("message"), "category")`. Translations live in `TRANSLATIONS` dict in `i18n.py`. English string is the key; only non-English translations stored. When adding new user-facing text, add it to `TRANSLATIONS` for fr/de. - **i18n:** Every user-visible string in templates uses `{{ t('English text') }}`. In Python, `flash(t("message"), "category")`. Translations live in `TRANSLATIONS` dict in `i18n.py`. English string is the key; only non-English translations stored. When adding new user-facing text, add it to `TRANSLATIONS` for fr/de.
- **DB:** All access through `get_db()` context manager in `models.py`. Returns `sqlite3.Row` objects accessed as dicts. Path via `DB_PATH` env var (defaults to `/data/meals.db` in Docker, `./meals.db` locally). - **DB:** All access through `get_db()` context manager in `models.py`. Returns `sqlite3.Row` objects accessed as dicts. Path via `DB_PATH` env var (defaults to `/data/meals.db` in Docker, `./meals.db` locally).
- **CSS:** No framework. Variables in `:root`. Use existing utility classes (`.btn`, `.chip`, `.badge`, etc.). - **CSS:** No framework. Variables in `:root`. Use existing utility classes (`.btn`, `.chip`, `.badge`, etc.).
- **Ntfy:** Per-user topic URL + Bearer token stored in `users` table. Callback token auto-generated on first ntfy setup. `/api/ntfy-callback` is unauthenticated (uses callback token). Send reminders via `flask send-reminders [--date] [--meal] [--base-url]`. Schedule with cron.
## Common Pitfalls ## Common Pitfalls
+1 -1
View File
@@ -7,7 +7,7 @@ COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt RUN pip install --no-cache-dir -r requirements.txt
# Copy application # Copy application
COPY app.py models.py auth.py meals.py i18n.py ./ COPY app.py models.py auth.py meals.py i18n.py ntfy.py ./
COPY templates/ templates/ COPY templates/ templates/
COPY static/ static/ COPY static/ static/
+37 -1
View File
@@ -1,13 +1,16 @@
"""Meal Tracker — Flask application entry point.""" """Meal Tracker — Flask application entry point."""
import os import os
from datetime import date
import click
from flask import Flask from flask import Flask
from models import init_db from models import init_db, get_users_with_ntfy, get_all_households
from auth import auth_bp from auth import auth_bp
from meals import meals_bp from meals import meals_bp
from i18n import init_i18n from i18n import init_i18n
from ntfy import send_meal_reminder
def create_app() -> Flask: def create_app() -> Flask:
@@ -22,6 +25,39 @@ def create_app() -> Flask:
init_i18n(app) init_i18n(app)
# ── CLI commands ──
@app.cli.command("send-reminders")
@click.option("--date", default=None, help="Date to send reminders for (YYYY-MM-DD, default: today)")
@click.option("--meal", default="both", type=click.Choice(["lunch", "dinner", "both"]),
help="Which meal to send reminders for (default: both)")
@click.option("--base-url", default=None, help="Base URL of the app (default: $APP_BASE_URL or http://localhost:5000)")
def send_reminders_command(date, meal, base_url):
"""Send ntfy meal reminders to all users who have ntfy configured."""
target_date = date and __import__("datetime").datetime.strptime(date, "%Y-%m-%d").date() or date.today()
app_base_url = base_url or os.environ.get("APP_BASE_URL", "http://localhost:5000")
meal_types = ["lunch", "dinner"] if meal == "both" else [meal]
total_sent = 0
for household in get_all_households():
for user in get_users_with_ntfy(household["id"]):
for mt in meal_types:
ok = send_meal_reminder(
ntfy_url=user["ntfy_url"],
ntfy_token=user["ntfy_token"],
callback_token=user["callback_token"],
meal_type=mt,
for_date=target_date,
app_base_url=app_base_url,
)
if ok:
total_sent += 1
click.echo(f"{user['username']} ({mt})")
else:
click.echo(f"{user['username']} ({mt}) — failed")
click.echo(f"\nSent {total_sent} reminder(s) for {target_date.isoformat()}.")
return app return app
+53
View File
@@ -32,6 +32,8 @@ A lightweight Flask web application for tracking household meal attendance. Memb
├── auth.py # Authentication blueprint (login, register, logout) ├── auth.py # Authentication blueprint (login, register, logout)
├── meals.py # Meals blueprint (dashboard, responses, history, admin) ├── meals.py # Meals blueprint (dashboard, responses, history, admin)
├── models.py # Database models and helpers ├── models.py # Database models and helpers
├── i18n.py # Internationalization (en/fr/de)
├── ntfy.py # Ntfy notification helpers
├── requirements.txt # Python dependencies ├── requirements.txt # Python dependencies
├── Dockerfile # Container image definition ├── Dockerfile # Container image definition
├── docker-compose.yml # Container orchestration ├── docker-compose.yml # Container orchestration
@@ -118,6 +120,19 @@ source .env
## API / Routes ## API / Routes
### Ntfy Callback (public)
| Method | Path | Description |
|--------|------|-------------|
| POST | `/api/ntfy-callback` | Receive ntfy action callbacks. Auth via per-user `callback_token`. Body: `{"token":"...", "meal_type":"lunch|dinner", "date":"YYYY-MM-DD", "status":"yes|no"}`. Returns 200 on success, 403 for bad token, 400 for invalid/past data. |
### Ntfy Settings (authenticated)
| Method | Path | Description |
|--------|------|-------------|
| POST | `/ntfy-settings` | Save ntfy topic URL and access token |
| POST | `/ntfy-test` | Send a test notification to verify configuration |
### Authentication ### Authentication
| Method | Path | Description | | Method | Path | Description |
@@ -189,3 +204,41 @@ pytest tests/ -v
``` ```
Tests use a temporary in-memory SQLite database (configured per test via `conftest.py`). Tests use a temporary in-memory SQLite database (configured per test via `conftest.py`).
## Ntfy Notifications
Users can configure [ntfy](https://ntfy.sh) to receive push notifications with action buttons, allowing them to respond to meals directly from their phone without opening the web UI.
### Setup
1. Install the ntfy app on your phone and subscribe to a topic (e.g. `your-name-meals`)
2. (Optional) Create an access token for authenticated publishing
3. In the web UI, open **Account Settings** and enter:
- **Ntfy Topic URL**: Your topic (e.g. `https://ntfy.sh/your-name-meals`)
- **Access Token**: Your ntfy access token (if using auth)
4. Click **Send Test** to verify it works
### How It Works
- Reminders are sent via the CLI command: `flask send-reminders`
- The notification contains **Home** and **Out** buttons
- Pressing a button sends an HTTP callback to the app, recording your response instantly
- Tapping the notification itself opens the dashboard
### Scheduling Reminders
Add a cron job on the server to send reminders automatically:
```bash
# Lunch reminder at 9:00 AM
0 9 * * * cd /path/to/app && docker compose exec meal-tracker flask send-reminders --meal lunch --base-url https://meals.ct.cozytren.ch
# Dinner reminder at 5:00 PM
0 17 * * * cd /path/to/app && docker compose exec meal-tracker flask send-reminders --meal dinner --base-url https://meals.ct.cozytren.ch
```
Or run manually:
```bash
docker compose exec meal-tracker flask send-reminders --date 2026-08-01 --meal both
```
+46
View File
@@ -320,6 +320,52 @@ TRANSLATIONS: dict[str, dict[str, str]] = {
"fr": "Le foyer et tous ses membres ont été supprimés.", "fr": "Le foyer et tous ses membres ont été supprimés.",
"de": "Haushalt und alle Mitglieder wurden gelöscht.", "de": "Haushalt und alle Mitglieder wurden gelöscht.",
}, },
# ── Ntfy notifications ──
"Ntfy Notifications": {
"fr": "Notifications Ntfy",
"de": "Ntfy-Benachrichtigungen",
},
"Get meal reminders on your phone via ntfy.": {
"fr": "Recevez des rappels de repas sur votre téléphone via ntfy.",
"de": "Erhalten Sie Essenserinnerungen auf Ihrem Handy via ntfy.",
},
"Ntfy Topic URL": {
"fr": "URL du sujet Ntfy",
"de": "Ntfy-Themen-URL",
},
"Access Token": {
"fr": "Jeton d'accès",
"de": "Zugriffstoken",
},
"(optional)": {
"fr": "(optionnel)",
"de": "(optional)",
},
"Save": {
"fr": "Enregistrer",
"de": "Speichern",
},
"Send Test": {
"fr": "Tester",
"de": "Test senden",
},
"Ntfy settings saved.": {
"fr": "Paramètres Ntfy enregistrés.",
"de": "Ntfy-Einstellungen gespeichert.",
},
"Please enter an ntfy URL first.": {
"fr": "Veuillez d'abord entrer une URL ntfy.",
"de": "Bitte geben Sie zuerst eine ntfy-URL ein.",
},
"Test notification sent! Check your device.": {
"fr": "Notification de test envoyée ! Vérifiez votre appareil.",
"de": "Testbenachrichtigung gesendet! Überprüfen Sie Ihr Gerät.",
},
"Failed to send test notification. Check your ntfy URL and token.": {
"fr": "Échec de l'envoi du test. Vérifiez votre URL ntfy et votre jeton.",
"de": "Testbenachrichtigung fehlgeschlagen. Überprüfen Sie Ihre ntfy-URL und Ihr Token.",
},
} }
+97
View File
@@ -22,8 +22,12 @@ from models import (
delete_user, delete_user,
delete_household, delete_household,
count_household_users, count_household_users,
set_user_ntfy,
get_user_ntfy,
get_user_by_callback_token,
) )
from i18n import t from i18n import t
from ntfy import send_test_notification
meals_bp = Blueprint("meals", __name__) meals_bp = Blueprint("meals", __name__)
@@ -68,6 +72,7 @@ def dashboard():
return redirect(url_for("auth.login")) return redirect(url_for("auth.login"))
data = get_dashboard_data(today, household_id) data = get_dashboard_data(today, household_id)
ntfy = get_user_ntfy(session["user_id"])
return render_template( return render_template(
"dashboard.html", "dashboard.html",
dashboard=data, dashboard=data,
@@ -77,6 +82,7 @@ def dashboard():
next_date=today + timedelta(days=1), next_date=today + timedelta(days=1),
current_user_id=session["user_id"], current_user_id=session["user_id"],
is_admin=session["is_admin"], is_admin=session["is_admin"],
ntfy=ntfy,
) )
@@ -125,6 +131,7 @@ def history():
target_date = date.today() target_date = date.today()
data = get_dashboard_data(target_date, session["household_id"]) data = get_dashboard_data(target_date, session["household_id"])
ntfy = get_user_ntfy(session["user_id"])
return render_template( return render_template(
"dashboard.html", "dashboard.html",
dashboard=data, dashboard=data,
@@ -135,6 +142,7 @@ def history():
current_user_id=session["user_id"], current_user_id=session["user_id"],
is_admin=session["is_admin"], is_admin=session["is_admin"],
history_mode=True, history_mode=True,
ntfy=ntfy,
) )
@@ -211,3 +219,92 @@ def admin_delete_household():
for key in ("user_id", "username", "is_admin", "household_id", "household_name"): for key in ("user_id", "username", "is_admin", "household_id", "household_name"):
session.pop(key, None) session.pop(key, None)
return redirect(url_for("auth.login")) return redirect(url_for("auth.login"))
# ── Ntfy integration ─────────────────────────────────────────────────────────
@meals_bp.route("/ntfy-settings", methods=["POST"])
@login_required
def ntfy_settings():
"""Save ntfy URL and token for the current user."""
import secrets
ntfy_url = request.form.get("ntfy_url", "").strip()
ntfy_token = request.form.get("ntfy_token", "").strip()
existing = get_user_ntfy(session["user_id"])
callback_token = existing.get("callback_token") if existing else ""
# Generate a new callback token if user is enabling ntfy for the first time
if ntfy_url and not callback_token:
callback_token = secrets.token_urlsafe(32)
# If clearing ntfy, also clear the callback token
if not ntfy_url:
callback_token = ""
ntfy_token = ""
set_user_ntfy(session["user_id"], ntfy_url, ntfy_token, callback_token)
flash(t("Ntfy settings saved."), "success")
return redirect(url_for("meals.dashboard"))
@meals_bp.route("/ntfy-test", methods=["POST"])
@login_required
def ntfy_test():
"""Send a test ntfy notification to verify the configuration."""
ntfy_url = request.form.get("ntfy_url", "").strip()
ntfy_token = request.form.get("ntfy_token", "").strip()
if not ntfy_url:
flash(t("Please enter an ntfy URL first."), "error")
return redirect(url_for("meals.dashboard"))
ok = send_test_notification(ntfy_url, ntfy_token)
if ok:
flash(t("Test notification sent! Check your device."), "success")
else:
flash(t("Failed to send test notification. Check your ntfy URL and token."), "error")
return redirect(url_for("meals.dashboard"))
@meals_bp.route("/api/ntfy-callback", methods=["POST"])
def ntfy_callback():
"""Receive ntfy action callbacks (no login — authenticated via callback_token).
Expects JSON body: {"token": "...", "meal_type": "lunch|dinner",
"date": "YYYY-MM-DD", "status": "yes|no"}
"""
data = request.get_json(silent=True)
if not data:
return {"error": "Invalid JSON"}, 400
token = data.get("token", "")
meal_type = data.get("meal_type", "")
date_str = data.get("date", "")
status = data.get("status", "")
user = get_user_by_callback_token(token)
if user is None:
return {"error": "Invalid token"}, 403
if meal_type not in ("lunch", "dinner"):
return {"error": "Invalid meal_type"}, 400
if status not in ("yes", "no"):
return {"error": "Invalid status"}, 400
try:
target_date = datetime.strptime(date_str, "%Y-%m-%d").date()
except (ValueError, TypeError):
return {"error": "Invalid date"}, 400
if target_date < date.today():
return {"error": "Cannot change past meals"}, 400
periods = ensure_meal_periods(target_date)
period = next((p for p in periods if p["meal_type"] == meal_type), None)
if period is None:
return {"error": "Meal period not found"}, 400
upsert_response(user["id"], period["id"], status)
return {"message": "Response recorded", "status": status}, 200
+52
View File
@@ -69,6 +69,12 @@ def init_db() -> None:
if "household_id" not in cols: if "household_id" not in cols:
db.execute("ALTER TABLE users ADD COLUMN household_id INTEGER REFERENCES households(id)") db.execute("ALTER TABLE users ADD COLUMN household_id INTEGER REFERENCES households(id)")
db.execute("CREATE INDEX IF NOT EXISTS idx_users_household ON users(household_id)") db.execute("CREATE INDEX IF NOT EXISTS idx_users_household ON users(household_id)")
if "ntfy_url" not in cols:
db.execute("ALTER TABLE users ADD COLUMN ntfy_url TEXT DEFAULT ''")
if "ntfy_token" not in cols:
db.execute("ALTER TABLE users ADD COLUMN ntfy_token TEXT DEFAULT ''")
if "callback_token" not in cols:
db.execute("ALTER TABLE users ADD COLUMN callback_token TEXT DEFAULT ''")
# ── Households ──────────────────────────────────────────────────────────────── # ── Households ────────────────────────────────────────────────────────────────
@@ -174,6 +180,52 @@ def get_household_users(household_id: int) -> list[dict]:
return [dict(r) for r in rows] return [dict(r) for r in rows]
# ── Ntfy / notifications ─────────────────────────────────────────────────────
def set_user_ntfy(user_id: int, ntfy_url: str, ntfy_token: str,
callback_token: str) -> None:
"""Save ntfy settings for a user."""
with get_db() as db:
db.execute(
"UPDATE users SET ntfy_url = ?, ntfy_token = ?, callback_token = ? "
"WHERE id = ?",
(ntfy_url, ntfy_token, callback_token, user_id),
)
def get_user_ntfy(user_id: int) -> dict:
"""Get ntfy settings for a user."""
with get_db() as db:
row = db.execute(
"SELECT ntfy_url, ntfy_token, callback_token FROM users WHERE id = ?",
(user_id,),
).fetchone()
return dict(row) if row else {}
def get_user_by_callback_token(token: str) -> dict | None:
"""Look up a user by their ntfy callback token."""
with get_db() as db:
row = db.execute(
"SELECT id, username, household_id FROM users WHERE callback_token = ?",
(token,),
).fetchone()
return dict(row) if row else None
def get_users_with_ntfy(household_id: int) -> list[dict]:
"""Return all users in a household who have ntfy configured."""
with get_db() as db:
rows = db.execute(
"SELECT id, username, ntfy_url, ntfy_token, callback_token "
"FROM users WHERE household_id = ? "
"AND ntfy_url != '' AND callback_token != '' "
"ORDER BY username",
(household_id,),
).fetchall()
return [dict(r) for r in rows]
# ── Meal periods ────────────────────────────────────────────────────────────── # ── Meal periods ──────────────────────────────────────────────────────────────
def ensure_meal_periods(for_date: date) -> list[dict]: def ensure_meal_periods(for_date: date) -> list[dict]:
+112
View File
@@ -0,0 +1,112 @@
"""Ntfy notification helpers — send meal reminders with action buttons."""
from __future__ import annotations
import json
import urllib.request
import urllib.error
from datetime import date
def _ntfy_headers(token: str) -> dict[str, str]:
"""Build Authorization header if token is provided."""
headers = {"Content-Type": "application/json"}
if token.strip():
headers["Authorization"] = f"Bearer {token.strip()}"
return headers
def send_meal_reminder(
ntfy_url: str,
ntfy_token: str,
callback_token: str,
meal_type: str,
for_date: date,
app_base_url: str,
) -> bool:
"""Send a meal reminder notification with Home/Out action buttons.
Args:
ntfy_url: Full topic URL (e.g. https://ntfy.sh/mytopic).
ntfy_token: Bearer token for auth (empty for public topics).
callback_token: Per-user token for the webhook callback.
meal_type: 'lunch' or 'dinner'.
for_date: The meal date.
app_base_url: Base URL of this app (e.g. https://meals.ct.cozytren.ch).
Returns:
True if the notification was sent successfully.
"""
date_str = for_date.isoformat()
meal_label = "Lunch" if meal_type == "lunch" else "Dinner"
emoji = "🌤" if meal_type == "lunch" else "🌙"
callback_url = f"{app_base_url.rstrip('/')}/api/ntfy-callback"
payload = {
"topic": ntfy_url.rstrip("/").rsplit("/", 1)[-1],
"title": f"{emoji} {meal_label} — Meal Reminder",
"message": f"Will you be home for {meal_label.lower()} on {date_str}?",
"priority": 4,
"tags": ["plate_with_cutlery"],
"actions": [
{
"action": "http",
"label": "🏠 Home",
"method": "POST",
"url": callback_url,
"headers": {"Content-Type": "application/json"},
"body": json.dumps({
"token": callback_token,
"meal_type": meal_type,
"date": date_str,
"status": "yes",
}),
},
{
"action": "http",
"label": "🍽 Out",
"method": "POST",
"url": callback_url,
"headers": {"Content-Type": "application/json"},
"body": json.dumps({
"token": callback_token,
"meal_type": meal_type,
"date": date_str,
"status": "no",
}),
},
],
"click": f"{app_base_url.rstrip('/')}/dashboard?date={date_str}",
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(ntfy_url, data=data, headers=_ntfy_headers(ntfy_token))
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return 200 <= resp.status < 300
except urllib.error.URLError as e:
print(f"[ntfy] Failed to send to {ntfy_url}: {e}")
return False
def send_test_notification(ntfy_url: str, ntfy_token: str) -> bool:
"""Send a simple test notification to verify ntfy configuration."""
payload = {
"topic": ntfy_url.rstrip("/").rsplit("/", 1)[-1],
"title": "✅ Meal Tracker — Test",
"message": "Your ntfy integration is working! You'll receive meal reminders here.",
"priority": 3,
"tags": ["white_check_mark"],
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(ntfy_url, data=data, headers=_ntfy_headers(ntfy_token))
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return 200 <= resp.status < 300
except urllib.error.URLError as e:
print(f"[ntfy] Test failed for {ntfy_url}: {e}")
return False
+17
View File
@@ -96,6 +96,23 @@
<button class="btn btn-danger">{{ t('Delete My Account') }}</button> <button class="btn btn-danger">{{ t('Delete My Account') }}</button>
</form> </form>
<form method="POST" action="{{ url_for('meals.ntfy_settings') }}" class="manage-form">
<h4>🔔 {{ t('Ntfy Notifications') }}</h4>
<p class="hint">{{ t('Get meal reminders on your phone via ntfy.') }}</p>
<label for="ntfy_url">{{ t('Ntfy Topic URL') }}</label>
<input type="text" id="ntfy_url" name="ntfy_url"
placeholder="https://ntfy.sh/your-topic"
value="{{ ntfy.ntfy_url or '' }}">
<label for="ntfy_token">{{ t('Access Token') }} <span class="hint">{{ t('(optional)') }}</span></label>
<input type="password" id="ntfy_token" name="ntfy_token"
placeholder="tk_..."
value="{{ ntfy.ntfy_token or '' }}">
<div class="btn-row" style="margin-top: 8px;">
<button type="submit" class="btn btn-primary btn-sm">{{ t('Save') }}</button>
<button type="submit" class="btn btn-sm" formaction="{{ url_for('meals.ntfy_test') }}">{{ t('Send Test') }}</button>
</div>
</form>
{% if is_admin %} {% if is_admin %}
<form method="POST" action="{{ url_for('meals.admin_delete_household') }}" <form method="POST" action="{{ url_for('meals.admin_delete_household') }}"
onsubmit="return confirm(this.dataset.confirmMsg);" onsubmit="return confirm(this.dataset.confirmMsg);"
+231
View File
@@ -0,0 +1,231 @@
"""Tests for ntfy integration — callback endpoint, settings, and CLI."""
import json
import secrets
from datetime import date, timedelta
from models import get_db, set_user_ntfy, get_user_ntfy
def _setup_ntfy_user(client, username="ntfyuser", password="testpass",
ntfy_url="https://ntfy.sh/test-topic", ntfy_token="tk_test123"):
"""Register a user and save ntfy settings. Returns callback_token."""
# Admin creates household and registers
client.post("/register", data={
"username": "admin2",
"password": "admin2pass",
"confirm": "admin2pass",
"household_action": "create",
"new_household_name": "Ntfy Household",
}, follow_redirects=True)
client.get("/logout")
from models import get_household_by_name
hh = get_household_by_name("Ntfy Household")
client.post("/register", data={
"username": username,
"password": password,
"confirm": password,
"household_action": "join",
"household_id": str(hh["id"]),
}, follow_redirects=True)
# User is now logged in
# Save ntfy settings
db = get_db()
import secrets as sec
token = sec.token_urlsafe(32)
row = db.execute("SELECT id FROM users WHERE username = ?", (username,)).fetchone()
set_user_ntfy(row["id"], ntfy_url, ntfy_token, token)
return token
class TestNtfyCallback:
def test_valid_callback_records_response(self, client):
token = _setup_ntfy_user(client)
today = date.today().isoformat()
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "lunch",
"date": today, "status": "yes"}),
content_type="application/json")
assert resp.status_code == 200
data = resp.get_json()
assert data["message"] == "Response recorded"
assert data["status"] == "yes"
def test_callback_records_no_status(self, client):
token = _setup_ntfy_user(client)
today = date.today().isoformat()
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "dinner",
"date": today, "status": "no"}),
content_type="application/json")
assert resp.status_code == 200
assert resp.get_json()["status"] == "no"
def test_callback_changes_mind(self, client):
token = _setup_ntfy_user(client)
today = date.today().isoformat()
# First response
client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "lunch",
"date": today, "status": "yes"}),
content_type="application/json")
# Change mind
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "lunch",
"date": today, "status": "no"}),
content_type="application/json")
assert resp.status_code == 200
assert resp.get_json()["status"] == "no"
def test_invalid_token_returns_403(self, client):
today = date.today().isoformat()
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": "not-a-valid-token",
"meal_type": "lunch",
"date": today, "status": "yes"}),
content_type="application/json")
assert resp.status_code == 403
assert resp.get_json()["error"] == "Invalid token"
def test_past_date_returns_400(self, client):
token = _setup_ntfy_user(client)
yesterday = (date.today() - timedelta(days=1)).isoformat()
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "lunch",
"date": yesterday, "status": "yes"}),
content_type="application/json")
assert resp.status_code == 400
assert "past" in resp.get_json()["error"].lower()
def test_invalid_meal_type_returns_400(self, client):
token = _setup_ntfy_user(client)
today = date.today().isoformat()
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "breakfast",
"date": today, "status": "yes"}),
content_type="application/json")
assert resp.status_code == 400
assert "meal_type" in resp.get_json()["error"].lower()
def test_invalid_status_returns_400(self, client):
token = _setup_ntfy_user(client)
today = date.today().isoformat()
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "lunch",
"date": today, "status": "maybe"}),
content_type="application/json")
assert resp.status_code == 400
assert "status" in resp.get_json()["error"].lower()
def test_invalid_date_format_returns_400(self, client):
token = _setup_ntfy_user(client)
resp = client.post("/api/ntfy-callback",
data=json.dumps({"token": token, "meal_type": "lunch",
"date": "not-a-date", "status": "yes"}),
content_type="application/json")
assert resp.status_code == 400
def test_missing_json_returns_400(self, client):
resp = client.post("/api/ntfy-callback",
data="not json",
content_type="text/plain")
assert resp.status_code == 400
def test_empty_body_returns_400(self, client):
resp = client.post("/api/ntfy-callback",
data=None,
content_type="application/json")
assert resp.status_code == 400
class TestNtfySettings:
def test_settings_requires_login(self, client):
resp = client.post("/ntfy-settings", data={
"ntfy_url": "https://ntfy.sh/test",
"ntfy_token": "",
})
assert resp.status_code == 302 # redirects to login
def test_settings_saves_url_and_generates_callback_token(self, admin_client):
resp = admin_client.post("/ntfy-settings", data={
"ntfy_url": "https://ntfy.sh/my-alerts",
"ntfy_token": "tk_secret",
}, follow_redirects=True)
assert resp.status_code == 200
db = get_db()
row = db.execute(
"SELECT ntfy_url, ntfy_token, callback_token FROM users WHERE username = ?",
("admin",)
).fetchone()
assert row["ntfy_url"] == "https://ntfy.sh/my-alerts"
assert row["ntfy_token"] == "tk_secret"
assert len(row["callback_token"]) >= 32 # random token generated
def test_settings_clears_on_empty_url(self, client):
token = _setup_ntfy_user(client)
# User is logged in as ntfyuser
# Verify settings exist
db = get_db()
row = db.execute(
"SELECT ntfy_url, callback_token FROM users WHERE username = ?",
("ntfyuser",)
).fetchone()
assert row["ntfy_url"] != ""
assert row["callback_token"] != ""
# Clear
client.post("/ntfy-settings", data={
"ntfy_url": "",
"ntfy_token": "",
}, follow_redirects=True)
row = db.execute(
"SELECT ntfy_url, callback_token FROM users WHERE username = ?",
("ntfyuser",)
).fetchone()
assert row["ntfy_url"] == ""
assert row["callback_token"] == ""
def test_settings_preserves_callback_token_on_update(self, admin_client):
# First save
admin_client.post("/ntfy-settings", data={
"ntfy_url": "https://ntfy.sh/initial",
"ntfy_token": "tk_old",
}, follow_redirects=True)
db = get_db()
row = db.execute(
"SELECT callback_token FROM users WHERE username = ?", ("admin",)
).fetchone()
first_token = row["callback_token"]
# Update URL, token should stay same
admin_client.post("/ntfy-settings", data={
"ntfy_url": "https://ntfy.sh/updated",
"ntfy_token": "tk_new",
}, follow_redirects=True)
row = db.execute(
"SELECT ntfy_url, callback_token FROM users WHERE username = ?", ("admin",)
).fetchone()
assert row["ntfy_url"] == "https://ntfy.sh/updated"
assert row["callback_token"] == first_token # unchanged
class TestNtfyTestEndpoint:
def test_test_requires_login(self, client):
resp = client.post("/ntfy-test", data={
"ntfy_url": "https://ntfy.sh/test",
})
assert resp.status_code == 302