Ryan Kegel
ebaef16daf
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 3m23s
- Introduced a dual-token system for user authentication: a short-lived access token and a long-lived rotating refresh token. - Created a new RefreshToken model to manage refresh tokens securely. - Updated auth_api.py to handle login, refresh, and logout processes with the new token system. - Enhanced security measures including token rotation and theft detection. - Updated frontend to handle token refresh on 401 errors and adjusted SSE authentication. - Removed CORS middleware as it's unnecessary behind the nginx proxy. - Added tests to ensure functionality and security of the new token system.
482 lines
19 KiB
Python
482 lines
19 KiB
Python
import pytest
|
|
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
|
|
import os
|
|
from werkzeug.security import generate_password_hash
|
|
from datetime import date as date_type
|
|
|
|
from flask import Flask
|
|
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, tracking_events_db
|
|
from tinydb import Query
|
|
from models.child import Child
|
|
from models.pending_confirmation import PendingConfirmation
|
|
from models.tracking_event import TrackingEvent
|
|
|
|
|
|
TEST_EMAIL = "testuser@example.com"
|
|
TEST_PASSWORD = "testpass"
|
|
TEST_USER_ID = "testuserid"
|
|
|
|
|
|
def add_test_user():
|
|
users_db.remove(Query().email == TEST_EMAIL)
|
|
users_db.insert({
|
|
"id": TEST_USER_ID,
|
|
"first_name": "Test",
|
|
"last_name": "User",
|
|
"email": TEST_EMAIL,
|
|
"password": generate_password_hash(TEST_PASSWORD),
|
|
"verified": True,
|
|
"image_id": "boy01"
|
|
})
|
|
|
|
|
|
def login_and_set_cookie(client):
|
|
resp = client.post('/auth/login', json={"email": TEST_EMAIL, "password": TEST_PASSWORD})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
@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
|
|
with app.test_client() as client:
|
|
add_test_user()
|
|
login_and_set_cookie(client)
|
|
yield client
|
|
|
|
|
|
def setup_child_and_chore(child_name='TestChild', age=8, chore_points=10):
|
|
"""Helper to create a child with one assigned chore."""
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
tracking_events_db.truncate()
|
|
|
|
task_db.insert({
|
|
'id': 'chore1', 'name': 'Sweep Floor', 'points': chore_points,
|
|
'type': 'chore', 'user_id': TEST_USER_ID, 'image_id': 'broom'
|
|
})
|
|
child = Child(name=child_name, age=age, image_id='boy01').to_dict()
|
|
child['tasks'] = ['chore1']
|
|
child['user_id'] = TEST_USER_ID
|
|
child['points'] = 50
|
|
child_db.insert(child)
|
|
return child['id'], 'chore1'
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Child Confirm Flow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_child_confirm_chore_success(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
resp = client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert 'confirmation_id' in data
|
|
|
|
# Verify PendingConfirmation was created
|
|
PQ = Query()
|
|
pending = pending_confirmations_db.get(
|
|
(PQ.child_id == child_id) & (PQ.entity_id == task_id) & (PQ.entity_type == 'chore')
|
|
)
|
|
assert pending is not None
|
|
assert pending['status'] == 'pending'
|
|
|
|
|
|
def test_child_confirm_chore_not_assigned(client):
|
|
child_id, _ = setup_child_and_chore()
|
|
resp = client.post(f'/child/{child_id}/confirm-chore', json={'task_id': 'nonexistent'})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'ENTITY_NOT_ASSIGNED'
|
|
|
|
|
|
def test_child_confirm_chore_not_found(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
child = Child(name='Kid', age=7, image_id='boy01').to_dict()
|
|
child['tasks'] = ['missing_task']
|
|
child['user_id'] = TEST_USER_ID
|
|
child_db.insert(child)
|
|
resp = client.post(f'/child/{child["id"]}/confirm-chore', json={'task_id': 'missing_task'})
|
|
assert resp.status_code == 404
|
|
assert resp.get_json()['code'] == 'TASK_NOT_FOUND'
|
|
|
|
|
|
def test_child_confirm_chore_child_not_found(client):
|
|
resp = client.post('/child/fake_child/confirm-chore', json={'task_id': 'chore1'})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_child_confirm_chore_already_pending(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
resp = client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'CHORE_ALREADY_PENDING'
|
|
|
|
|
|
def test_child_confirm_chore_already_completed_today(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
# Simulate an approved confirmation for today
|
|
from datetime import datetime, timezone
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
pending_confirmations_db.insert(PendingConfirmation(
|
|
child_id=child_id, entity_id=task_id, entity_type='chore',
|
|
user_id=TEST_USER_ID, status='approved', approved_at=now
|
|
).to_dict())
|
|
resp = client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'CHORE_ALREADY_COMPLETED'
|
|
|
|
|
|
def test_child_confirm_chore_creates_tracking_event(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
events = tracking_events_db.all()
|
|
confirmed_events = [e for e in events if e.get('action') == 'confirmed' and e.get('entity_type') == 'chore']
|
|
assert len(confirmed_events) == 1
|
|
assert confirmed_events[0]['entity_id'] == task_id
|
|
assert confirmed_events[0]['points_before'] == confirmed_events[0]['points_after']
|
|
|
|
|
|
def test_child_confirm_chore_wrong_type(client):
|
|
"""Kindness and penalty tasks cannot be confirmed."""
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
task_db.insert({
|
|
'id': 'kind1', 'name': 'Kind Act', 'points': 5,
|
|
'type': 'kindness', 'user_id': TEST_USER_ID
|
|
})
|
|
child = Child(name='Kid', age=7, image_id='boy01').to_dict()
|
|
child['tasks'] = ['kind1']
|
|
child['user_id'] = TEST_USER_ID
|
|
child_db.insert(child)
|
|
resp = client.post(f'/child/{child["id"]}/confirm-chore', json={'task_id': 'kind1'})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'INVALID_TASK_TYPE'
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Child Cancel Flow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_child_cancel_confirm_success(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
resp = client.post(f'/child/{child_id}/cancel-confirm-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
# Pending record should be deleted
|
|
PQ = Query()
|
|
assert pending_confirmations_db.get(
|
|
(PQ.child_id == child_id) & (PQ.entity_id == task_id) & (PQ.entity_type == 'chore')
|
|
) is None
|
|
|
|
|
|
def test_child_cancel_confirm_not_pending(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
resp = client.post(f'/child/{child_id}/cancel-confirm-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'PENDING_NOT_FOUND'
|
|
|
|
|
|
def test_child_cancel_confirm_creates_tracking_event(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
tracking_events_db.truncate()
|
|
client.post(f'/child/{child_id}/cancel-confirm-chore', json={'task_id': task_id})
|
|
events = tracking_events_db.all()
|
|
cancelled = [e for e in events if e.get('action') == 'cancelled']
|
|
assert len(cancelled) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parent Approve Flow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_parent_approve_chore_success(client):
|
|
child_id, task_id = setup_child_and_chore(chore_points=10)
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
|
|
child_before = child_db.get(Query().id == child_id)
|
|
points_before = child_before['points']
|
|
|
|
resp = client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['points'] == points_before + 10
|
|
|
|
# Verify confirmation is now approved
|
|
PQ = Query()
|
|
conf = pending_confirmations_db.get(
|
|
(PQ.child_id == child_id) & (PQ.entity_id == task_id) & (PQ.entity_type == 'chore')
|
|
)
|
|
assert conf['status'] == 'approved'
|
|
assert conf['approved_at'] is not None
|
|
|
|
|
|
def test_parent_approve_chore_not_pending(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
resp = client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'PENDING_NOT_FOUND'
|
|
|
|
|
|
def test_parent_approve_chore_creates_tracking_event(client):
|
|
child_id, task_id = setup_child_and_chore(chore_points=15)
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
tracking_events_db.truncate()
|
|
client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
events = tracking_events_db.all()
|
|
approved = [e for e in events if e.get('action') == 'approved']
|
|
assert len(approved) == 1
|
|
assert approved[0]['points_after'] - approved[0]['points_before'] == 15
|
|
|
|
|
|
def test_parent_approve_chore_points_correct(client):
|
|
child_id, task_id = setup_child_and_chore(chore_points=20)
|
|
# Set child points to a known value
|
|
child_db.update({'points': 100}, Query().id == child_id)
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
resp = client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['points'] == 120
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parent Reject Flow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_parent_reject_chore_success(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
child_db.update({'points': 50}, Query().id == child_id)
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
resp = client.post(f'/child/{child_id}/reject-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
# Points unchanged
|
|
child = child_db.get(Query().id == child_id)
|
|
assert child['points'] == 50
|
|
# Pending record removed
|
|
PQ = Query()
|
|
assert pending_confirmations_db.get(
|
|
(PQ.child_id == child_id) & (PQ.entity_id == task_id)
|
|
) is None
|
|
|
|
|
|
def test_parent_reject_chore_not_pending(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
resp = client.post(f'/child/{child_id}/reject-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['code'] == 'PENDING_NOT_FOUND'
|
|
|
|
|
|
def test_parent_reject_chore_creates_tracking_event(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
tracking_events_db.truncate()
|
|
client.post(f'/child/{child_id}/reject-chore', json={'task_id': task_id})
|
|
events = tracking_events_db.all()
|
|
rejected = [e for e in events if e.get('action') == 'rejected']
|
|
assert len(rejected) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parent Reset Flow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_parent_reset_chore_success(client):
|
|
child_id, task_id = setup_child_and_chore(chore_points=10)
|
|
# Confirm and approve first
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
# Now reset
|
|
child_before = child_db.get(Query().id == child_id)
|
|
points_before = child_before['points']
|
|
resp = client.post(f'/child/{child_id}/reset-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
# Points unchanged after reset
|
|
child_after = child_db.get(Query().id == child_id)
|
|
assert child_after['points'] == points_before
|
|
# Confirmation record removed
|
|
PQ = Query()
|
|
assert pending_confirmations_db.get(
|
|
(PQ.child_id == child_id) & (PQ.entity_id == task_id)
|
|
) is None
|
|
|
|
|
|
def test_parent_reset_chore_not_completed(client):
|
|
child_id, task_id = setup_child_and_chore()
|
|
resp = client.post(f'/child/{child_id}/reset-chore', json={'task_id': task_id})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_parent_reset_chore_creates_tracking_event(client):
|
|
child_id, task_id = setup_child_and_chore(chore_points=10)
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
tracking_events_db.truncate()
|
|
client.post(f'/child/{child_id}/reset-chore', json={'task_id': task_id})
|
|
events = tracking_events_db.all()
|
|
reset_events = [e for e in events if e.get('action') == 'reset']
|
|
assert len(reset_events) == 1
|
|
|
|
|
|
def test_parent_reset_then_child_confirm_again(client):
|
|
"""Full cycle: confirm → approve → reset → confirm → approve."""
|
|
child_id, task_id = setup_child_and_chore(chore_points=10)
|
|
child_db.update({'points': 0}, Query().id == child_id)
|
|
|
|
# First cycle
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
child = child_db.get(Query().id == child_id)
|
|
assert child['points'] == 10
|
|
|
|
# Reset
|
|
client.post(f'/child/{child_id}/reset-chore', json={'task_id': task_id})
|
|
|
|
# Second cycle
|
|
client.post(f'/child/{child_id}/confirm-chore', json={'task_id': task_id})
|
|
client.post(f'/child/{child_id}/approve-chore', json={'task_id': task_id})
|
|
child = child_db.get(Query().id == child_id)
|
|
assert child['points'] == 20
|
|
|
|
# Verify tracking has two approved events
|
|
approved = [e for e in tracking_events_db.all() if e.get('action') == 'approved']
|
|
assert len(approved) == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parent Direct Trigger
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_parent_trigger_chore_directly_creates_approved_confirmation(client):
|
|
child_id, task_id = setup_child_and_chore(chore_points=10)
|
|
child_db.update({'points': 0}, Query().id == child_id)
|
|
resp = client.post(f'/child/{child_id}/trigger-task', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['points'] == 10
|
|
|
|
# Verify an approved PendingConfirmation exists
|
|
PQ = Query()
|
|
conf = pending_confirmations_db.get(
|
|
(PQ.child_id == child_id) & (PQ.entity_id == task_id) & (PQ.entity_type == 'chore')
|
|
)
|
|
assert conf is not None
|
|
assert conf['status'] == 'approved'
|
|
assert conf['approved_at'] is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pending Confirmations List
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_list_pending_confirmations_returns_chores_and_rewards(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
reward_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
|
|
child_db.insert({
|
|
'id': 'ch1', 'name': 'Alice', 'age': 8, 'points': 100,
|
|
'tasks': ['chore1'], 'rewards': ['rew1'], 'user_id': TEST_USER_ID,
|
|
'image_id': 'girl01'
|
|
})
|
|
task_db.insert({'id': 'chore1', 'name': 'Mop Floor', 'points': 5, 'type': 'chore', 'user_id': TEST_USER_ID, 'image_id': 'broom'})
|
|
reward_db.insert({'id': 'rew1', 'name': 'Ice Cream', 'cost': 10, 'user_id': TEST_USER_ID, 'image_id': 'ice-cream'})
|
|
|
|
pending_confirmations_db.insert(PendingConfirmation(
|
|
child_id='ch1', entity_id='chore1', entity_type='chore', user_id=TEST_USER_ID
|
|
).to_dict())
|
|
pending_confirmations_db.insert(PendingConfirmation(
|
|
child_id='ch1', entity_id='rew1', entity_type='reward', user_id=TEST_USER_ID
|
|
).to_dict())
|
|
|
|
resp = client.get('/pending-confirmations')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['count'] == 2
|
|
types = {c['entity_type'] for c in data['confirmations']}
|
|
assert types == {'chore', 'reward'}
|
|
|
|
|
|
def test_list_pending_confirmations_empty(client):
|
|
pending_confirmations_db.truncate()
|
|
resp = client.get('/pending-confirmations')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['count'] == 0
|
|
assert data['confirmations'] == []
|
|
|
|
|
|
def test_list_pending_confirmations_hydrates_names_and_images(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
|
|
child_db.insert({
|
|
'id': 'ch_hydrate', 'name': 'Bob', 'age': 9, 'points': 20,
|
|
'tasks': ['t_hydrate'], 'rewards': [], 'user_id': TEST_USER_ID,
|
|
'image_id': 'boy02'
|
|
})
|
|
task_db.insert({'id': 't_hydrate', 'name': 'Clean Room', 'points': 5, 'type': 'chore', 'user_id': TEST_USER_ID, 'image_id': 'broom'})
|
|
pending_confirmations_db.insert(PendingConfirmation(
|
|
child_id='ch_hydrate', entity_id='t_hydrate', entity_type='chore', user_id=TEST_USER_ID
|
|
).to_dict())
|
|
|
|
resp = client.get('/pending-confirmations')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['count'] == 1
|
|
conf = data['confirmations'][0]
|
|
assert conf['child_name'] == 'Bob'
|
|
assert conf['entity_name'] == 'Clean Room'
|
|
assert conf['child_image_id'] == 'boy02'
|
|
assert conf['entity_image_id'] == 'broom'
|
|
|
|
|
|
def test_list_pending_confirmations_excludes_approved(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
|
|
child_db.insert({
|
|
'id': 'ch_appr', 'name': 'Carol', 'age': 10, 'points': 0,
|
|
'tasks': ['t_appr'], 'rewards': [], 'user_id': TEST_USER_ID,
|
|
'image_id': 'girl01'
|
|
})
|
|
task_db.insert({'id': 't_appr', 'name': 'Chore', 'points': 5, 'type': 'chore', 'user_id': TEST_USER_ID})
|
|
from datetime import datetime, timezone
|
|
pending_confirmations_db.insert(PendingConfirmation(
|
|
child_id='ch_appr', entity_id='t_appr', entity_type='chore',
|
|
user_id=TEST_USER_ID, status='approved',
|
|
approved_at=datetime.now(timezone.utc).isoformat()
|
|
).to_dict())
|
|
|
|
resp = client.get('/pending-confirmations')
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['count'] == 0
|
|
|
|
|
|
def test_list_pending_confirmations_filters_by_user(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
pending_confirmations_db.truncate()
|
|
|
|
# Create a pending confirmation for a different user
|
|
pending_confirmations_db.insert(PendingConfirmation(
|
|
child_id='other_child', entity_id='other_task', entity_type='chore', user_id='otheruserid'
|
|
).to_dict())
|
|
|
|
resp = client.get('/pending-confirmations')
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['count'] == 0
|