Ryan Kegel
8da61ce335
Some checks failed
Chore App Build, Test, and Push Docker Images / build-and-push (push) Failing after 2m12s
- Implemented a new endpoint `/admin/test/send-digest` in `admin_api.py` to trigger digest emails for specific users. - Added a script `send_digest.py` to facilitate sending digest emails via the admin API. - Enhanced the digest action handling in `digest_action_api.py` to support token peeking without consuming it. - Updated the `send_digests` function in `digest_scheduler.py` to utilize the new `send_digest_for_user` function for sending emails. - Introduced a new utility function `peek_token` in `digest_token.py` to validate tokens without consuming them. - Modified the `ParentView.vue` component to handle digest actions upon receiving a digest token in the URL. - Updated `.gitignore` to exclude sensitive certificate files.
293 lines
11 KiB
Python
293 lines
11 KiB
Python
import os
|
|
import jwt
|
|
import pytest
|
|
from datetime import datetime, timedelta, timezone
|
|
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"
|
|
OTHER_USER_ID = "daa_other_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 make_auth_token(user_id: str) -> str:
|
|
"""Create a JWT access token for the given user_id, for use in tests."""
|
|
payload = {
|
|
'user_id': user_id,
|
|
'token_version': 0,
|
|
'exp': datetime.now(timezone.utc) + timedelta(minutes=15),
|
|
}
|
|
return jwt.encode(payload, TEST_SECRET_KEY, algorithm='HS256')
|
|
|
|
|
|
def seed_data():
|
|
users_db.remove(Query().id == TEST_USER_ID)
|
|
users_db.remove(Query().id == OTHER_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",
|
|
"token_version": 0,
|
|
})
|
|
users_db.insert({
|
|
"id": OTHER_USER_ID,
|
|
"first_name": "Other",
|
|
"last_name": "User",
|
|
"email": "other@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",
|
|
"token_version": 0,
|
|
})
|
|
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()
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_client(client):
|
|
"""Client pre-authenticated as TEST_USER_ID via JWT cookie."""
|
|
client.set_cookie('access_token', make_auth_token(TEST_USER_ID))
|
|
return client
|
|
|
|
|
|
class TestHandleDigestActionGet:
|
|
"""GET /digest-action/<token_id>: validates token, redirects with digestToken param, executes nothing."""
|
|
|
|
def test_redirects_to_parent_view_with_digest_token(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 f'digestToken={token.id}' in location
|
|
assert 'scrollTo=' in location
|
|
|
|
def test_does_not_execute_action(self, client):
|
|
"""GET must not change any data; action is deferred to the authenticated POST."""
|
|
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'] == 100 # unchanged
|
|
|
|
def test_does_not_consume_token(self, client):
|
|
"""GET can be called multiple times without consuming the token."""
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
res1 = client.get(f'/digest-action/{token.id}')
|
|
res2 = client.get(f'/digest-action/{token.id}')
|
|
assert res1.status_code == 302
|
|
assert res2.status_code == 302
|
|
|
|
def test_invalid_token_returns_400(self, client):
|
|
res = client.get('/digest-action/nonexistent-token-id')
|
|
assert res.status_code == 400
|
|
|
|
def test_already_resolved_chore_still_redirects(self, client):
|
|
"""GET still redirects even if the pending chore no longer exists; token is valid."""
|
|
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 TestExecuteDigestActionPost:
|
|
"""POST /digest-action/<token_id>: requires auth, consumes token, executes action."""
|
|
|
|
def test_approve_chore_awards_points(self, auth_client):
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
res = auth_client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 200
|
|
child = child_db.get(Query().id == TEST_CHILD_ID)
|
|
assert child['points'] == 110 # 100 + 10
|
|
|
|
def test_deny_chore_removes_pending(self, auth_client):
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'deny')
|
|
res = auth_client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 200
|
|
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, auth_client):
|
|
add_pending_reward()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_REWARD_ID, 'reward', 'approve')
|
|
res = auth_client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 200
|
|
child = child_db.get(Query().id == TEST_CHILD_ID)
|
|
assert child['points'] == 80 # 100 - 20
|
|
|
|
def test_deny_reward_removes_pending(self, auth_client):
|
|
add_pending_reward()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_REWARD_ID, 'reward', 'deny')
|
|
res = auth_client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 200
|
|
pending = pending_confirmations_db.get(
|
|
(Query().child_id == TEST_CHILD_ID) & (Query().entity_id == TEST_REWARD_ID)
|
|
)
|
|
assert pending is None
|
|
|
|
def test_response_contains_success_payload(self, auth_client):
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
res = auth_client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 200
|
|
data = res.get_json()
|
|
assert data['success'] is True
|
|
assert data['child_id'] == TEST_CHILD_ID
|
|
assert data['entity_id'] == TEST_TASK_ID
|
|
assert data['action'] == 'approve'
|
|
|
|
def test_unauthenticated_returns_401(self, client):
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
res = client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 401
|
|
|
|
def test_wrong_user_returns_403(self, client):
|
|
"""A token created for TEST_USER_ID cannot be used by OTHER_USER_ID."""
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
client.set_cookie('access_token', make_auth_token(OTHER_USER_ID))
|
|
res = client.post(f'/digest-action/{token.id}')
|
|
assert res.status_code == 403
|
|
|
|
def test_invalid_token_returns_400(self, auth_client):
|
|
res = auth_client.post('/digest-action/nonexistent-token-id')
|
|
assert res.status_code == 400
|
|
|
|
def test_used_token_returns_400(self, auth_client):
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
auth_client.post(f'/digest-action/{token.id}') # first use — consumes token
|
|
res = auth_client.post(f'/digest-action/{token.id}') # second use
|
|
assert res.status_code == 400
|
|
|
|
def test_get_after_post_returns_400(self, auth_client):
|
|
"""Once the token is consumed via POST, the GET redirect should also fail."""
|
|
add_pending_chore()
|
|
token = create_action_token(TEST_USER_ID, TEST_CHILD_ID, TEST_TASK_ID, 'chore', 'approve')
|
|
auth_client.post(f'/digest-action/{token.id}') # consumes token
|
|
res = auth_client.get(f'/digest-action/{token.id}')
|
|
assert res.status_code == 400
|
|
|
|
|
|
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()
|