ryan/chore
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 5 Releases Wiki Activity

Add push notification functionality with tests and digest scheduler
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 2m14s
Details

Browse Source 
- Implemented push subscription API with tests for subscribing and unsubscribing users.
- Created web push notification tests triggered by child actions.
- Added digest scheduler to send email digests to users at 9 PM local time.
- Developed utility functions for creating and validating digest action tokens.
- Integrated web push sender to handle sending notifications to users.
- Added service worker for handling push notifications in the frontend.
- Created a push opt-in component for user notification preferences.
- Implemented tests for the push opt-in component to ensure correct behavior.
- Updated frontend services to manage push subscriptions and permissions.
This commit is contained in:
Ryan Kegel
2026-04-15 21:56:10 -04:00
parent 0d50a324a3
commit ad2bdf4c4f
47 changed files with 3177 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/tests/conftest.py
View File

@@ -2,6 +2,9 @@ import os
os.environ['DB_ENV'] = 'test'
os.environ.setdefault('SECRET_KEY', 'test-secret-key')
os.environ.setdefault('REFRESH_TOKEN_EXPIRY_DAYS', '90')
os.environ.setdefault('DIGEST_TOKEN_SECRET', 'test-digest-secret')
os.environ.setdefault('VAPID_PUBLIC_KEY', 'test-vapid-public-key')
os.environ.setdefault('VAPID_PRIVATE_KEY', 'test-vapid-private-key')
import sys
import pytest

100
backend/tests/test_child_api.py
View File

@@ -5,7 +5,7 @@ import os
from flask import Flask
from api.child_api import child_api
from api.auth_api import auth_api
from db.db import child_db, reward_db, task_db, users_db, chore_schedules_db, task_extensions_db
from db.db import child_db, reward_db, task_db, users_db, chore_schedules_db, task_extensions_db, pending_confirmations_db
from tinydb import Query
from models.child import Child
import jwt
@@ -514,4 +514,100 @@ def test_list_child_tasks_no_server_side_filtering(client):
returned_ids = {t['id'] for t in resp.get_json()['tasks']}
# Both good tasks must be present; server never filters based on schedule/time
assert TASK_GOOD_ID in returned_ids
assert extra_id in returned_ids
assert extra_id in returned_ids
# ---------------------------------------------------------------------------
# request-reward: duplicate guard
# ---------------------------------------------------------------------------
def test_request_reward_duplicate_returns_409(client):
"""Requesting the same reward twice without resolution returns 409 Conflict."""
reward_db.insert({'id': 'r_dup', 'name': 'Duplicate Reward', 'cost': 5, 'user_id': 'testuserid'})
child_db.insert({
'id': 'child_dup',
'name': 'Dupe Kid',
'age': 8,
'points': 20,
'tasks': [],
'rewards': ['r_dup'],
'user_id': 'testuserid',
})
first = client.post('/child/child_dup/request-reward', json={'reward_id': 'r_dup'})
assert first.status_code == 200
second = client.post('/child/child_dup/request-reward', json={'reward_id': 'r_dup'})
assert second.status_code == 409
assert second.get_json()['code'] == 'DUPLICATE_REWARD_REQUEST'
# Cleanup
pending_confirmations_db.remove(Query().child_id == 'child_dup')
child_db.remove(Query().id == 'child_dup')
reward_db.remove(Query().id == 'r_dup')
# ---------------------------------------------------------------------------
# deny-reward-request endpoint
# ---------------------------------------------------------------------------
def test_deny_reward_request_removes_pending(client):
"""Denying a reward request removes the pending confirmation and fires REQUEST_CANCELLED."""
reward_db.insert({'id': 'r_deny1', 'name': 'Deny Reward', 'cost': 5, 'user_id': 'testuserid'})
child_db.insert({
'id': 'child_deny1',
'name': 'Deny Kid',
'age': 9,
'points': 10,
'tasks': [],
'rewards': ['r_deny1'],
'user_id': 'testuserid',
})
pending_confirmations_db.insert({
'id': 'pend_deny1',
'child_id': 'child_deny1',
'entity_id': 'r_deny1',
'entity_type': 'reward',
'user_id': 'testuserid',
'status': 'pending',
'approved_at': None,
})
resp = client.post('/child/child_deny1/deny-reward-request', json={'reward_id': 'r_deny1'})
assert resp.status_code == 200
remaining = pending_confirmations_db.get(
(Query().child_id == 'child_deny1') & (Query().entity_id == 'r_deny1')
)
assert remaining is None
# Cleanup
child_db.remove(Query().id == 'child_deny1')
reward_db.remove(Query().id == 'r_deny1')
def test_deny_reward_request_already_resolved_returns_200(client):
"""If no pending request exists, denying returns 200 with an informational message."""
resp = client.post('/child/child_gone/deny-reward-request', json={'reward_id': 'r_gone'})
assert resp.status_code == 200
data = resp.get_json()
assert 'already been resolved' in data['message']
def test_deny_reward_request_requires_auth(client):
"""Deny-reward-request endpoint rejects unauthenticated requests."""
# Create a fresh unauthenticated client
from flask import Flask
from api.child_api import child_api as child_api_bp
from api.auth_api import auth_api as auth_api_bp
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
app2 = Flask(__name__)
app2.register_blueprint(child_api_bp)
app2.register_blueprint(auth_api_bp, url_prefix='/auth')
app2.config['TESTING'] = True
app2.config['SECRET_KEY'] = TEST_SECRET_KEY
app2.config['REFRESH_TOKEN_EXPIRY_DAYS'] = TEST_REFRESH_TOKEN_EXPIRY_DAYS
with app2.test_client() as anon:
resp = anon.post('/child/someid/deny-reward-request', json={'reward_id': 'r1'})
assert resp.status_code == 401

190
backend/tests/test_digest_action_api.py Normal file
View File

@@ -0,0 +1,190 @@
import os
import pytest
from flask import Flask
from werkzeug.security import generate_password_hash
from tinydb import Query
from api.digest_action_api import digest_action_api
from db.db import (
users_db, child_db, task_db, reward_db,
pending_confirmations_db, digest_action_tokens_db,
tracking_events_db,
)
from utils.digest_token import (
create_action_token, create_unsubscribe_token, validate_unsubscribe_token
)
from tests.conftest import TEST_SECRET_KEY
TEST_USER_ID = "daa_user_id"
TEST_CHILD_ID = "daa_child_id"
TEST_TASK_ID = "daa_task_id"
TEST_REWARD_ID = "daa_reward_id"
FRONTEND_URL = "http://localhost:5173"
def seed_data():
users_db.remove(Query().id == TEST_USER_ID)
child_db.remove(Query().id == TEST_CHILD_ID)
task_db.remove(Query().id == TEST_TASK_ID)
reward_db.remove(Query().id == TEST_REWARD_ID)
pending_confirmations_db.remove(Query().user_id == TEST_USER_ID)
digest_action_tokens_db.truncate()
tracking_events_db.remove(Query().user_id == TEST_USER_ID)
users_db.insert({
"id": TEST_USER_ID,
"first_name": "Digest",
"last_name": "Tester",
"email": "digest@example.com",
"password": generate_password_hash("password"),
"verified": True,
"role": "user",
"image_id": None,
"marked_for_deletion": False,
"marked_for_deletion_at": None,
"email_digest_enabled": True,
"timezone": "America/New_York",
})
child_db.insert({
"id": TEST_CHILD_ID,
"user_id": TEST_USER_ID,
"name": "Test Child",
"age": 8,
"tasks": [TEST_TASK_ID],
"rewards": [TEST_REWARD_ID],
"points": 100,
"image_id": None,
})
task_db.insert({
"id": TEST_TASK_ID,
"user_id": TEST_USER_ID,
"name": "Clean Room",
"points": 10,
"type": "chore",
"description": "",
"image_id": None,
})
reward_db.insert({
"id": TEST_REWARD_ID,
"user_id": TEST_USER_ID,
"name": "Extra Screen Time",
"cost": 20,
"description": "",
"image_id": None,
})
def add_pending_chore():
pending_confirmations_db.insert({
"id": "pending_chore_id",
"user_id": TEST_USER_ID,
"child_id": TEST_CHILD_ID,
"entity_id": TEST_TASK_ID,
"entity_type": "chore",
"status": "pending",
"created_at": "2024-01-01T10:00:00+00:00",
})
def add_pending_reward():
pending_confirmations_db.insert({
"id": "pending_reward_id",
"user_id": TEST_USER_ID,
"child_id": TEST_CHILD_ID,
"entity_id": TEST_REWARD_ID,
"entity_type": "reward",
"status": "pending",
"created_at": "2024-01-01T10:00:00+00:00",
})
@pytest.fixture
def client():
app = Flask(__name__)
app.register_blueprint(digest_action_api)
app.config['TESTING'] = True
app.config['SECRET_KEY'] = TEST_SECRET_KEY
app.config['FRONTEND_URL'] = FRONTEND_URL
seed_data()
with app.test_client() as c:
yield c
seed_data()
class TestHandleDigestAction:
def test_approve_chore_redirects_to_child_view(self, client):
add_pending_chore()
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
res = client.get(f'/digest-action/{token.id}')
assert res.status_code == 302
location = res.headers['Location']
assert f'/parent/{TEST_CHILD_ID}' in location
assert 'scrollTo=' in location
def test_approve_chore_awards_points(self, client):
add_pending_chore()
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
client.get(f'/digest-action/{token.id}')
child = child_db.get(Query().id == TEST_CHILD_ID)
assert child['points'] == 110 # 100 + 10
def test_deny_chore_removes_pending(self, client):
add_pending_chore()
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'deny')
client.get(f'/digest-action/{token.id}')
pending = pending_confirmations_db.get(
(Query().child_id == TEST_CHILD_ID) & (Query().entity_id == TEST_TASK_ID)
)
assert pending is None
def test_approve_reward_deducts_points(self, client):
add_pending_reward()
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_REWARD_ID, 'reward', 'approve')
client.get(f'/digest-action/{token.id}')
child = child_db.get(Query().id == TEST_CHILD_ID)
assert child['points'] == 80 # 100 - 20
def test_deny_reward_removes_pending(self, client):
add_pending_reward()
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_REWARD_ID, 'reward', 'deny')
client.get(f'/digest-action/{token.id}')
pending = pending_confirmations_db.get(
(Query().child_id == TEST_CHILD_ID) & (Query().entity_id == TEST_REWARD_ID)
)
assert pending is None
def test_invalid_token_returns_400(self, client):
res = client.get('/digest-action/nonexistent-token-id')
assert res.status_code == 400
def test_used_token_returns_400(self, client):
add_pending_chore()
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
client.get(f'/digest-action/{token.id}') # first use
res = client.get(f'/digest-action/{token.id}') # second use
assert res.status_code == 400
def test_approve_already_resolved_chore_still_redirects(self, client):
"""If the chore was already resolved, action still returns a redirect (idempotent)."""
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
# No pending chore added — already resolved
res = client.get(f'/digest-action/{token.id}')
assert res.status_code == 302
class TestHandleDigestUnsubscribe:
def test_valid_token_unsubscribes_user(self, client):
token = create_unsubscribe_token(TEST_USER_ID)
res = client.get(f'/digest-unsubscribe/{token}')
assert res.status_code == 200
user = users_db.get(Query().id == TEST_USER_ID)
assert user['email_digest_enabled'] is False
def test_invalid_token_returns_400(self, client):
res = client.get('/digest-unsubscribe/garbage-token')
assert res.status_code == 400
def test_response_contains_unsubscribed_message(self, client):
token = create_unsubscribe_token(TEST_USER_ID)
res = client.get(f'/digest-unsubscribe/{token}')
assert b'unsubscribed' in res.data.lower()

262
backend/tests/test_digest_scheduler.py Normal file
View File

@@ -0,0 +1,262 @@
"""Tests for the digest scheduler and email HTML generation."""
import os
import pytest
from unittest.mock import patch, MagicMock
from flask import Flask
from werkzeug.security import generate_password_hash
from tinydb import Query
from utils.digest_scheduler import send_digests
from utils.email_sender import send_digest_email
from db.db import users_db, pending_confirmations_db, child_db, task_db, reward_db
from tests.conftest import TEST_SECRET_KEY
SCHED_USER_ID = "sched_test_user"
SCHED_EMAIL = "schedtest@example.com"
SCHED_CHILD_ID = "sched_child_id"
SCHED_TASK_ID = "sched_task_id"
SCHED_REWARD_ID = "sched_reward_id"
def seed_scheduler_data(
verified=True,
email_digest_enabled=True,
timezone="UTC",
has_pending=True,
):
users_db.remove(Query().id == SCHED_USER_ID)
child_db.remove(Query().id == SCHED_CHILD_ID)
task_db.remove(Query().id == SCHED_TASK_ID)
reward_db.remove(Query().id == SCHED_REWARD_ID)
pending_confirmations_db.remove(Query().user_id == SCHED_USER_ID)
users_db.insert({
"id": SCHED_USER_ID,
"first_name": "Sched",
"last_name": "Tester",
"email": SCHED_EMAIL,
"password": generate_password_hash("schedpass"),
"verified": verified,
"role": "user",
"image_id": None,
"marked_for_deletion": False,
"marked_for_deletion_at": None,
"timezone": timezone,
"email_digest_enabled": email_digest_enabled,
})
child_db.insert({
"id": SCHED_CHILD_ID,
"user_id": SCHED_USER_ID,
"name": "Sched Child",
"age": 8,
"points": 50,
"tasks": [SCHED_TASK_ID],
"rewards": [SCHED_REWARD_ID],
"image_id": None,
})
task_db.insert({
"id": SCHED_TASK_ID,
"user_id": SCHED_USER_ID,
"name": "Clean Room",
"points": 10,
"type": "chore",
"image_id": None,
})
reward_db.insert({
"id": SCHED_REWARD_ID,
"user_id": SCHED_USER_ID,
"name": "Movie Night",
"cost": 20,
"image_id": None,
})
if has_pending:
pending_confirmations_db.insert({
"id": "sched_pending_id",
"user_id": SCHED_USER_ID,
"child_id": SCHED_CHILD_ID,
"entity_id": SCHED_TASK_ID,
"entity_type": "chore",
"status": "pending",
"approved_at": None,
})
def cleanup_scheduler_data():
users_db.remove(Query().id == SCHED_USER_ID)
child_db.remove(Query().id == SCHED_CHILD_ID)
task_db.remove(Query().id == SCHED_TASK_ID)
reward_db.remove(Query().id == SCHED_REWARD_ID)
pending_confirmations_db.remove(Query().user_id == SCHED_USER_ID)
@pytest.fixture
def app():
flask_app = Flask(__name__)
flask_app.config['TESTING'] = True
flask_app.config['SECRET_KEY'] = TEST_SECRET_KEY
flask_app.config['FRONTEND_URL'] = 'http://localhost:5173'
flask_app.config['MAIL_DEFAULT_SENDER'] = 'no-reply@reward-app.local'
return flask_app
class TestSendDigests:
def setup_method(self):
cleanup_scheduler_data()
def teardown_method(self):
cleanup_scheduler_data()
def test_sends_digest_to_eligible_user_at_9pm(self, app):
"""Identifies verified, digest-enabled users whose local time is 9 pm and sends digest."""
seed_scheduler_data()
with patch('utils.digest_scheduler._get_local_hour', return_value=21), \
patch('utils.email_sender.Mail') as mock_mail:
send_digests(app)
assert mock_mail.return_value.send.called
def test_skips_user_with_no_pending_items(self, app):
"""Digest is not sent if there are no pending items."""
seed_scheduler_data(has_pending=False)
with patch('utils.digest_scheduler._get_local_hour', return_value=21), \
patch('utils.email_sender.Mail') as mock_mail:
send_digests(app)
assert not mock_mail.return_value.send.called
def test_skips_unverified_user(self, app):
"""Digest is not sent to unverified users."""
seed_scheduler_data(verified=False)
with patch('utils.digest_scheduler._get_local_hour', return_value=21), \
patch('utils.email_sender.Mail') as mock_mail:
send_digests(app)
assert not mock_mail.return_value.send.called
def test_skips_user_with_digest_disabled(self, app):
"""Digest is not sent to users who have email_digest_enabled == False."""
seed_scheduler_data(email_digest_enabled=False)
with patch('utils.digest_scheduler._get_local_hour', return_value=21), \
patch('utils.email_sender.Mail') as mock_mail:
send_digests(app)
assert not mock_mail.return_value.send.called
def test_skips_user_not_at_9pm(self, app):
"""Digest is not sent when the user's local time is not 21."""
seed_scheduler_data()
with patch('utils.digest_scheduler._get_local_hour', return_value=10), \
patch('utils.email_sender.Mail') as mock_mail:
send_digests(app)
assert not mock_mail.return_value.send.called
def test_reads_timezone_and_falls_back_to_utc(self, app):
"""_get_local_hour uses User.timezone; None falls back to UTC."""
from utils.digest_scheduler import _get_local_hour
from datetime import datetime, timezone as tz
utc_hour = datetime.now(tz.utc).hour
assert _get_local_hour(None) == utc_hour
# A real timezone that differs from UTC (New York is UTC-4 or UTC-5)
result = _get_local_hour("America/New_York")
assert 0 <= result <= 23
def test_skips_when_db_env_is_e2e(self, app):
"""Digest scheduler does nothing in the e2e test environment."""
seed_scheduler_data()
original = os.environ.get('DB_ENV')
try:
os.environ['DB_ENV'] = 'e2e'
with patch('utils.email_sender.Mail') as mock_mail:
send_digests(app)
assert not mock_mail.return_value.send.called
finally:
if original is not None:
os.environ['DB_ENV'] = original
else:
os.environ.pop('DB_ENV', None)
class TestDigestEmailHtml:
"""Tests for the HTML content of send_digest_email."""
@pytest.fixture
def app_ctx(self, app):
with app.app_context():
yield
def _build_items(self):
return [
{
'child_name': 'Alice',
'entity_name': 'Clean Room',
'entity_type': 'chore',
'view_url': 'http://localhost:5173/parent/child1?scrollTo=task1&entityType=chore',
'approve_url': 'http://localhost:5173/api/digest-action/approve_tok',
'deny_url': 'http://localhost:5173/api/digest-action/deny_tok',
'child_id': 'child1',
'entity_id': 'task1',
},
{
'child_name': 'Alice',
'entity_name': 'Movie Night',
'entity_type': 'reward',
'view_url': 'http://localhost:5173/parent/child1?scrollTo=reward1&entityType=reward',
'approve_url': 'http://localhost:5173/api/digest-action/approve_tok2',
'deny_url': 'http://localhost:5173/api/digest-action/deny_tok2',
'child_id': 'child1',
'entity_id': 'reward1',
},
]
def test_email_contains_child_section(self, app_ctx):
"""Email HTML contains a section for each child."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
assert 'Alice' in msg.html
def test_email_contains_item_names(self, app_ctx):
"""Email HTML lists each pending item's name."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
assert 'Clean Room' in msg.html
assert 'Movie Night' in msg.html
def test_email_contains_approve_and_deny_links(self, app_ctx):
"""Email HTML contains Approve and Deny links for each item."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
assert 'approve_tok' in msg.html
assert 'deny_tok' in msg.html
assert 'Approve' in msg.html
assert 'Deny' in msg.html
def test_email_contains_view_links(self, app_ctx):
"""Email HTML contains a View link for each item."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
assert 'View' in msg.html
assert 'scrollTo=task1' in msg.html
def test_email_contains_unsubscribe_link(self, app_ctx):
"""Email HTML footer contains the unsubscribe link."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
assert 'unsub_tok' in msg.html
assert 'Unsubscribe' in msg.html
def test_approve_link_styled_green(self, app_ctx):
"""Approve links use green color styling."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
# Find the Approve link and verify green color is adjacent
assert '#22863a' in msg.html # green used for Approve
def test_deny_link_styled_red(self, app_ctx):
"""Deny links use red color styling."""
with patch('utils.email_sender.Mail') as mock_mail:
send_digest_email(SCHED_EMAIL, self._build_items(), 'unsub_tok')
msg = mock_mail.return_value.send.call_args[0][0]
assert '#cb2431' in msg.html # red used for Deny

133
backend/tests/test_digest_token.py Normal file
View File

@@ -0,0 +1,133 @@
import os
import time
import pytest
from tinydb import Query
from db.db import digest_action_tokens_db
from utils.digest_token import (
create_action_token,
validate_and_consume_token,
create_unsubscribe_token,
validate_unsubscribe_token,
)
def cleanup():
digest_action_tokens_db.truncate()
@pytest.fixture(autouse=True)
def clear_tokens():
cleanup()
yield
cleanup()
class TestCreateActionToken:
def test_returns_token_with_correct_fields(self):
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
assert token.user_id == 'u1'
assert token.child_id == 'c1'
assert token.entity_id == 'e1'
assert token.entity_type == 'chore'
assert token.action == 'approve'
assert token.used is False
assert token.signature
def test_persists_to_db(self):
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
stored = digest_action_tokens_db.get(Query().id == token.id)
assert stored is not None
assert stored['entity_type'] == 'chore'
def test_different_tokens_have_unique_ids(self):
t1 = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
t2 = create_action_token('u1', 'c1', 'e1', 'reward', 'deny')
assert t1.id != t2.id
class TestValidateAndConsumeToken:
def test_valid_token_is_returned_and_marked_used(self):
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
result = validate_and_consume_token(token.id)
assert result is not None
assert result.id == token.id
stored = digest_action_tokens_db.get(Query().id == token.id)
assert stored['used'] is True
def test_nonexistent_token_returns_none(self):
assert validate_and_consume_token('nonexistent-id') is None
def test_already_used_token_returns_none(self):
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
validate_and_consume_token(token.id) # first use
result = validate_and_consume_token(token.id) # second use
assert result is None
def test_expired_token_returns_none(self):
# Create a token that is already expired
from datetime import datetime, timedelta, timezone
from db.digest_action_tokens import insert_token
from models.digest_action_token import DigestActionToken
import uuid, json, hmac, hashlib
token_id = str(uuid.uuid4())
expires_at = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
payload = {
'id': token_id,
'user_id': 'u1',
'child_id': 'c1',
'entity_id': 'e1',
'entity_type': 'chore',
'action': 'approve',
'expires_at': expires_at,
}
secret = os.environ.get('DIGEST_TOKEN_SECRET')
canonical = json.dumps(payload, sort_keys=True, separators=(',', ':'))
sig = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).hexdigest()
token = DigestActionToken(
id=token_id, user_id='u1', child_id='c1', entity_id='e1',
entity_type='chore', action='approve', expires_at=expires_at,
used=False, signature=sig,
)
insert_token(token)
assert validate_and_consume_token(token_id) is None
def test_tampered_signature_returns_none(self):
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
# Tamper the signature in DB
digest_action_tokens_db.update(
{'signature': 'deadbeef' * 8},
Query().id == token.id
)
assert validate_and_consume_token(token.id) is None
class TestUnsubscribeToken:
def test_create_and_validate(self):
token = create_unsubscribe_token('user123')
assert token
result = validate_unsubscribe_token(token)
assert result == 'user123'
def test_invalid_token_returns_none(self):
assert validate_unsubscribe_token('garbage-token') is None
def test_tampered_token_returns_none(self):
token = create_unsubscribe_token('user123')
# Change one character
tampered = token[:-1] + ('A' if token[-1] != 'A' else 'B')
assert validate_unsubscribe_token(tampered) is None
def test_expired_token_returns_none(self):
"""Build a token with a past expiry timestamp by mocking time."""
import base64, hmac, hashlib
user_id = 'user_expired'
expiry_ts = int(time.time()) - 1 # already expired
payload_str = f"{user_id}:{expiry_ts}"
secret = os.environ.get('DIGEST_TOKEN_SECRET')
sig = hmac.new(secret.encode(), payload_str.encode(), hashlib.sha256).hexdigest()
raw = f"{payload_str}:{sig}"
token = base64.urlsafe_b64encode(raw.encode()).decode()
assert validate_unsubscribe_token(token) is None

170
backend/tests/test_push_subscription_api.py Normal file
View File

@@ -0,0 +1,170 @@
import os
import pytest
from flask import Flask
from werkzeug.security import generate_password_hash
from tinydb import Query
from api.push_subscription_api import push_subscription_api
from api.auth_api import auth_api
from db.db import users_db, push_subscriptions_db
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
TEST_EMAIL = "pushtest@example.com"
TEST_PASSWORD = "pushpassword123"
TEST_USER_ID = "push_test_user_id"
TEST_ENDPOINT = "https://fcm.googleapis.com/fcm/send/test-endpoint-abc"
TEST_KEYS = {"p256dh": "BNgz3XcMv1", "auth": "abc123"}
def seed_user():
users_db.remove(Query().email == TEST_EMAIL)
users_db.insert({
"id": TEST_USER_ID,
"first_name": "Push",
"last_name": "Tester",
"email": TEST_EMAIL,
"password": generate_password_hash(TEST_PASSWORD),
"verified": True,
"image_id": None,
"marked_for_deletion": False,
"marked_for_deletion_at": None,
"role": "user",
"timezone": None,
"email_digest_enabled": True,
})
def cleanup():
users_db.remove(Query().email == TEST_EMAIL)
push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
@pytest.fixture
def client():
app = Flask(__name__)
app.register_blueprint(push_subscription_api)
app.register_blueprint(auth_api, url_prefix='/auth')
app.config['TESTING'] = True
app.config['SECRET_KEY'] = TEST_SECRET_KEY
app.config['REFRESH_TOKEN_EXPIRY_DAYS'] = TEST_REFRESH_TOKEN_EXPIRY_DAYS
app.config['VAPID_PUBLIC_KEY'] = 'test-vapid-public-key'
app.config['FRONTEND_URL'] = 'http://localhost:5173'
cleanup()
seed_user()
with app.test_client() as c:
yield c
cleanup()
@pytest.fixture
def auth_client(client):
client.post('/auth/login', json={"email": TEST_EMAIL, "password": TEST_PASSWORD})
return client
class TestVapidKey:
def test_returns_public_key(self, client):
res = client.get('/push-vapid-key')
assert res.status_code == 200
assert res.get_json()['public_key'] == 'test-vapid-public-key'
def test_unauthenticated_ok(self, client):
# VAPID key endpoint is public
res = client.get('/push-vapid-key')
assert res.status_code == 200
class TestSubscribe:
def test_subscribe_stores_subscription(self, auth_client):
res = auth_client.post('/push-subscription', json={
"endpoint": TEST_ENDPOINT,
"keys": TEST_KEYS,
})
assert res.status_code == 200
data = res.get_json()
assert 'id' in data
saved = push_subscriptions_db.get(
(Query().user_id == TEST_USER_ID) & (Query().endpoint == TEST_ENDPOINT)
)
assert saved is not None
def test_subscribe_updates_timezone(self, auth_client):
auth_client.post('/push-subscription', json={
"endpoint": TEST_ENDPOINT,
"keys": TEST_KEYS,
"timezone": "America/New_York",
})
user = users_db.get(Query().id == TEST_USER_ID)
assert user['timezone'] == 'America/New_York'
def test_subscribe_upserts_same_endpoint(self, auth_client):
auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
count = len(push_subscriptions_db.search(
(Query().user_id == TEST_USER_ID) & (Query().endpoint == TEST_ENDPOINT)
))
assert count == 1
def test_subscribe_requires_endpoint(self, auth_client):
res = auth_client.post('/push-subscription', json={"keys": TEST_KEYS})
assert res.status_code == 400
def test_subscribe_requires_keys(self, auth_client):
res = auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT})
assert res.status_code == 400
def test_subscribe_requires_p256dh_and_auth(self, auth_client):
res = auth_client.post('/push-subscription', json={
"endpoint": TEST_ENDPOINT,
"keys": {"p256dh": "only-one-key"},
})
assert res.status_code == 400
def test_subscribe_requires_auth(self, client):
res = client.post('/push-subscription', json={
"endpoint": TEST_ENDPOINT,
"keys": TEST_KEYS,
})
assert res.status_code == 401
def test_subscribe_multiple_endpoints_per_user(self, auth_client):
"""Multiple subscriptions can coexist for the same user (one per device)."""
endpoint2 = "https://fcm.googleapis.com/fcm/send/second-device-endpoint"
keys2 = {"p256dh": "BNgz3XcMv2", "auth": "def456"}
auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
auth_client.post('/push-subscription', json={"endpoint": endpoint2, "keys": keys2})
subs = push_subscriptions_db.search(Query().user_id == TEST_USER_ID)
assert len(subs) == 2
endpoints = {s['endpoint'] for s in subs}
assert TEST_ENDPOINT in endpoints
assert endpoint2 in endpoints
class TestUnsubscribe:
def test_unsubscribe_removes_subscription(self, auth_client):
auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
res = auth_client.delete('/push-subscription', json={"endpoint": TEST_ENDPOINT})
assert res.status_code == 200
data = res.get_json()
assert data['removed'] >= 1
saved = push_subscriptions_db.get(
(Query().user_id == TEST_USER_ID) & (Query().endpoint == TEST_ENDPOINT)
)
assert saved is None
def test_unsubscribe_nonexistent_endpoint_ok(self, auth_client):
res = auth_client.delete('/push-subscription', json={"endpoint": "https://nonexistent"})
assert res.status_code == 200
assert res.get_json()['removed'] == 0
def test_unsubscribe_requires_endpoint(self, auth_client):
res = auth_client.delete('/push-subscription', json={})
assert res.status_code == 400
def test_unsubscribe_requires_auth(self, client):
res = client.delete('/push-subscription', json={"endpoint": TEST_ENDPOINT})
assert res.status_code == 401

33
backend/tests/test_user_api.py
View File

@@ -227,3 +227,36 @@ def test_update_profile_success(authenticated_client):
assert user['first_name'] == 'Updated'
assert user['last_name'] == 'Name'
assert user['image_id'] == 'new_image'
def test_get_profile_includes_email_digest_enabled(authenticated_client):
"""GET /user/profile response includes email_digest_enabled field."""
response = authenticated_client.get('/user/profile')
assert response.status_code == 200
data = response.get_json()
assert 'email_digest_enabled' in data
assert isinstance(data['email_digest_enabled'], bool)
def test_update_profile_disables_digest(authenticated_client):
"""PUT /user/profile with email_digest_enabled: false disables the digest."""
# Ensure it starts enabled
users_db.update({'email_digest_enabled': True}, Query().email == TEST_EMAIL)
response = authenticated_client.put('/user/profile', json={'email_digest_enabled': False})
assert response.status_code == 200
user = users_db.search(Query().email == TEST_EMAIL)[0]
assert user['email_digest_enabled'] is False
def test_update_profile_enables_digest(authenticated_client):
"""PUT /user/profile with email_digest_enabled: true re-enables the digest."""
# Start disabled
users_db.update({'email_digest_enabled': False}, Query().email == TEST_EMAIL)
response = authenticated_client.put('/user/profile', json={'email_digest_enabled': True})
assert response.status_code == 200
user = users_db.search(Query().email == TEST_EMAIL)[0]
assert user['email_digest_enabled'] is True

201
backend/tests/test_web_push.py Normal file
View File

@@ -0,0 +1,201 @@
"""Tests for web push notifications triggered by child actions.
Patches `utils.push_sender.webpush` so no real HTTP requests are made.
A push subscription is seeded in push_subscriptions_db so send_push_to_user
actually calls webpush (rather than short-circuiting at "no subscriptions").
"""
import json
import pytest
from unittest.mock import patch, MagicMock
from flask import Flask
from werkzeug.security import generate_password_hash
from tinydb import Query
from api.child_api import child_api
from api.auth_api import auth_api
from db.db import (
child_db, task_db, reward_db, users_db,
pending_confirmations_db, push_subscriptions_db,
)
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
TEST_USER_ID = "wp_test_user_id"
TEST_EMAIL = "wptest@example.com"
TEST_PASSWORD = "wptestpass"
TEST_CHILD_ID = "wp_child_id"
TEST_TASK_ID = "wp_task_id"
TEST_REWARD_ID_AFFORD = "wp_reward_afford"
TEST_REWARD_ID_CANT = "wp_reward_cant"
TEST_ENDPOINT = "https://push.example.com/wp_endpoint"
def seed_data():
users_db.remove(Query().id == TEST_USER_ID)
child_db.remove(Query().id == TEST_CHILD_ID)
task_db.remove(Query().id == TEST_TASK_ID)
reward_db.remove((Query().id == TEST_REWARD_ID_AFFORD) | (Query().id == TEST_REWARD_ID_CANT))
pending_confirmations_db.remove(Query().user_id == TEST_USER_ID)
push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
users_db.insert({
"id": TEST_USER_ID,
"first_name": "Push",
"last_name": "Tester",
"email": TEST_EMAIL,
"password": generate_password_hash(TEST_PASSWORD),
"verified": True,
"role": "user",
"image_id": None,
"marked_for_deletion": False,
"marked_for_deletion_at": None,
"timezone": "America/New_York",
"email_digest_enabled": True,
})
child_db.insert({
"id": TEST_CHILD_ID,
"user_id": TEST_USER_ID,
"name": "Push Child",
"age": 8,
"points": 50,
"tasks": [TEST_TASK_ID],
"rewards": [TEST_REWARD_ID_AFFORD, TEST_REWARD_ID_CANT],
"image_id": None,
})
task_db.insert({
"id": TEST_TASK_ID,
"user_id": TEST_USER_ID,
"name": "Clean Room",
"points": 10,
"type": "chore",
"image_id": None,
})
# Affordable reward (cost <= child.points = 50)
reward_db.insert({
"id": TEST_REWARD_ID_AFFORD,
"user_id": TEST_USER_ID,
"name": "Movie Night",
"cost": 20,
"image_id": None,
})
# Unaffordable reward (cost > child.points = 50)
reward_db.insert({
"id": TEST_REWARD_ID_CANT,
"user_id": TEST_USER_ID,
"name": "New Bike",
"cost": 200,
"image_id": None,
})
def seed_subscription():
push_subscriptions_db.insert({
"id": "wp_sub_id",
"user_id": TEST_USER_ID,
"endpoint": TEST_ENDPOINT,
"keys": {"p256dh": "BNgz3test", "auth": "authtest"},
"created_at": "2024-01-01T00:00:00+00:00",
})
def cleanup():
users_db.remove(Query().id == TEST_USER_ID)
child_db.remove(Query().id == TEST_CHILD_ID)
task_db.remove(Query().id == TEST_TASK_ID)
reward_db.remove((Query().id == TEST_REWARD_ID_AFFORD) | (Query().id == TEST_REWARD_ID_CANT))
pending_confirmations_db.remove(Query().user_id == TEST_USER_ID)
push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
@pytest.fixture
def client():
app = Flask(__name__)
app.register_blueprint(child_api)
app.register_blueprint(auth_api, url_prefix='/auth')
app.config['TESTING'] = True
app.config['SECRET_KEY'] = TEST_SECRET_KEY
app.config['REFRESH_TOKEN_EXPIRY_DAYS'] = TEST_REFRESH_TOKEN_EXPIRY_DAYS
app.config['FRONTEND_URL'] = 'http://localhost:5173'
cleanup()
seed_data()
with app.test_client() as c:
c.post('/auth/login', json={"email": TEST_EMAIL, "password": TEST_PASSWORD})
yield c
cleanup()
class TestWebPushOnChoreConfirm:
def test_push_fired_to_all_subscriptions_on_chore_confirm(self, client):
"""Web push is sent to all stored subscriptions when a chore is confirmed."""
seed_subscription()
with patch('utils.push_sender.webpush') as mock_webpush:
resp = client.post(
f'/child/{TEST_CHILD_ID}/confirm-chore',
json={'task_id': TEST_TASK_ID},
)
assert resp.status_code == 200
assert mock_webpush.called
def test_push_payload_includes_required_fields(self, client):
"""Push payload includes user_id, approve_token, and deny_token."""
seed_subscription()
with patch('utils.push_sender.webpush') as mock_webpush:
client.post(
f'/child/{TEST_CHILD_ID}/confirm-chore',
json={'task_id': TEST_TASK_ID},
)
call_kwargs = mock_webpush.call_args[1]
payload = json.loads(call_kwargs['data'])
assert payload['user_id'] == TEST_USER_ID
assert 'approve_token' in payload
assert 'deny_token' in payload
def test_push_not_fired_when_no_subscriptions(self, client):
"""No error is raised when the parent has no push subscriptions."""
push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
with patch('utils.push_sender.webpush') as mock_webpush:
resp = client.post(
f'/child/{TEST_CHILD_ID}/confirm-chore',
json={'task_id': TEST_TASK_ID},
)
assert resp.status_code == 200
assert not mock_webpush.called
class TestWebPushOnRewardRequest:
def test_push_fired_for_affordable_reward_request(self, client):
"""Web push is fired when a child requests a reward they can afford."""
seed_subscription()
with patch('utils.push_sender.webpush') as mock_webpush:
resp = client.post(
f'/child/{TEST_CHILD_ID}/request-reward',
json={'reward_id': TEST_REWARD_ID_AFFORD},
)
assert resp.status_code == 200
assert mock_webpush.called
# Cleanup pending
pending_confirmations_db.remove(Query().child_id == TEST_CHILD_ID)
def test_push_not_fired_for_unaffordable_reward(self, client):
"""Web push is NOT fired when a child requests a reward they cannot afford."""
seed_subscription()
with patch('utils.push_sender.webpush') as mock_webpush:
resp = client.post(
f'/child/{TEST_CHILD_ID}/request-reward',
json={'reward_id': TEST_REWARD_ID_CANT},
)
# request-reward rejects unaffordable requests with 400
assert resp.status_code == 400
assert not mock_webpush.called
def test_push_not_fired_for_reward_when_no_subscriptions(self, client):
"""No error is raised for reward requests when the parent has no subscriptions."""
push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
with patch('utils.push_sender.webpush') as mock_webpush:
resp = client.post(
f'/child/{TEST_CHILD_ID}/request-reward',
json={'reward_id': TEST_REWARD_ID_AFFORD},
)
assert resp.status_code == 200
assert not mock_webpush.called
# Cleanup pending
pending_confirmations_db.remove(Query().child_id == TEST_CHILD_ID)