Ryan Kegel
ad2bdf4c4f
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 2m14s
- 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.
263 lines
10 KiB
Python
263 lines
10 KiB
Python
"""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
|