Add chore expiry notification system with scheduling and tests
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 3m59s

- Implemented `trigger_chore_expiry.py` script to manually trigger chore expiry notifications for users via the admin API.
- Developed `chore_expiry_notification_scheduler.py` to handle the logic for sending notifications for chores expiring within the next 75 minutes.
- Created utility functions in `schedule_utils.py` to determine scheduling and deadlines for chores.
- Added comprehensive tests for the chore expiry notification system in `test_chore_expiry_notification_scheduler.py`, covering various scenarios including scheduled chores, confirmations, and user settings.
This commit is contained in:
2026-04-22 15:37:40 -04:00
parent 6982fa561f
commit 8907184fde
9 changed files with 1181 additions and 4 deletions

View File

@@ -0,0 +1,112 @@
"""
Script to trigger the chore expiry notification check for a specific user via the admin API.
Usage:
python scripts/trigger_chore_expiry.py <email>
The script logs in as an admin using the ADMIN_EMAIL and ADMIN_PASSWORD
environment variables (defaults to localhost:5000).
Environment variables:
ADMIN_EMAIL - Admin account email (required)
ADMIN_PASSWORD - Admin account password (required)
API_BASE_URL - Base URL of the backend API (default: http://localhost:5000)
NOTE: When targeting a proxied frontend URL (e.g. nginx), append /api to the URL.
e.g. API_BASE_URL=https://dev.chores.ryankegel.com/api
"""
import sys
import os
import json
import urllib.request
import urllib.error
API_BASE_URL = os.environ.get('API_BASE_URL', 'http://localhost:5000').rstrip('/')
def _post(path: str, body: dict, cookie: str | None = None) -> tuple[int, dict]:
url = f"{API_BASE_URL}{path}"
data = json.dumps(body).encode()
headers = {'Content-Type': 'application/json'}
if cookie:
headers['Cookie'] = cookie
req = urllib.request.Request(url, data=data, headers=headers, method='POST')
try:
with urllib.request.urlopen(req) as resp:
return resp.status, json.loads(resp.read())
except urllib.error.HTTPError as e:
try:
return e.code, json.loads(e.read())
except Exception:
return e.code, {}
def _login() -> str:
"""Log in as admin and return the cookie header string."""
admin_email = os.environ.get('ADMIN_EMAIL', '').strip()
admin_password = os.environ.get('ADMIN_PASSWORD', '').strip()
if not admin_email or not admin_password:
print('Error: ADMIN_EMAIL and ADMIN_PASSWORD environment variables are required')
sys.exit(1)
login_url = f"{API_BASE_URL}/auth/login"
login_data = json.dumps({'email': admin_email, 'password': admin_password}).encode()
login_req = urllib.request.Request(
login_url,
data=login_data,
headers={'Content-Type': 'application/json'},
method='POST',
)
try:
with urllib.request.urlopen(login_req) as resp:
raw_cookies = resp.headers.get_all('Set-Cookie') or []
except urllib.error.HTTPError as e:
print(f'Login failed ({e.code})')
sys.exit(1)
cookie_header = '; '.join(part.split(';')[0] for part in raw_cookies)
if not cookie_header:
print('Login succeeded but no cookies were returned')
sys.exit(1)
return cookie_header
def main() -> None:
if len(sys.argv) != 2:
print('Usage: python scripts/trigger_chore_expiry.py <email>')
sys.exit(1)
target_email = sys.argv[1].strip()
cookie = _login()
status, body = _post(
'/admin/test/trigger-chore-expiry',
{'email': target_email},
cookie=cookie,
)
if status == 200:
count = body.get('chores_notified', 0)
if count == 0:
print(f'No expiring chores found for {target_email} in the next 75 minutes — no push sent.')
else:
print(f'Push notification sent for {count} expiring chore(s) for {target_email}.')
elif status == 404 and body.get('code') == 'NOT_FOUND':
print('Error: This endpoint is disabled in the current environment (DB_ENV=production).')
sys.exit(1)
elif status == 404 and body.get('code') == 'USER_NOT_FOUND':
print(f'Error: No user found with email {target_email}')
sys.exit(1)
elif status == 403:
admin_email = os.environ.get('ADMIN_EMAIL', '')
print(f'Error: {admin_email} is not an admin account.')
sys.exit(1)
else:
print(f'Error ({status}): {body.get("error", body)}')
sys.exit(1)
if __name__ == '__main__':
main()