Refactor cron to call API via HTTP, capture base_url from browser
CI / test-and-package (push) Successful in 33s
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)
This commit is contained in:
+62
-4
@@ -36,10 +36,33 @@ def _setup_ntfy_user(client, username="ntfyuser", password="testpass",
|
||||
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)
|
||||
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)
|
||||
@@ -164,12 +187,13 @@ class TestNtfySettings:
|
||||
|
||||
db = get_db()
|
||||
row = db.execute(
|
||||
"SELECT ntfy_url, ntfy_token, callback_token FROM users WHERE username = ?",
|
||||
"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 # random token generated
|
||||
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)
|
||||
@@ -191,11 +215,12 @@ class TestNtfySettings:
|
||||
}, follow_redirects=True)
|
||||
|
||||
row = db.execute(
|
||||
"SELECT ntfy_url, callback_token FROM users WHERE username = ?",
|
||||
"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
|
||||
@@ -291,3 +316,36 @@ class TestTimezoneSettings:
|
||||
"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"
|
||||
|
||||
Reference in New Issue
Block a user