Ryan Kegel
ce3d1b3d54
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 2m52s
797 lines
33 KiB
Python
797 lines
33 KiB
Python
import pytest
|
|
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
|
|
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, pending_confirmations_db
|
|
from tinydb import Query
|
|
from models.child import Child
|
|
import jwt
|
|
from werkzeug.security import generate_password_hash
|
|
from datetime import date as date_type, datetime, timedelta, timezone
|
|
|
|
|
|
# Test user credentials
|
|
TEST_EMAIL = "testuser@example.com"
|
|
TEST_PASSWORD = "testpass"
|
|
|
|
def add_test_user():
|
|
# Remove if exists
|
|
users_db.remove(Query().email == TEST_EMAIL)
|
|
users_db.insert({
|
|
"id": "testuserid",
|
|
"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
|
|
# Set cookie for subsequent requests
|
|
cookies = resp.headers.getlist("Set-Cookie")
|
|
cookie_str = ' '.join(cookies)
|
|
assert cookie_str and "access_token=" in cookie_str
|
|
# Flask test client automatically handles cookies
|
|
|
|
@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
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def cleanup_db():
|
|
yield
|
|
child_db.close()
|
|
reward_db.close()
|
|
task_db.close()
|
|
if os.path.exists('children.json'):
|
|
os.remove('children.json')
|
|
if os.path.exists('rewards.json'):
|
|
os.remove('rewards.json')
|
|
if os.path.exists('tasks.json'):
|
|
os.remove('tasks.json')
|
|
|
|
def test_add_child(client):
|
|
response = client.put('/child/add', json={'name': 'Alice', 'age': 8})
|
|
assert response.status_code == 201
|
|
assert b'Child Alice added.' in response.data
|
|
children = child_db.all()
|
|
assert any(c.get('name') == 'Alice' and c.get('age') == 8 and c.get('image_id') == 'boy01' for c in children)
|
|
|
|
response = client.put('/child/add', json={'name': 'Mike', 'age': 4, 'image_id': 'girl02'})
|
|
assert response.status_code == 201
|
|
assert b'Child Mike added.' in response.data
|
|
children = child_db.all()
|
|
assert any(c.get('name') == 'Mike' and c.get('age') == 4 and c.get('image_id') == 'girl02' for c in children)
|
|
|
|
def test_list_children(client):
|
|
child_db.truncate()
|
|
# Insert children for the test user
|
|
child_db.insert({**Child(name='Alice', age=8, image_id="boy01").to_dict(), 'user_id': 'testuserid'})
|
|
child_db.insert({**Child(name='Mike', age=4, image_id="boy01").to_dict(), 'user_id': 'testuserid'})
|
|
response = client.get('/child/list')
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
# Only children for the test user should be returned
|
|
assert len(data['children']) == 2
|
|
|
|
def test_assign_and_remove_task(client):
|
|
client.put('/child/add', json={'name': 'Bob', 'age': 10})
|
|
children = client.get('/child/list').get_json()['children']
|
|
child_id = children[0]['id']
|
|
response = client.post(f'/child/{child_id}/assign-task', json={'task_id': 'task123'})
|
|
assert response.status_code == 200
|
|
ChildQuery = Query()
|
|
child = child_db.search(ChildQuery.id == child_id)[0]
|
|
assert 'task123' in child.get('tasks', [])
|
|
response = client.post(f'/child/{child_id}/remove-task', json={'task_id': 'task123'})
|
|
assert response.status_code == 200
|
|
child = child_db.search(ChildQuery.id == child_id)[0]
|
|
assert 'task123' not in child.get('tasks', [])
|
|
|
|
def test_get_child_not_found(client):
|
|
response = client.get('/child/nonexistent-id')
|
|
assert response.status_code == 404
|
|
assert b'Child not found' in response.data
|
|
|
|
def test_remove_task_not_found(client):
|
|
response = client.post('/child/nonexistent-id/remove-task', json={'task_id': 'task123'})
|
|
assert response.status_code == 404
|
|
assert b'Child not found' in response.data
|
|
|
|
def test_remove_reward_not_found(client):
|
|
response = client.post('/child/nonexistent-id/remove-reward', json={'reward_id': 'reward123'})
|
|
assert response.status_code == 404
|
|
assert b'Child not found' in response.data
|
|
|
|
def test_affordable_rewards(client):
|
|
reward_db.insert({'id': 'r_cheep', 'name': 'Sticker', 'cost': 5, 'user_id': 'testuserid'})
|
|
reward_db.insert({'id': 'r_exp', 'name': 'Bike', 'cost': 20, 'user_id': 'testuserid'})
|
|
client.put('/child/add', json={'name': 'Charlie', 'age': 9})
|
|
children = client.get('/child/list').get_json()['children']
|
|
child_id = next((c['id'] for c in children if c.get('name') == 'Charlie'), children[0]['id'])
|
|
ChildQuery = Query()
|
|
child_db.update({'points': 10}, ChildQuery.id == child_id)
|
|
client.post(f'/child/{child_id}/assign-reward', json={'reward_id': 'r_cheep'})
|
|
client.post(f'/child/{child_id}/assign-reward', json={'reward_id': 'r_exp'})
|
|
resp = client.get(f'/child/{child_id}/affordable-rewards')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
affordable_ids = [r['id'] for r in data['affordable_rewards']]
|
|
assert 'r_cheep' in affordable_ids and 'r_exp' not in affordable_ids
|
|
|
|
def test_reward_status(client):
|
|
reward_db.insert({'id': 'r1', 'name': 'Candy', 'cost': 3, 'image': 'candy.png', 'user_id': 'testuserid'})
|
|
reward_db.insert({'id': 'r2', 'name': 'Game', 'cost': 8, 'image': 'game.png', 'user_id': 'testuserid'})
|
|
reward_db.insert({'id': 'r3', 'name': 'Trip', 'cost': 15, 'image': 'trip.png', 'user_id': 'testuserid'})
|
|
client.put('/child/add', json={'name': 'Dana', 'age': 11})
|
|
children = client.get('/child/list').get_json()['children']
|
|
child_id = next((c['id'] for c in children if c.get('name') == 'Dana'), children[0]['id'])
|
|
ChildQuery = Query()
|
|
child_db.update({'points': 7}, ChildQuery.id == child_id)
|
|
for rid in ['r1', 'r2', 'r3']:
|
|
client.post(f'/child/{child_id}/assign-reward', json={'reward_id': rid})
|
|
resp = client.get(f'/child/{child_id}/reward-status')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
mapping = {s['id']: s['points_needed'] for s in data['reward_status']}
|
|
assert mapping['r1'] == 0 and mapping['r2'] == 1 and mapping['r3'] == 8
|
|
|
|
def test_list_child_tasks_returns_tasks(client):
|
|
task_db.insert({'id': 't_list_1', 'name': 'Task One', 'points': 2, 'type': 'chore', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': 't_list_2', 'name': 'Task Two', 'points': 3, 'type': 'penalty', 'user_id': 'testuserid'})
|
|
child_db.insert({
|
|
'id': 'child_list_1',
|
|
'name': 'Eve',
|
|
'age': 8,
|
|
'points': 0,
|
|
'tasks': ['t_list_1', 't_list_2', 't_missing'],
|
|
'rewards': [],
|
|
'user_id': 'testuserid'
|
|
})
|
|
resp = client.get('/child/child_list_1/list-tasks')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
returned_ids = {t['id'] for t in data['tasks']}
|
|
assert returned_ids == {'t_list_1', 't_list_2'}
|
|
for t in data['tasks']:
|
|
assert 'name' in t and 'points' in t and 'type' in t
|
|
|
|
def test_list_assignable_tasks_returns_expected_ids(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
task_db.insert({'id': 'tA', 'name': 'Task A', 'points': 1, 'type': 'chore', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': 'tB', 'name': 'Task B', 'points': 2, 'type': 'chore', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': 'tC', 'name': 'Task C', 'points': 3, 'type': 'penalty', 'user_id': 'testuserid'})
|
|
client.put('/child/add', json={'name': 'Zoe', 'age': 7})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
client.post(f'/child/{child_id}/assign-task', json={'task_id': 'tA'})
|
|
client.post(f'/child/{child_id}/assign-task', json={'task_id': 'tC'})
|
|
resp = client.get(f'/child/{child_id}/list-assignable-tasks')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert len(data['tasks']) == 1
|
|
assert data['tasks'][0]['id'] == 'tB'
|
|
assert data['count'] == 1
|
|
|
|
def test_list_assignable_tasks_when_none_assigned(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
ids = ['t1', 't2', 't3']
|
|
for i, tid in enumerate(ids, 1):
|
|
task_db.insert({'id': tid, 'name': f'Task {i}', 'points': i, 'type': 'chore', 'user_id': 'testuserid'})
|
|
client.put('/child/add', json={'name': 'Liam', 'age': 6})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
resp = client.get(f'/child/{child_id}/list-assignable-tasks')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
returned_ids = {t['id'] for t in data['tasks']}
|
|
assert returned_ids == set(ids)
|
|
assert data['count'] == len(ids)
|
|
|
|
def test_list_assignable_tasks_empty_task_db(client):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
client.put('/child/add', json={'name': 'Mia', 'age': 5})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
resp = client.get(f'/child/{child_id}/list-assignable-tasks')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['tasks'] == []
|
|
assert data['count'] == 0
|
|
|
|
def test_list_assignable_tasks_child_not_found(client):
|
|
resp = client.get('/child/does-not-exist/list-assignable-tasks')
|
|
assert resp.status_code == 404
|
|
assert b'Child not found' in resp.data
|
|
|
|
def setup_child_with_tasks(child_name='TestChild', age=8, assigned=None):
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
assigned = assigned or []
|
|
# Seed tasks
|
|
task_db.insert({'id': 't1', 'name': 'Task 1', 'points': 1, 'type': 'chore', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': 't2', 'name': 'Task 2', 'points': 2, 'type': 'penalty', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': 't3', 'name': 'Task 3', 'points': 3, 'type': 'chore', 'user_id': 'testuserid'})
|
|
# Seed child
|
|
child = Child(name=child_name, age=age, image_id='boy01').to_dict()
|
|
child['tasks'] = assigned[:]
|
|
child['user_id'] = 'testuserid'
|
|
child_db.insert(child)
|
|
return child['id']
|
|
|
|
def test_list_all_tasks_partitions_assigned_and_assignable(client):
|
|
child_id = setup_child_with_tasks(assigned=['t1', 't3'])
|
|
resp = client.get(f'/child/{child_id}/list-all-tasks')
|
|
# New backend may return 400 for missing type or structure change
|
|
if resp.status_code == 400:
|
|
data = resp.get_json()
|
|
assert 'error' in data
|
|
else:
|
|
data = resp.get_json()
|
|
# Accept either new or old structure
|
|
assigned_ids = set()
|
|
assignable_ids = set()
|
|
if 'assigned_tasks' in data and 'assignable_tasks' in data:
|
|
assigned_ids = {t['id'] for t in data['assigned_tasks']}
|
|
assignable_ids = {t['id'] for t in data['assignable_tasks']}
|
|
assert assigned_ids == {'t1', 't3'}
|
|
assert assignable_ids == {'t2'}
|
|
assert data['assigned_count'] == 2
|
|
assert data['assignable_count'] == 1
|
|
|
|
def test_set_child_tasks_replaces_existing(client):
|
|
child_id = setup_child_with_tasks(assigned=['t1', 't2'])
|
|
payload = {'task_ids': ['t3', 'missing', 't3'], 'type': 'chore'}
|
|
resp = client.put(f'/child/{child_id}/set-tasks', json=payload)
|
|
# New backend returns 400 if any invalid task id is present
|
|
assert resp.status_code in (200, 400)
|
|
data = resp.get_json()
|
|
if resp.status_code == 400:
|
|
assert 'error' in data
|
|
|
|
def test_set_child_tasks_requires_list(client):
|
|
child_id = setup_child_with_tasks(assigned=['t2'])
|
|
resp = client.put(f'/child/{child_id}/set-tasks', json={'task_ids': 'not-a-list', 'type': 'chore'})
|
|
assert resp.status_code == 400
|
|
# Accept any error message
|
|
assert b'error' in resp.data
|
|
|
|
def test_set_child_tasks_child_not_found(client):
|
|
resp = client.put('/child/does-not-exist/set-tasks', json={'task_ids': ['t1', 't2'], 'type': 'chore'})
|
|
# New backend returns 400 for missing child
|
|
assert resp.status_code in (400, 404)
|
|
assert b'error' in resp.data
|
|
|
|
def test_assignable_tasks_user_overrides_system(client):
|
|
"""If a user task exists with the same name as a system task, only the user task should be shown in assignable list."""
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
# System task (user_id=None)
|
|
task_db.insert({'id': 'sys1', 'name': 'Duplicate', 'points': 1, 'type': 'chore', 'user_id': None})
|
|
# User task (same name)
|
|
task_db.insert({'id': 'user1', 'name': 'Duplicate', 'points': 2, 'type': 'chore', 'user_id': 'testuserid'})
|
|
client.put('/child/add', json={'name': 'Sam', 'age': 8})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
resp = client.get(f'/child/{child_id}/list-assignable-tasks')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
names = [t['name'] for t in data['tasks']]
|
|
ids = [t['id'] for t in data['tasks']]
|
|
# Only the user task should be present
|
|
assert names == ['Duplicate']
|
|
assert ids == ['user1']
|
|
|
|
def test_assignable_tasks_multiple_user_same_name(client):
|
|
"""If two user tasks exist with the same name as a system task, both user tasks are shown, not the system one."""
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
# System task (user_id=None)
|
|
task_db.insert({'id': 'sys1', 'name': 'Duplicate', 'points': 1, 'type': 'chore', 'user_id': None})
|
|
# User tasks (same name, different user_ids)
|
|
task_db.insert({'id': 'user1', 'name': 'Duplicate', 'points': 2, 'type': 'chore', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': 'user2', 'name': 'Duplicate', 'points': 3, 'type': 'chore', 'user_id': 'otheruserid'})
|
|
client.put('/child/add', json={'name': 'Sam', 'age': 8})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
resp = client.get(f'/child/{child_id}/list-assignable-tasks')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
names = [t['name'] for t in data['tasks']]
|
|
ids = [t['id'] for t in data['tasks']]
|
|
# Both user tasks should be present, not the system one
|
|
assert set(names) == {'Duplicate'}
|
|
assert set(ids) == {'user1', 'user2'}
|
|
|
|
def test_assignable_rewards_user_overrides_system(client):
|
|
"""If a user reward exists with the same name as a system reward, only the user reward should be shown in assignable list."""
|
|
child_db.truncate()
|
|
reward_db.truncate()
|
|
# System reward (user_id=None)
|
|
reward_db.insert({'id': 'sysr1', 'name': 'Prize', 'cost': 5, 'user_id': None})
|
|
# User reward (same name)
|
|
reward_db.insert({'id': 'userr1', 'name': 'Prize', 'cost': 10, 'user_id': 'testuserid'})
|
|
client.put('/child/add', json={'name': 'Sam', 'age': 8})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
resp = client.get(f'/child/{child_id}/list-assignable-rewards')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
names = [r['name'] for r in data['rewards']]
|
|
ids = [r['id'] for r in data['rewards']]
|
|
# Only the user reward should be present
|
|
assert names == ['Prize']
|
|
assert ids == ['userr1']
|
|
|
|
def test_assignable_rewards_multiple_user_same_name(client):
|
|
"""If two user rewards exist with the same name as a system reward, both user rewards are shown, not the system one."""
|
|
child_db.truncate()
|
|
reward_db.truncate()
|
|
# System reward (user_id=None)
|
|
reward_db.insert({'id': 'sysr1', 'name': 'Prize', 'cost': 5, 'user_id': None})
|
|
# User rewards (same name, different user_ids)
|
|
reward_db.insert({'id': 'userr1', 'name': 'Prize', 'cost': 10, 'user_id': 'testuserid'})
|
|
reward_db.insert({'id': 'userr2', 'name': 'Prize', 'cost': 15, 'user_id': 'otheruserid'})
|
|
client.put('/child/add', json={'name': 'Sam', 'age': 8})
|
|
child_id = client.get('/child/list').get_json()['children'][0]['id']
|
|
resp = client.get(f'/child/{child_id}/list-assignable-rewards')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
names = [r['name'] for r in data['rewards']]
|
|
ids = [r['id'] for r in data['rewards']]
|
|
# Both user rewards should be present, not the system one
|
|
assert set(names) == {'Prize'}
|
|
assert set(ids) == {'userr1', 'userr2'}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list-tasks: schedule and extension_date fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CHILD_SCHED_ID = 'child_sched_test'
|
|
TASK_GOOD_ID = 'task_sched_good'
|
|
TASK_BAD_ID = 'task_sched_bad'
|
|
|
|
|
|
def _setup_sched_child_and_tasks(task_db, child_db):
|
|
task_db.remove(Query().id == TASK_GOOD_ID)
|
|
task_db.remove(Query().id == TASK_BAD_ID)
|
|
task_db.insert({'id': TASK_GOOD_ID, 'name': 'Sweep', 'points': 3, 'type': 'chore', 'user_id': 'testuserid'})
|
|
task_db.insert({'id': TASK_BAD_ID, 'name': 'Yell', 'points': 2, 'type': 'penalty', 'user_id': 'testuserid'})
|
|
child_db.remove(Query().id == CHILD_SCHED_ID)
|
|
child_db.insert({
|
|
'id': CHILD_SCHED_ID,
|
|
'name': 'SchedKid',
|
|
'age': 7,
|
|
'points': 0,
|
|
'tasks': [TASK_GOOD_ID, TASK_BAD_ID],
|
|
'rewards': [],
|
|
'user_id': 'testuserid',
|
|
})
|
|
chore_schedules_db.remove(Query().child_id == CHILD_SCHED_ID)
|
|
task_extensions_db.remove(Query().child_id == CHILD_SCHED_ID)
|
|
pending_confirmations_db.remove(Query().child_id == CHILD_SCHED_ID)
|
|
|
|
|
|
def test_list_child_tasks_always_has_schedule_and_extension_date_keys(client):
|
|
"""Every task in the response must have 'schedule' and 'extension_date' keys."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
assert resp.status_code == 200
|
|
for task in resp.get_json()['tasks']:
|
|
assert 'schedule' in task
|
|
assert 'extension_date' in task
|
|
|
|
|
|
def test_list_child_tasks_returns_schedule_when_set(client):
|
|
"""Good chore with a saved schedule returns that schedule object."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
chore_schedules_db.insert({
|
|
'id': 'sched-1',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'task_id': TASK_GOOD_ID,
|
|
'mode': 'days',
|
|
'day_configs': [{'day': 1, 'hour': 8, 'minute': 0}],
|
|
'interval_days': 2,
|
|
'anchor_weekday': 0,
|
|
'interval_hour': 0,
|
|
'interval_minute': 0,
|
|
})
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
sched = tasks[TASK_GOOD_ID]['schedule']
|
|
assert sched is not None
|
|
assert sched['mode'] == 'days'
|
|
assert sched['day_configs'] == [{'day': 1, 'hour': 8, 'minute': 0}]
|
|
|
|
|
|
def test_list_child_tasks_schedule_null_when_not_set(client):
|
|
"""Good chore with no schedule returns schedule=null."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_GOOD_ID]['schedule'] is None
|
|
|
|
|
|
def test_list_child_tasks_returns_extension_date_when_set(client):
|
|
"""Good chore with a TaskExtension for today returns today's ISO date."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
today = date_type.today().isoformat()
|
|
task_extensions_db.insert({
|
|
'id': 'ext-1',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'task_id': TASK_GOOD_ID,
|
|
'date': today,
|
|
})
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_GOOD_ID]['extension_date'] == today
|
|
|
|
|
|
def test_list_child_tasks_returns_extension_date_for_yesterday(client):
|
|
"""Regression: extension inserted for yesterday's date must still be returned.
|
|
|
|
Previously, list-tasks used date_type.today() on the server to look up
|
|
extensions, causing a timezone mismatch when the client's local date
|
|
differed from the server's UTC date. The fix removes the date filter so
|
|
any stored extension is returned regardless of which date it was created
|
|
for.
|
|
"""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
from datetime import date as date_type_local, timedelta
|
|
yesterday = (date_type_local.today() - timedelta(days=1)).isoformat()
|
|
task_extensions_db.insert({
|
|
'id': 'ext-yesterday',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'task_id': TASK_GOOD_ID,
|
|
'date': yesterday,
|
|
})
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_GOOD_ID]['extension_date'] == yesterday
|
|
|
|
|
|
def test_list_child_tasks_extension_date_null_when_not_set(client):
|
|
"""Good chore with no extension returns extension_date=null."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_GOOD_ID]['extension_date'] is None
|
|
|
|
|
|
def test_list_child_tasks_schedule_and_extension_null_for_penalties(client):
|
|
"""Penalty tasks (type='penalty') always return schedule=null and extension_date=null."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
# Even if we insert a schedule entry for the penalty task, the endpoint should ignore it
|
|
chore_schedules_db.insert({
|
|
'id': 'sched-bad',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'task_id': TASK_BAD_ID,
|
|
'mode': 'days',
|
|
'day_configs': [{'day': 0, 'hour': 9, 'minute': 0}],
|
|
'interval_days': 2,
|
|
'anchor_weekday': 0,
|
|
'interval_hour': 0,
|
|
'interval_minute': 0,
|
|
})
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_BAD_ID]['schedule'] is None
|
|
assert tasks[TASK_BAD_ID]['extension_date'] is None
|
|
|
|
|
|
def test_list_child_tasks_no_server_side_filtering(client):
|
|
"""All assigned tasks are returned regardless of schedule — no server-side day/time filtering."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
# Add a second good task that has a schedule for only Sunday (day=0)
|
|
extra_id = 'task_sched_extra'
|
|
task_db.remove(Query().id == extra_id)
|
|
task_db.insert({'id': extra_id, 'name': 'Extra', 'points': 1, 'type': 'chore', 'user_id': 'testuserid'})
|
|
child_db.update({'tasks': [TASK_GOOD_ID, TASK_BAD_ID, extra_id]}, Query().id == CHILD_SCHED_ID)
|
|
chore_schedules_db.insert({
|
|
'id': 'sched-extra',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'task_id': extra_id,
|
|
'mode': 'days',
|
|
'day_configs': [{'day': 0, 'hour': 7, 'minute': 0}], # Sunday only
|
|
'interval_days': 2,
|
|
'anchor_weekday': 0,
|
|
'interval_hour': 0,
|
|
'interval_minute': 0,
|
|
})
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
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
|
|
|
|
|
|
def test_list_child_tasks_shows_pending_for_today(client):
|
|
"""A chore confirmed today should return pending_status='pending'."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
now_ts = datetime.now(timezone.utc).timestamp()
|
|
pending_confirmations_db.insert({
|
|
'id': 'pend_today_chore',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'entity_id': TASK_GOOD_ID,
|
|
'entity_type': 'chore',
|
|
'user_id': 'testuserid',
|
|
'status': 'pending',
|
|
'approved_at': None,
|
|
'created_at': now_ts,
|
|
'updated_at': now_ts,
|
|
})
|
|
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
assert resp.status_code == 200
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_GOOD_ID]['pending_status'] == 'pending'
|
|
assert tasks[TASK_GOOD_ID]['approved_at'] is None
|
|
|
|
|
|
def test_list_child_tasks_clears_stale_approved_and_pending(client):
|
|
"""Yesterday's chore pending/approved records should be reset and ignored."""
|
|
_setup_sched_child_and_tasks(task_db, child_db)
|
|
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).timestamp()
|
|
old_approved = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat()
|
|
|
|
pending_confirmations_db.insert({
|
|
'id': 'pend_old_chore_pending',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'entity_id': TASK_GOOD_ID,
|
|
'entity_type': 'chore',
|
|
'user_id': 'testuserid',
|
|
'status': 'pending',
|
|
'approved_at': None,
|
|
'created_at': old_ts,
|
|
'updated_at': old_ts,
|
|
})
|
|
|
|
resp = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
assert resp.status_code == 200
|
|
tasks = {t['id']: t for t in resp.get_json()['tasks']}
|
|
assert tasks[TASK_GOOD_ID]['pending_status'] is None
|
|
assert tasks[TASK_GOOD_ID]['approved_at'] is None
|
|
|
|
# Reinsert as stale approved and ensure it is also cleared.
|
|
pending_confirmations_db.insert({
|
|
'id': 'pend_old_chore_approved',
|
|
'child_id': CHILD_SCHED_ID,
|
|
'entity_id': TASK_GOOD_ID,
|
|
'entity_type': 'chore',
|
|
'user_id': 'testuserid',
|
|
'status': 'approved',
|
|
'approved_at': old_approved,
|
|
'created_at': old_ts,
|
|
'updated_at': old_ts,
|
|
})
|
|
|
|
resp2 = client.get(f'/child/{CHILD_SCHED_ID}/list-tasks')
|
|
assert resp2.status_code == 200
|
|
tasks2 = {t['id']: t for t in resp2.get_json()['tasks']}
|
|
assert tasks2[TASK_GOOD_ID]['pending_status'] is None
|
|
assert tasks2[TASK_GOOD_ID]['approved_at'] is None
|
|
|
|
|
|
def test_confirm_chore_allows_when_previous_pending_is_stale(client):
|
|
"""A stale pending chore record from a prior day must not block confirm-chore."""
|
|
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).timestamp()
|
|
task_db.insert({'id': 't_stale_chore', 'name': 'Stale Chore', 'points': 2, 'type': 'chore', 'user_id': 'testuserid'})
|
|
child_db.insert({
|
|
'id': 'child_stale_chore',
|
|
'name': 'Stale Chore Kid',
|
|
'age': 8,
|
|
'points': 0,
|
|
'tasks': ['t_stale_chore'],
|
|
'rewards': [],
|
|
'user_id': 'testuserid',
|
|
})
|
|
pending_confirmations_db.insert({
|
|
'id': 'pend_stale_chore',
|
|
'child_id': 'child_stale_chore',
|
|
'entity_id': 't_stale_chore',
|
|
'entity_type': 'chore',
|
|
'user_id': 'testuserid',
|
|
'status': 'pending',
|
|
'approved_at': None,
|
|
'created_at': old_ts,
|
|
'updated_at': old_ts,
|
|
})
|
|
|
|
resp = client.post('/child/child_stale_chore/confirm-chore', json={'task_id': 't_stale_chore'})
|
|
assert resp.status_code == 200
|
|
|
|
active_pending = pending_confirmations_db.search(
|
|
(Query().child_id == 'child_stale_chore') & (Query().entity_id == 't_stale_chore') &
|
|
(Query().entity_type == 'chore') & (Query().status == 'pending')
|
|
)
|
|
assert len(active_pending) == 1
|
|
assert active_pending[0].get('id') != 'pend_stale_chore'
|
|
|
|
pending_confirmations_db.remove(Query().child_id == 'child_stale_chore')
|
|
child_db.remove(Query().id == 'child_stale_chore')
|
|
task_db.remove(Query().id == 't_stale_chore')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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')
|
|
|
|
|
|
def test_request_reward_allows_new_when_stale_pending_exists(client):
|
|
"""A stale pending reward from a prior day must not block a new request."""
|
|
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).timestamp()
|
|
reward_db.insert({'id': 'r_stale', 'name': 'Stale Reward', 'cost': 5, 'user_id': 'testuserid'})
|
|
child_db.insert({
|
|
'id': 'child_stale',
|
|
'name': 'Stale Kid',
|
|
'age': 8,
|
|
'points': 20,
|
|
'tasks': [],
|
|
'rewards': ['r_stale'],
|
|
'user_id': 'testuserid',
|
|
})
|
|
pending_confirmations_db.insert({
|
|
'id': 'pend_stale_reward',
|
|
'child_id': 'child_stale',
|
|
'entity_id': 'r_stale',
|
|
'entity_type': 'reward',
|
|
'user_id': 'testuserid',
|
|
'status': 'pending',
|
|
'approved_at': None,
|
|
'created_at': old_ts,
|
|
'updated_at': old_ts,
|
|
})
|
|
|
|
resp = client.post('/child/child_stale/request-reward', json={'reward_id': 'r_stale'})
|
|
assert resp.status_code == 200
|
|
|
|
active_pending = pending_confirmations_db.search(
|
|
(Query().child_id == 'child_stale') & (Query().entity_id == 'r_stale') &
|
|
(Query().entity_type == 'reward') & (Query().status == 'pending')
|
|
)
|
|
assert len(active_pending) == 1
|
|
assert active_pending[0].get('id') != 'pend_stale_reward'
|
|
|
|
pending_confirmations_db.remove(Query().child_id == 'child_stale')
|
|
child_db.remove(Query().id == 'child_stale')
|
|
reward_db.remove(Query().id == 'r_stale')
|
|
|
|
|
|
def test_reward_status_ignores_stale_pending_reward(client):
|
|
"""reward-status should not mark a reward as redeeming if pending is stale."""
|
|
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).timestamp()
|
|
reward_db.insert({'id': 'r_status_stale', 'name': 'Status Reward', 'cost': 4, 'user_id': 'testuserid'})
|
|
child_db.insert({
|
|
'id': 'child_status_stale',
|
|
'name': 'Status Kid',
|
|
'age': 9,
|
|
'points': 10,
|
|
'tasks': [],
|
|
'rewards': ['r_status_stale'],
|
|
'user_id': 'testuserid',
|
|
})
|
|
pending_confirmations_db.insert({
|
|
'id': 'pend_status_stale',
|
|
'child_id': 'child_status_stale',
|
|
'entity_id': 'r_status_stale',
|
|
'entity_type': 'reward',
|
|
'user_id': 'testuserid',
|
|
'status': 'pending',
|
|
'approved_at': None,
|
|
'created_at': old_ts,
|
|
'updated_at': old_ts,
|
|
})
|
|
|
|
resp = client.get('/child/child_status_stale/reward-status')
|
|
assert resp.status_code == 200
|
|
statuses = {s['id']: s for s in resp.get_json()['reward_status']}
|
|
assert statuses['r_status_stale']['redeeming'] is False
|
|
assert pending_confirmations_db.get(Query().id == 'pend_status_stale') is None
|
|
|
|
pending_confirmations_db.remove(Query().child_id == 'child_status_stale')
|
|
child_db.remove(Query().id == 'child_status_stale')
|
|
reward_db.remove(Query().id == 'r_status_stale')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 |