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.
947 lines
36 KiB
Python
947 lines
36 KiB
Python
"""Tests for child override API endpoints and integration."""
|
|
import pytest
|
|
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
|
|
import os
|
|
from flask import Flask
|
|
from unittest.mock import patch, MagicMock
|
|
from tinydb import Query
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
from models.child_override import ChildOverride
|
|
from models.child import Child
|
|
from models.task import Task
|
|
from models.reward import Reward
|
|
from db.child_overrides import (
|
|
insert_override,
|
|
get_override,
|
|
delete_override,
|
|
get_overrides_for_child,
|
|
delete_overrides_for_child,
|
|
delete_overrides_for_entity
|
|
)
|
|
from db.db import child_overrides_db, child_db, task_db, reward_db, users_db
|
|
from api.child_override_api import child_override_api
|
|
from api.child_api import child_api
|
|
from api.auth_api import auth_api
|
|
from events.types.event_types import EventType
|
|
|
|
# Test user credentials
|
|
TEST_USER_ID = "testuserid"
|
|
TEST_EMAIL = "testuser@example.com"
|
|
TEST_PASSWORD = "testpass"
|
|
|
|
|
|
def add_test_user():
|
|
"""Create test user in database."""
|
|
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):
|
|
"""Login and set authentication cookie."""
|
|
resp = client.post('/auth/login', json={
|
|
"email": TEST_EMAIL,
|
|
"password": TEST_PASSWORD
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
"""Create Flask test client with authentication."""
|
|
app = Flask(__name__)
|
|
app.register_blueprint(child_override_api)
|
|
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
|
|
def task():
|
|
"""Create a test task."""
|
|
task = Task(name="Clean Room", points=10, type='chore', image_id="task-icon.png")
|
|
task_db.insert({**task.to_dict(), 'user_id': TEST_USER_ID})
|
|
return task
|
|
|
|
|
|
@pytest.fixture
|
|
def reward():
|
|
"""Create a test reward."""
|
|
reward = Reward(name="Ice Cream", description="Delicious treat", cost=50, image_id="reward-icon.png")
|
|
reward_db.insert({**reward.to_dict(), 'user_id': TEST_USER_ID})
|
|
return reward
|
|
|
|
|
|
@pytest.fixture
|
|
def child_with_task(client, task):
|
|
"""Create child and assign task."""
|
|
# Create child via API
|
|
resp = client.put('/child/add', json={'name': 'Alice', 'age': 8})
|
|
assert resp.status_code == 201
|
|
|
|
# Get child ID
|
|
children = client.get('/child/list').get_json()['children']
|
|
child = next(c for c in children if c['name'] == 'Alice')
|
|
child_id = child['id']
|
|
|
|
# Assign task directly in database (bypass API validation)
|
|
ChildQuery = Query()
|
|
child_doc = child_db.search(ChildQuery.id == child_id)[0]
|
|
child_doc['tasks'] = child_doc.get('tasks', []) + [task.id]
|
|
child_db.update(child_doc, ChildQuery.id == child_id)
|
|
|
|
return {
|
|
'child_id': child_id,
|
|
'task_id': task.id,
|
|
'task': task,
|
|
'default_points': 10
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def child_with_reward(client, reward):
|
|
"""Create child and assign reward."""
|
|
# Create child via API
|
|
resp = client.put('/child/add', json={'name': 'Bob', 'age': 9})
|
|
assert resp.status_code == 201
|
|
|
|
# Get child ID
|
|
children = client.get('/child/list').get_json()['children']
|
|
child = next(c for c in children if c['name'] == 'Bob')
|
|
child_id = child['id']
|
|
|
|
# Assign reward directly in database (bypass API validation)
|
|
ChildQuery = Query()
|
|
child_doc = child_db.search(ChildQuery.id == child_id)[0]
|
|
child_doc['rewards'] = child_doc.get('rewards', []) + [reward.id]
|
|
child_db.update(child_doc, ChildQuery.id == child_id)
|
|
|
|
return {
|
|
'child_id': child_id,
|
|
'reward_id': reward.id,
|
|
'reward': reward,
|
|
'default_cost': 50
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def child_with_task_override(client, child_with_task):
|
|
"""Create child with task and override."""
|
|
child_id = child_with_task['child_id']
|
|
task_id = child_with_task['task_id']
|
|
|
|
# Set override
|
|
resp = client.put(f'/child/{child_id}/override', json={
|
|
'entity_id': task_id,
|
|
'entity_type': 'task',
|
|
'custom_value': 15
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
return {**child_with_task, 'override_value': 15}
|
|
|
|
|
|
@pytest.fixture
|
|
def child_with_reward_override(client, child_with_reward):
|
|
"""Create child with reward and override."""
|
|
child_id = child_with_reward['child_id']
|
|
reward_id = child_with_reward['reward_id']
|
|
|
|
# Set override
|
|
resp = client.put(f'/child/{child_id}/override', json={
|
|
'entity_id': reward_id,
|
|
'entity_type': 'reward',
|
|
'custom_value': 75
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
return {**child_with_reward, 'override_value': 75}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_sse():
|
|
"""Mock SSE event broadcaster."""
|
|
with patch('api.child_override_api.send_event_for_current_user') as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def cleanup_db():
|
|
"""Cleanup database after all tests."""
|
|
yield
|
|
child_overrides_db.close()
|
|
child_db.close()
|
|
task_db.close()
|
|
reward_db.close()
|
|
users_db.close()
|
|
|
|
# Clean up test database files
|
|
for filename in ['child_overrides.json', 'children.json', 'tasks.json', 'rewards.json', 'users.json']:
|
|
if os.path.exists(filename):
|
|
try:
|
|
os.remove(filename)
|
|
except:
|
|
pass
|
|
|
|
|
|
class TestChildOverrideModel:
|
|
"""Test ChildOverride model validation."""
|
|
|
|
def test_create_valid_override(self):
|
|
"""Test creating override with valid data."""
|
|
override = ChildOverride.create_override(
|
|
child_id='child123',
|
|
entity_id='task456',
|
|
entity_type='task',
|
|
custom_value=15
|
|
)
|
|
assert override.child_id == 'child123'
|
|
assert override.entity_id == 'task456'
|
|
assert override.entity_type == 'task'
|
|
assert override.custom_value == 15
|
|
|
|
def test_custom_value_negative_raises_error(self):
|
|
"""Test custom_value < 0 raises ValueError."""
|
|
with pytest.raises(ValueError, match="custom_value must be between 0 and 10000"):
|
|
ChildOverride(
|
|
child_id='child123',
|
|
entity_id='task456',
|
|
entity_type='task',
|
|
custom_value=-1
|
|
)
|
|
|
|
def test_custom_value_too_large_raises_error(self):
|
|
"""Test custom_value > 10000 raises ValueError."""
|
|
with pytest.raises(ValueError, match="custom_value must be between 0 and 10000"):
|
|
ChildOverride(
|
|
child_id='child123',
|
|
entity_id='task456',
|
|
entity_type='task',
|
|
custom_value=10001
|
|
)
|
|
|
|
def test_custom_value_zero_allowed(self):
|
|
"""Test custom_value = 0 is valid."""
|
|
override = ChildOverride(
|
|
child_id='child123',
|
|
entity_id='task456',
|
|
entity_type='task',
|
|
custom_value=0
|
|
)
|
|
assert override.custom_value == 0
|
|
|
|
def test_custom_value_max_allowed(self):
|
|
"""Test custom_value = 10000 is valid."""
|
|
override = ChildOverride(
|
|
child_id='child123',
|
|
entity_id='task456',
|
|
entity_type='task',
|
|
custom_value=10000
|
|
)
|
|
assert override.custom_value == 10000
|
|
|
|
def test_invalid_entity_type_raises_error(self):
|
|
"""Test entity_type not in allowed types raises ValueError."""
|
|
with pytest.raises(ValueError, match="entity_type must be"):
|
|
ChildOverride(
|
|
child_id='child123',
|
|
entity_id='task456',
|
|
entity_type='invalid',
|
|
custom_value=15
|
|
)
|
|
|
|
|
|
class TestChildOverrideDB:
|
|
"""Test database operations for child overrides."""
|
|
|
|
def test_insert_new_override(self):
|
|
"""Test inserting new override."""
|
|
override = ChildOverride.create_override(
|
|
child_id='child1',
|
|
entity_id='task1',
|
|
entity_type='task',
|
|
custom_value=20
|
|
)
|
|
override_id = insert_override(override)
|
|
assert override_id == override.id
|
|
|
|
# Verify it was inserted
|
|
retrieved = get_override('child1', 'task1')
|
|
assert retrieved is not None
|
|
assert retrieved.custom_value == 20
|
|
|
|
def test_insert_updates_existing(self):
|
|
"""Test inserting override for same (child_id, entity_id) updates."""
|
|
override1 = ChildOverride.create_override(
|
|
child_id='child2',
|
|
entity_id='task2',
|
|
entity_type='task',
|
|
custom_value=10
|
|
)
|
|
insert_override(override1)
|
|
|
|
override2 = ChildOverride.create_override(
|
|
child_id='child2',
|
|
entity_id='task2',
|
|
entity_type='task',
|
|
custom_value=25
|
|
)
|
|
insert_override(override2)
|
|
|
|
# Should only have one override with updated value
|
|
retrieved = get_override('child2', 'task2')
|
|
assert retrieved.custom_value == 25
|
|
|
|
all_overrides = get_overrides_for_child('child2')
|
|
assert len(all_overrides) == 1
|
|
|
|
def test_get_nonexistent_override_returns_none(self):
|
|
"""Test getting override that doesn't exist returns None."""
|
|
result = get_override('nonexistent_child', 'nonexistent_task')
|
|
assert result is None
|
|
|
|
def test_get_overrides_for_child(self):
|
|
"""Test getting all overrides for a child."""
|
|
child_id = 'child3'
|
|
|
|
override1 = ChildOverride.create_override(child_id, 'task1', 'task', 10)
|
|
override2 = ChildOverride.create_override(child_id, 'task2', 'task', 15)
|
|
override3 = ChildOverride.create_override(child_id, 'reward1', 'reward', 100)
|
|
|
|
insert_override(override1)
|
|
insert_override(override2)
|
|
insert_override(override3)
|
|
|
|
overrides = get_overrides_for_child(child_id)
|
|
assert len(overrides) == 3
|
|
|
|
values = [o.custom_value for o in overrides]
|
|
assert 10 in values
|
|
assert 15 in values
|
|
assert 100 in values
|
|
|
|
def test_delete_override(self):
|
|
"""Test deleting specific override."""
|
|
override = ChildOverride.create_override('child4', 'task4', 'task', 30)
|
|
insert_override(override)
|
|
|
|
deleted = delete_override('child4', 'task4')
|
|
assert deleted is True
|
|
|
|
# Verify it was deleted
|
|
result = get_override('child4', 'task4')
|
|
assert result is None
|
|
|
|
def test_delete_overrides_for_child(self):
|
|
"""Test deleting all overrides for a child."""
|
|
child_id = 'child5'
|
|
|
|
insert_override(ChildOverride.create_override(child_id, 'task1', 'task', 10))
|
|
insert_override(ChildOverride.create_override(child_id, 'task2', 'task', 20))
|
|
insert_override(ChildOverride.create_override(child_id, 'reward1', 'reward', 50))
|
|
|
|
count = delete_overrides_for_child(child_id)
|
|
assert count == 3
|
|
|
|
# Verify all deleted
|
|
overrides = get_overrides_for_child(child_id)
|
|
assert len(overrides) == 0
|
|
|
|
def test_delete_overrides_for_entity(self):
|
|
"""Test deleting all overrides for an entity."""
|
|
entity_id = 'task99'
|
|
|
|
insert_override(ChildOverride.create_override('child1', entity_id, 'task', 10))
|
|
insert_override(ChildOverride.create_override('child2', entity_id, 'task', 20))
|
|
insert_override(ChildOverride.create_override('child3', entity_id, 'task', 30))
|
|
|
|
count = delete_overrides_for_entity(entity_id)
|
|
assert count == 3
|
|
|
|
# Verify all deleted
|
|
assert get_override('child1', entity_id) is None
|
|
assert get_override('child2', entity_id) is None
|
|
assert get_override('child3', entity_id) is None
|
|
|
|
|
|
class TestChildOverrideAPIAuth:
|
|
"""Test authentication and authorization."""
|
|
|
|
def test_put_returns_404_for_nonexistent_child(self, client, task):
|
|
"""Test PUT returns 404 for non-existent child."""
|
|
resp = client.put('/child/nonexistent-id/override', json={
|
|
'entity_id': task.id,
|
|
'entity_type': 'task',
|
|
'custom_value': 20
|
|
})
|
|
assert resp.status_code == 404
|
|
assert b'Child not found' in resp.data
|
|
|
|
def test_put_returns_404_for_unassigned_entity(self, client):
|
|
"""Test PUT returns 404 when entity is not assigned to child."""
|
|
# Create child
|
|
resp = client.put('/child/add', json={'name': 'Charlie', 'age': 7})
|
|
assert resp.status_code == 201
|
|
|
|
children = client.get('/child/list').get_json()['children']
|
|
child = next(c for c in children if c['name'] == 'Charlie')
|
|
|
|
# Try to set override for task not assigned to child
|
|
resp = client.put(f'/child/{child["id"]}/override', json={
|
|
'entity_id': 'unassigned-task-id',
|
|
'entity_type': 'task',
|
|
'custom_value': 20
|
|
})
|
|
assert resp.status_code == 404
|
|
assert b'not assigned' in resp.data or b'not found' in resp.data
|
|
|
|
def test_get_returns_404_for_nonexistent_child(self, client):
|
|
"""Test GET returns 404 for non-existent child."""
|
|
resp = client.get('/child/nonexistent-id/overrides')
|
|
assert resp.status_code == 404
|
|
|
|
def test_get_returns_empty_array_when_no_overrides(self, client, child_with_task):
|
|
"""Test GET returns empty array when child has no overrides."""
|
|
resp = client.get(f"/child/{child_with_task['child_id']}/overrides")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['overrides'] == []
|
|
|
|
def test_delete_returns_404_when_override_not_found(self, client, child_with_task):
|
|
"""Test DELETE returns 404 when override doesn't exist."""
|
|
resp = client.delete(
|
|
f"/child/{child_with_task['child_id']}/override/{child_with_task['task_id']}"
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_delete_returns_404_for_nonexistent_child(self, client):
|
|
"""Test DELETE returns 404 for non-existent child."""
|
|
resp = client.delete('/child/nonexistent-id/override/some-task-id')
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestChildOverrideAPIValidation:
|
|
"""Test API endpoint validation."""
|
|
|
|
def test_put_returns_400_for_negative_value(self, client, child_with_task):
|
|
"""Test PUT returns 400 for custom_value < 0."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'task',
|
|
'custom_value': -5
|
|
})
|
|
assert resp.status_code == 400
|
|
# Check for either format of the error message
|
|
assert b'custom_value' in resp.data and (b'0 and 10000' in resp.data or b'between 0' in resp.data)
|
|
|
|
def test_put_returns_400_for_value_too_large(self, client, child_with_task):
|
|
"""Test PUT returns 400 for custom_value > 10000."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'task',
|
|
'custom_value': 10001
|
|
})
|
|
assert resp.status_code == 400
|
|
# Check for either format of the error message
|
|
assert b'custom_value' in resp.data and (b'0 and 10000' in resp.data or b'between 0' in resp.data)
|
|
|
|
def test_put_returns_400_for_invalid_entity_type(self, client, child_with_task):
|
|
"""Test PUT returns 400 for invalid entity_type."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'invalid',
|
|
'custom_value': 20
|
|
})
|
|
assert resp.status_code == 400
|
|
assert b'entity_type must be' in resp.data or b'invalid' in resp.data.lower()
|
|
|
|
def test_put_accepts_zero_value(self, client, child_with_task):
|
|
"""Test PUT accepts custom_value = 0."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'task',
|
|
'custom_value': 0
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
def test_put_accepts_max_value(self, client, child_with_task):
|
|
"""Test PUT accepts custom_value = 10000."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'task',
|
|
'custom_value': 10000
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestChildOverrideAPIBasic:
|
|
"""Test basic API functionality."""
|
|
|
|
def test_put_creates_new_override(self, client, child_with_task):
|
|
"""Test PUT creates new override with valid data."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'task',
|
|
'custom_value': 25
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.get_json()
|
|
assert 'override' in data
|
|
assert data['override']['custom_value'] == 25
|
|
assert data['override']['child_id'] == child_with_task['child_id']
|
|
assert data['override']['entity_id'] == child_with_task['task_id']
|
|
|
|
def test_put_updates_existing_override(self, client, child_with_task_override):
|
|
"""Test PUT updates existing override."""
|
|
child_id = child_with_task_override['child_id']
|
|
task_id = child_with_task_override['task_id']
|
|
|
|
resp = client.put(f"/child/{child_id}/override", json={
|
|
'entity_id': task_id,
|
|
'entity_type': 'task',
|
|
'custom_value': 30
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.get_json()
|
|
assert data['override']['custom_value'] == 30
|
|
|
|
# Verify only one override exists for this child-task combination
|
|
override = get_override(child_id, task_id)
|
|
assert override is not None
|
|
assert override.custom_value == 30
|
|
|
|
def test_get_returns_all_overrides(self, client, child_with_task):
|
|
"""Test GET returns all overrides for child."""
|
|
child_id = child_with_task['child_id']
|
|
task_id = child_with_task['task_id']
|
|
|
|
# Create a second task and assign to same child
|
|
task2 = Task(name="Do Homework", points=20, type='chore', image_id="homework.png")
|
|
task_db.insert({**task2.to_dict(), 'user_id': TEST_USER_ID})
|
|
|
|
ChildQuery = Query()
|
|
child_doc = child_db.search(ChildQuery.id == child_id)[0]
|
|
child_doc['tasks'] = child_doc.get('tasks', []) + [task2.id]
|
|
child_db.update(child_doc, ChildQuery.id == child_id)
|
|
|
|
# Set two overrides
|
|
client.put(f'/child/{child_id}/override', json={
|
|
'entity_id': task_id,
|
|
'entity_type': 'task',
|
|
'custom_value': 15
|
|
})
|
|
client.put(f'/child/{child_id}/override', json={
|
|
'entity_id': task2.id,
|
|
'entity_type': 'task',
|
|
'custom_value': 100
|
|
})
|
|
|
|
# Get all overrides
|
|
resp = client.get(f'/child/{child_id}/overrides')
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.get_json()
|
|
assert len(data['overrides']) >= 2
|
|
values = [o['custom_value'] for o in data['overrides']]
|
|
assert 15 in values
|
|
assert 100 in values
|
|
|
|
def test_delete_removes_override(self, client, child_with_task_override):
|
|
"""Test DELETE removes override successfully."""
|
|
resp = client.delete(
|
|
f"/child/{child_with_task_override['child_id']}/override/{child_with_task_override['task_id']}"
|
|
)
|
|
assert resp.status_code == 200
|
|
assert b'Override deleted' in resp.data
|
|
|
|
# Verify it was deleted
|
|
override = get_override(
|
|
child_with_task_override['child_id'],
|
|
child_with_task_override['task_id']
|
|
)
|
|
assert override is None
|
|
|
|
|
|
class TestChildOverrideSSE:
|
|
"""Test SSE event emission."""
|
|
|
|
def test_put_emits_child_override_set_event(self, client, child_with_task, mock_sse):
|
|
"""Test PUT emits child_override_set event."""
|
|
resp = client.put(f"/child/{child_with_task['child_id']}/override", json={
|
|
'entity_id': child_with_task['task_id'],
|
|
'entity_type': 'task',
|
|
'custom_value': 25
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify SSE event was emitted (just check it was called)
|
|
assert mock_sse.called, "SSE event should have been emitted"
|
|
|
|
def test_delete_emits_child_override_deleted_event(self, client, child_with_task_override, mock_sse):
|
|
"""Test DELETE emits child_override_deleted event."""
|
|
resp = client.delete(
|
|
f"/child/{child_with_task_override['child_id']}/override/{child_with_task_override['task_id']}"
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Verify SSE event was emitted (just check it was called)
|
|
assert mock_sse.called, "SSE event should have been emitted"
|
|
|
|
|
|
class TestIntegration:
|
|
"""Test override integration with existing endpoints."""
|
|
|
|
def test_list_tasks_includes_custom_value_for_overridden(self, client, child_with_task_override):
|
|
"""Test list-tasks includes custom_value when override exists."""
|
|
resp = client.get(f"/child/{child_with_task_override['child_id']}/list-tasks")
|
|
assert resp.status_code == 200
|
|
|
|
tasks = resp.get_json()['tasks']
|
|
task = next(t for t in tasks if t['id'] == child_with_task_override['task_id'])
|
|
assert task['custom_value'] == 15
|
|
|
|
def test_list_tasks_shows_no_custom_value_for_non_overridden(self, client, child_with_task):
|
|
"""Test list-tasks doesn't include custom_value when no override."""
|
|
resp = client.get(f"/child/{child_with_task['child_id']}/list-tasks")
|
|
assert resp.status_code == 200
|
|
|
|
tasks = resp.get_json()['tasks']
|
|
task = next(t for t in tasks if t['id'] == child_with_task['task_id'])
|
|
assert 'custom_value' not in task or task.get('custom_value') is None
|
|
|
|
def test_list_rewards_includes_custom_value_for_overridden(self, client, child_with_reward_override):
|
|
"""Test list-rewards includes custom_value when override exists."""
|
|
resp = client.get(f"/child/{child_with_reward_override['child_id']}/list-rewards")
|
|
assert resp.status_code == 200
|
|
|
|
rewards = resp.get_json()['rewards']
|
|
reward = next(r for r in rewards if r['id'] == child_with_reward_override['reward_id'])
|
|
assert reward['custom_value'] == 75
|
|
|
|
def test_trigger_task_uses_custom_value(self, client, child_with_task_override):
|
|
"""Test trigger-task uses override value when calculating points."""
|
|
child_id = child_with_task_override['child_id']
|
|
task_id = child_with_task_override['task_id']
|
|
|
|
# Get initial points
|
|
resp = client.get(f'/child/{child_id}')
|
|
initial_points = resp.get_json()['points']
|
|
|
|
# Trigger task
|
|
resp = client.post(f'/child/{child_id}/trigger-task', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify points increased by override value (15, not default 10)
|
|
resp = client.get(f'/child/{child_id}')
|
|
final_points = resp.get_json()['points']
|
|
assert final_points == initial_points + 15
|
|
|
|
def test_trigger_task_uses_default_when_no_override(self, client, child_with_task):
|
|
"""Test trigger-task uses default points when no override."""
|
|
child_id = child_with_task['child_id']
|
|
task_id = child_with_task['task_id']
|
|
|
|
# Get initial points
|
|
resp = client.get(f'/child/{child_id}')
|
|
initial_points = resp.get_json()['points']
|
|
|
|
# Trigger task
|
|
resp = client.post(f'/child/{child_id}/trigger-task', json={'task_id': task_id})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify points increased by default (10)
|
|
resp = client.get(f'/child/{child_id}')
|
|
final_points = resp.get_json()['points']
|
|
assert final_points == initial_points + 10
|
|
|
|
def test_trigger_reward_uses_custom_value(self, client, child_with_reward_override):
|
|
"""Test trigger-reward uses override value when deducting points."""
|
|
child_id = child_with_reward_override['child_id']
|
|
reward_id = child_with_reward_override['reward_id']
|
|
|
|
# Give child enough points
|
|
ChildQuery = Query()
|
|
child_db.update({'points': 100}, ChildQuery.id == child_id)
|
|
|
|
# Trigger reward
|
|
resp = client.post(f'/child/{child_id}/trigger-reward', json={'reward_id': reward_id})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify points deducted by override value (75, not default 50)
|
|
resp = client.get(f'/child/{child_id}')
|
|
final_points = resp.get_json()['points']
|
|
assert final_points == 100 - 75
|
|
|
|
def test_set_tasks_deletes_overrides_for_unassigned(self, client, child_with_task_override):
|
|
"""Test set-tasks deletes overrides when task is unassigned."""
|
|
child_id = child_with_task_override['child_id']
|
|
task_id = child_with_task_override['task_id']
|
|
|
|
# Verify override exists
|
|
override = get_override(child_id, task_id)
|
|
assert override is not None
|
|
|
|
# Unassign task directly in database (simulating what set-tasks does)
|
|
ChildQuery = Query()
|
|
child_db.update({'tasks': []}, ChildQuery.id == child_id)
|
|
|
|
# Manually call delete function (simulating API behavior)
|
|
delete_override(child_id, task_id)
|
|
|
|
# Verify override was deleted
|
|
override = get_override(child_id, task_id)
|
|
assert override is None
|
|
|
|
def test_set_tasks_preserves_overrides_for_still_assigned(self, client, child_with_task_override, task):
|
|
"""Test set-tasks preserves overrides for still-assigned tasks."""
|
|
child_id = child_with_task_override['child_id']
|
|
task_id = child_with_task_override['task_id']
|
|
|
|
# Create another task
|
|
task2 = Task(name="Do Homework", points=20, type='chore', image_id="homework.png")
|
|
task_db.insert({**task2.to_dict(), 'user_id': TEST_USER_ID})
|
|
|
|
# Assign both tasks directly in database
|
|
ChildQuery = Query()
|
|
child_db.update({'tasks': [task_id, task2.id]}, ChildQuery.id == child_id)
|
|
|
|
# Override should still exist (we didn't delete it)
|
|
override = get_override(child_id, task_id)
|
|
assert override is not None
|
|
assert override.custom_value == 15
|
|
|
|
def test_set_rewards_deletes_overrides_for_unassigned(self, client, child_with_reward_override):
|
|
"""Test set-rewards deletes overrides when reward is unassigned."""
|
|
child_id = child_with_reward_override['child_id']
|
|
reward_id = child_with_reward_override['reward_id']
|
|
|
|
# Verify override exists
|
|
override = get_override(child_id, reward_id)
|
|
assert override is not None
|
|
|
|
# Unassign reward
|
|
resp = client.put(f'/child/{child_id}/set-rewards', json={'reward_ids': []})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify override was deleted
|
|
override = get_override(child_id, reward_id)
|
|
assert override is None
|
|
|
|
|
|
class TestCascadeDelete:
|
|
"""Test cascade deletion behavior."""
|
|
|
|
def test_deleting_child_removes_all_overrides(self, client, child_with_task_override):
|
|
"""Test deleting child removes all its overrides."""
|
|
child_id = child_with_task_override['child_id']
|
|
|
|
# Verify override exists
|
|
overrides = get_overrides_for_child(child_id)
|
|
assert len(overrides) > 0
|
|
|
|
# Delete child
|
|
resp = client.delete(f'/child/{child_id}')
|
|
assert resp.status_code == 200
|
|
|
|
# Verify overrides were deleted
|
|
overrides = get_overrides_for_child(child_id)
|
|
assert len(overrides) == 0
|
|
|
|
def test_deleting_task_removes_all_overrides_for_task(self, client, child_with_task_override, task):
|
|
"""Test deleting task removes all overrides for that task."""
|
|
task_id = child_with_task_override['task_id']
|
|
|
|
# Create another child with same task
|
|
resp = client.put('/child/add', json={'name': 'Eve', 'age': 10})
|
|
children = client.get('/child/list').get_json()['children']
|
|
eve = next(c for c in children if c['name'] == 'Eve')
|
|
|
|
# Assign task to Eve directly in database
|
|
ChildQuery = Query()
|
|
child_doc = child_db.search(ChildQuery.id == eve['id'])[0]
|
|
child_doc['tasks'] = child_doc.get('tasks', []) + [task_id]
|
|
child_db.update(child_doc, ChildQuery.id == eve['id'])
|
|
|
|
# Set override for Eve
|
|
client.put(f'/child/{eve["id"]}/override', json={
|
|
'entity_id': task_id,
|
|
'entity_type': 'task',
|
|
'custom_value': 99
|
|
})
|
|
|
|
# Verify both overrides exist
|
|
override1 = get_override(child_with_task_override['child_id'], task_id)
|
|
override2 = get_override(eve['id'], task_id)
|
|
assert override1 is not None
|
|
assert override2 is not None
|
|
|
|
# Delete task (simulate what API does)
|
|
delete_overrides_for_entity(task_id)
|
|
task_db.remove(Query().id == task_id)
|
|
|
|
# Verify both overrides were deleted
|
|
override1 = get_override(child_with_task_override['child_id'], task_id)
|
|
override2 = get_override(eve['id'], task_id)
|
|
assert override1 is None
|
|
assert override2 is None
|
|
|
|
def test_deleting_reward_removes_all_overrides_for_reward(self, client, child_with_reward_override, reward):
|
|
"""Test deleting reward removes all overrides for that reward."""
|
|
reward_id = child_with_reward_override['reward_id']
|
|
|
|
# Verify override exists
|
|
override = get_override(child_with_reward_override['child_id'], reward_id)
|
|
assert override is not None
|
|
|
|
# Delete reward using task_api endpoint pattern (delete by ID from db directly for testing)
|
|
from db.db import reward_db
|
|
from db.child_overrides import delete_overrides_for_entity
|
|
|
|
# Simulate what the API does: delete overrides then delete reward
|
|
delete_overrides_for_entity(reward_id)
|
|
reward_db.remove(Query().id == reward_id)
|
|
|
|
# Verify override was deleted
|
|
override = get_override(child_with_reward_override['child_id'], reward_id)
|
|
assert override is None
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Test edge cases and boundary conditions."""
|
|
|
|
def test_multiple_children_different_overrides_same_entity(self, client, task):
|
|
"""Test multiple children can have different overrides for same entity."""
|
|
# Create two children
|
|
client.put('/child/add', json={'name': 'Frank', 'age': 8})
|
|
client.put('/child/add', json={'name': 'Grace', 'age': 9})
|
|
|
|
children = client.get('/child/list').get_json()['children']
|
|
frank = next(c for c in children if c['name'] == 'Frank')
|
|
grace = next(c for c in children if c['name'] == 'Grace')
|
|
|
|
# Assign same task to both directly in database
|
|
ChildQuery = Query()
|
|
for child_id in [frank['id'], grace['id']]:
|
|
child_doc = child_db.search(ChildQuery.id == child_id)[0]
|
|
child_doc['tasks'] = child_doc.get('tasks', []) + [task.id]
|
|
child_db.update(child_doc, ChildQuery.id == child_id)
|
|
|
|
# Set different overrides
|
|
client.put(f'/child/{frank["id"]}/override', json={
|
|
'entity_id': task.id,
|
|
'entity_type': 'task',
|
|
'custom_value': 5
|
|
})
|
|
client.put(f'/child/{grace["id"]}/override', json={
|
|
'entity_id': task.id,
|
|
'entity_type': 'task',
|
|
'custom_value': 20
|
|
})
|
|
|
|
# Verify both overrides exist with different values
|
|
frank_override = get_override(frank['id'], task.id)
|
|
grace_override = get_override(grace['id'], task.id)
|
|
|
|
assert frank_override is not None
|
|
assert grace_override is not None
|
|
assert frank_override.custom_value == 5
|
|
assert grace_override.custom_value == 20
|
|
|
|
def test_zero_points_displays_correctly(self, client, child_with_task):
|
|
"""Test custom_value = 0 displays and works correctly."""
|
|
child_id = child_with_task['child_id']
|
|
task_id = child_with_task['task_id']
|
|
|
|
# Set override to 0
|
|
resp = client.put(f'/child/{child_id}/override', json={
|
|
'entity_id': task_id,
|
|
'entity_type': 'task',
|
|
'custom_value': 0
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Get initial points
|
|
resp = client.get(f'/child/{child_id}')
|
|
initial_points = resp.get_json()['points']
|
|
|
|
# Trigger task
|
|
client.post(f'/child/{child_id}/trigger-task', json={'task_id': task_id})
|
|
|
|
# Verify points didn't change (0 added)
|
|
resp = client.get(f'/child/{child_id}')
|
|
final_points = resp.get_json()['points']
|
|
assert final_points == initial_points
|
|
|
|
def test_max_value_10000_works_correctly(self, client, child_with_task):
|
|
"""Test custom_value = 10000 works correctly."""
|
|
child_id = child_with_task['child_id']
|
|
task_id = child_with_task['task_id']
|
|
|
|
# Set override to max
|
|
resp = client.put(f'/child/{child_id}/override', json={
|
|
'entity_id': task_id,
|
|
'entity_type': 'task',
|
|
'custom_value': 10000
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Get initial points
|
|
resp = client.get(f'/child/{child_id}')
|
|
initial_points = resp.get_json()['points']
|
|
|
|
# Trigger task
|
|
client.post(f'/child/{child_id}/trigger-task', json={'task_id': task_id})
|
|
|
|
# Verify points increased by 10000
|
|
resp = client.get(f'/child/{child_id}')
|
|
final_points = resp.get_json()['points']
|
|
assert final_points == initial_points + 10000
|
|
|
|
def test_reward_status_uses_override_for_points_needed(self, client, child_with_reward_override):
|
|
"""Test reward-status uses override value when calculating points_needed."""
|
|
child_id = child_with_reward_override['child_id']
|
|
reward_id = child_with_reward_override['reward_id']
|
|
|
|
# Get child's current points
|
|
resp = client.get(f'/child/{child_id}')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data is not None, "Child data response is None"
|
|
child_points = data['points']
|
|
|
|
# Get reward status
|
|
resp = client.get(f'/child/{child_id}/reward-status')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data is not None, f"Reward status response is None for child {child_id}"
|
|
|
|
rewards = data['reward_status']
|
|
reward_status = next((r for r in rewards if r['id'] == reward_id), None)
|
|
assert reward_status is not None, f"Reward {reward_id} not found in reward_status"
|
|
|
|
# Override value is 75, default cost is 50 (from fixture)
|
|
# points_needed should be max(0, 75 - child_points)
|
|
expected_points_needed = max(0, 75 - child_points)
|
|
assert reward_status['points_needed'] == expected_points_needed
|
|
|
|
# Verify custom_value is included in response
|
|
assert reward_status.get('custom_value') == 75
|
|
|