agentbox
710cd1fbf9
CI / test-and-package (push) Successful in 33s
- meal-cron now runs send_reminders_cron.sh (simple wget loop, no DB mount) - New /api/cron-tick endpoint handles all timezone logic server-side - base_url captured from request.host_url when user saves ntfy settings, stored per-user — no APP_BASE_URL env var needed - Cron endpoint protected by CRON_SECRET shared env var - Removed send_reminders_cron.py, DB volume from cron container - tests: 3 new cron-tick + base_url capture tests (69 total)
352 lines
13 KiB
Python
352 lines
13 KiB
Python
"""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, "http://localhost")
|
|
return token
|
|
|
|
|
|
def _setup_ntfy_household(client):
|
|
"""Register an admin, set timezone, configure ntfy. Returns (client, callback_token)."""
|
|
client.post("/register", data={
|
|
"username": "tzadmin",
|
|
"password": "tzpass123",
|
|
"confirm": "tzpass123",
|
|
"household_action": "create",
|
|
"new_household_name": "TZ Cron Household",
|
|
}, follow_redirects=True)
|
|
client.post("/admin/timezone", data={"timezone": "UTC"}, follow_redirects=True)
|
|
client.post("/ntfy-settings", data={
|
|
"ntfy_url": "https://ntfy.sh/test-topic",
|
|
"ntfy_token": "tk_test",
|
|
}, follow_redirects=True)
|
|
|
|
from models import get_db
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT callback_token, base_url FROM users WHERE username = ?", ("tzadmin",)
|
|
).fetchone()
|
|
return client, row["callback_token"], row["base_url"]
|
|
|
|
|
|
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, base_url 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
|
|
assert row["base_url"] == "http://localhost" # captured from request # 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, base_url FROM users WHERE username = ?",
|
|
("ntfyuser",)
|
|
).fetchone()
|
|
assert row["ntfy_url"] == ""
|
|
assert row["callback_token"] == ""
|
|
assert row["base_url"] == ""
|
|
|
|
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
|
|
|
|
|
|
class TestTimezoneSettings:
|
|
def test_admin_can_set_timezone(self, admin_client):
|
|
resp = admin_client.post("/admin/timezone", data={
|
|
"timezone": "Europe/Zurich",
|
|
}, follow_redirects=True)
|
|
assert resp.status_code == 200
|
|
|
|
from models import get_db
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT timezone FROM households WHERE name = ?", ("Admin Household",)
|
|
).fetchone()
|
|
assert row["timezone"] == "Europe/Zurich"
|
|
|
|
def test_non_admin_cannot_set_timezone(self, client):
|
|
# Register a regular user in an existing household
|
|
client.post("/register", data={
|
|
"username": "admin_tz",
|
|
"password": "admin_tz_pass",
|
|
"confirm": "admin_tz_pass",
|
|
"household_action": "create",
|
|
"new_household_name": "TZ Household",
|
|
}, follow_redirects=True)
|
|
client.get("/logout")
|
|
|
|
from models import get_household_by_name
|
|
hh = get_household_by_name("TZ Household")
|
|
client.post("/register", data={
|
|
"username": "regular_tz",
|
|
"password": "regular_tz_pass",
|
|
"confirm": "regular_tz_pass",
|
|
"household_action": "join",
|
|
"household_id": str(hh["id"]),
|
|
}, follow_redirects=True)
|
|
|
|
resp = client.post("/admin/timezone", data={
|
|
"timezone": "Europe/Zurich",
|
|
}, follow_redirects=True)
|
|
assert resp.status_code == 200
|
|
# Timezone should NOT have changed
|
|
from models import get_db
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT timezone FROM households WHERE id = ?", (hh["id"],)
|
|
).fetchone()
|
|
assert row["timezone"] == "UTC" # unchanged
|
|
|
|
def test_default_timezone_is_utc(self, admin_client):
|
|
from models import get_db
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT timezone FROM households WHERE name = ?", ("Admin Household",)
|
|
).fetchone()
|
|
assert row["timezone"] == "UTC"
|
|
|
|
def test_timezone_requires_login(self, client):
|
|
resp = client.post("/admin/timezone", data={
|
|
"timezone": "Europe/Zurich",
|
|
})
|
|
assert resp.status_code == 302
|
|
|
|
|
|
class TestCronTick:
|
|
def test_cron_tick_requires_secret(self, client, monkeypatch):
|
|
monkeypatch.setenv("CRON_SECRET", "my-secret")
|
|
|
|
resp = client.post("/api/cron-tick")
|
|
assert resp.status_code == 403
|
|
|
|
resp = client.post("/api/cron-tick", headers={"X-Cron-Secret": "wrong"})
|
|
assert resp.status_code == 403
|
|
|
|
def test_cron_tick_works_with_secret(self, client, monkeypatch):
|
|
monkeypatch.setenv("CRON_SECRET", "my-secret")
|
|
_setup_ntfy_household(client)
|
|
|
|
resp = client.post("/api/cron-tick",
|
|
headers={"X-Cron-Secret": "my-secret"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "sent" in data
|
|
|
|
def test_ntfy_settings_captures_base_url(self, client):
|
|
"""When saving ntfy settings, the base_url should be captured from the request."""
|
|
_setup_ntfy_household(client)
|
|
|
|
from models import get_db
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT base_url FROM users WHERE username = ?", ("tzadmin",)
|
|
).fetchone()
|
|
# In test client, host_url is http://localhost
|
|
assert row["base_url"] == "http://localhost"
|