Add routine management features for child and parent views
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 3m8s

- Implemented routine child-mode flow tests to ensure proper functionality of routine assignment and task completion.
- Created notification tests for parent view to verify routine completion notifications for children.
- Developed ChildRoutineOverlay component for displaying routine tasks and handling user interactions.
- Added RoutineApproveDialog component for approving or rejecting completed routines.
- Created unit tests for ChildRoutineOverlay and RoutineEditView components to ensure correct behavior and rendering.
- Enhanced RoutineEditView with proper handling of task addition and form submission.

Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
2026-05-17 23:47:12 -04:00
parent eb775ba7d8
commit 5392e5af70
31 changed files with 3859 additions and 265 deletions

View File

@@ -9,12 +9,14 @@ from datetime import datetime, timezone
from tinydb import Query
from db.db import child_db, task_db, reward_db, pending_confirmations_db
from db.db import child_db, task_db, reward_db, pending_confirmations_db, routine_db
from db.child_overrides import get_override
from db.chore_schedules import get_schedule
from db.routine_schedules import get_schedule as get_routine_schedule
from db.tracking import insert_tracking_event
from events.sse import send_event_to_user
from events.types.child_chore_confirmation import ChildChoreConfirmation
from events.types.child_routine_confirmation import ChildRoutineConfirmation
from events.types.child_reward_request import ChildRewardRequest
from events.types.child_reward_triggered import ChildRewardTriggered
from events.types.child_task_triggered import ChildTaskTriggered
@@ -23,6 +25,7 @@ from events.types.event import Event
from events.types.event_types import EventType
from models.child import Child
from models.reward import Reward
from models.routine import Routine
from models.task import Task
from models.tracking_event import TrackingEvent
from utils.tracking_logger import log_tracking_event
@@ -271,3 +274,84 @@ def deny_reward(user_id: str, child_id: str, reward_id: str) -> dict | None:
ChildRewardRequest(child_id, reward_id, ChildRewardRequest.REQUEST_CANCELLED)))
return {'child_name': child_name}
def approve_routine(user_id: str, child_id: str, routine_id: str) -> dict | None:
"""Award points for a completed routine and mark the pending confirmation approved.
Returns a result dict on success, or None if already resolved.
Raises ValueError if the child or routine cannot be found.
"""
ChildQ = Query()
child_result = child_db.get((ChildQ.id == child_id) & (ChildQ.user_id == user_id))
if not child_result:
raise ValueError(f'Child {child_id} not found for user {user_id}')
child = Child.from_dict(child_result)
if routine_id not in child.routines:
logger.info(f'Routine {routine_id} no longer assigned to child {child_id}; skipping approve')
return None
PendingQ = Query()
existing = pending_confirmations_db.get(
(PendingQ.child_id == child_id) & (PendingQ.entity_id == routine_id) &
(PendingQ.entity_type == 'routine') & (PendingQ.status == 'pending') &
(PendingQ.user_id == user_id)
)
if not existing:
logger.info(f'No pending routine for child {child_id}, routine {routine_id} — already resolved')
return None
RoutineQ = Query()
routine_result = routine_db.get(
(RoutineQ.id == routine_id) & ((RoutineQ.user_id == user_id) | (RoutineQ.user_id == None))
)
if not routine_result:
raise ValueError(f'Routine {routine_id} not found')
routine = Routine.from_dict(routine_result)
override = get_override(child_id, routine_id)
points_value = override.custom_value if override and override.entity_type == 'routine' else routine.points
points_before = child.points
child.points += points_value
child_db.update({'points': child.points}, ChildQ.id == child_id)
schedule = get_routine_schedule(child_id, routine_id)
now_str = datetime.now(timezone.utc).isoformat()
if schedule:
pending_confirmations_db.update(
{'status': 'approved', 'approved_at': now_str},
(PendingQ.child_id == child_id) & (PendingQ.entity_id == routine_id) &
(PendingQ.entity_type == 'routine') & (PendingQ.user_id == user_id)
)
else:
pending_confirmations_db.remove(
(PendingQ.child_id == child_id) & (PendingQ.entity_id == routine_id) &
(PendingQ.entity_type == 'routine') & (PendingQ.user_id == user_id)
)
send_event_to_user(user_id, Event(EventType.CHILD_ROUTINE_CONFIRMATION.value,
ChildRoutineConfirmation(child_id, routine_id, ChildRoutineConfirmation.OPERATION_APPROVED)))
return {'routine_name': routine.name, 'child_name': child.name, 'child_id': child_id, 'points': child.points}
def reject_routine(user_id: str, child_id: str, routine_id: str) -> None:
"""Reject a pending routine confirmation. No-op if already resolved."""
PendingQ = Query()
existing = pending_confirmations_db.get(
(PendingQ.child_id == child_id) & (PendingQ.entity_id == routine_id) &
(PendingQ.entity_type == 'routine') & (PendingQ.status == 'pending') &
(PendingQ.user_id == user_id)
)
if not existing:
logger.info(f'No pending routine for child {child_id}, routine {routine_id} — already resolved')
return
pending_confirmations_db.remove(
(PendingQ.child_id == child_id) & (PendingQ.entity_id == routine_id) &
(PendingQ.entity_type == 'routine') & (PendingQ.user_id == user_id)
)
send_event_to_user(user_id, Event(EventType.CHILD_ROUTINE_CONFIRMATION.value,
ChildRoutineConfirmation(child_id, routine_id, ChildRoutineConfirmation.OPERATION_REJECTED)))

View File

@@ -1,5 +1,6 @@
from time import sleep
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from flask import Blueprint, request, jsonify
from tinydb import Query
@@ -43,15 +44,49 @@ child_api = Blueprint('child_api', __name__)
logger = logging.getLogger(__name__)
def _is_iso_timestamp_today_utc(timestamp: str | None, today_utc: str) -> bool:
return bool(timestamp) and timestamp[:10] == today_utc
def _get_user_timezone(user_id: str) -> str | None:
user = users_db.get(Query().id == user_id)
if not user:
return None
return user.get('timezone')
def _is_epoch_timestamp_today_utc(epoch_ts, today_utc: str) -> bool:
def _get_user_today_local(user_id: str) -> tuple[str, str | None]:
tz_str = _get_user_timezone(user_id)
try:
now_local = datetime.now(ZoneInfo(tz_str)) if tz_str else datetime.now(timezone.utc)
except Exception:
tz_str = None
now_local = datetime.now(timezone.utc)
return now_local.strftime('%Y-%m-%d'), tz_str
def _is_iso_timestamp_on_local_day(timestamp: str | None, local_day: str, tz_str: str | None) -> bool:
if not timestamp:
return False
try:
normalized = timestamp.replace('Z', '+00:00')
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
try:
tz = ZoneInfo(tz_str) if tz_str else timezone.utc
except Exception:
tz = timezone.utc
return parsed.astimezone(tz).strftime('%Y-%m-%d') == local_day
except (TypeError, ValueError):
return False
def _is_epoch_timestamp_on_local_day(epoch_ts, local_day: str, tz_str: str | None) -> bool:
if epoch_ts is None:
return False
try:
return datetime.fromtimestamp(float(epoch_ts), timezone.utc).strftime('%Y-%m-%d') == today_utc
tz = ZoneInfo(tz_str) if tz_str else timezone.utc
except Exception:
tz = timezone.utc
try:
return datetime.fromtimestamp(float(epoch_ts), tz).strftime('%Y-%m-%d') == local_day
except (TypeError, ValueError, OSError):
return False
@@ -305,6 +340,7 @@ def list_child_tasks(id):
task_ids = child.get('tasks', [])
TaskQuery = Query()
today_local, tz_str = _get_user_today_local(user_id)
child_tasks = []
for tid in task_ids:
task = task_db.get((TaskQuery.id == tid) & ((TaskQuery.user_id == user_id) | (TaskQuery.user_id == None)))
@@ -339,12 +375,10 @@ def list_child_tasks(id):
status = pending.get('status')
approved_at = pending.get('approved_at')
created_at = pending.get('created_at')
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
if status == 'approved' and _is_iso_timestamp_today_utc(approved_at, today_utc):
if status == 'approved' and _is_iso_timestamp_on_local_day(approved_at, today_local, tz_str):
ct_dict['pending_status'] = 'approved'
ct_dict['approved_at'] = approved_at
elif status == 'pending' and _is_epoch_timestamp_today_utc(created_at, today_utc):
elif status == 'pending' and _is_epoch_timestamp_on_local_day(created_at, today_local, tz_str):
ct_dict['pending_status'] = 'pending'
ct_dict['approved_at'] = None
else:
@@ -892,6 +926,7 @@ def reward_status(id):
reward_ids = child.rewards
RewardQuery = Query()
today_local, tz_str = _get_user_today_local(user_id)
statuses = []
for reward_id in reward_ids:
reward_dict = reward_db.get((RewardQuery.id == reward_id) & ((RewardQuery.user_id == user_id) | (RewardQuery.user_id == None)))
@@ -912,8 +947,7 @@ def reward_status(id):
)
redeeming = False
if pending and pending.get('status') == 'pending':
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
if _is_epoch_timestamp_today_utc(pending.get('created_at'), today_utc):
if _is_epoch_timestamp_on_local_day(pending.get('created_at'), today_local, tz_str):
redeeming = True
else:
pending_id = pending.get('id')
@@ -975,8 +1009,8 @@ def request_reward(id):
(DupQuery.user_id == user_id)
)
if duplicate:
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
if _is_epoch_timestamp_today_utc(duplicate.get('created_at'), today_utc):
today_local, tz_str = _get_user_today_local(user_id)
if _is_epoch_timestamp_on_local_day(duplicate.get('created_at'), today_local, tz_str):
return jsonify({'error': 'Reward request already pending', 'code': 'DUPLICATE_REWARD_REQUEST'}), 409
pending_id = duplicate.get('id')
if pending_id:
@@ -1106,7 +1140,7 @@ def list_pending_confirmations():
if not user_id:
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
PendingQuery = Query()
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
today_local, tz_str = _get_user_today_local(user_id)
pending_items = pending_confirmations_db.search(
(PendingQuery.user_id == user_id) & (PendingQuery.status == 'pending')
)
@@ -1120,7 +1154,7 @@ def list_pending_confirmations():
for pr in pending_items:
pending = PendingConfirmation.from_dict(pr)
if not _is_epoch_timestamp_today_utc(pending.created_at, today_utc):
if not _is_epoch_timestamp_on_local_day(pending.created_at, today_local, tz_str):
pending_confirmations_db.remove(PendingQuery.id == pending.id)
continue
@@ -1200,16 +1234,16 @@ def confirm_chore(id):
(PendingQuery.entity_type == 'chore') & (PendingQuery.user_id == user_id)
)
if existing:
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
today_local, tz_str = _get_user_today_local(user_id)
if existing.get('status') == 'pending':
if _is_epoch_timestamp_today_utc(existing.get('created_at'), today_utc):
if _is_epoch_timestamp_on_local_day(existing.get('created_at'), today_local, tz_str):
return jsonify({'error': 'Chore already pending confirmation', 'code': 'CHORE_ALREADY_PENDING'}), 400
pending_id = existing.get('id')
if pending_id:
pending_confirmations_db.remove(PendingQuery.id == pending_id)
if existing.get('status') == 'approved':
approved_at = existing.get('approved_at', '')
if _is_iso_timestamp_today_utc(approved_at, today_utc):
if _is_iso_timestamp_on_local_day(approved_at, today_local, tz_str):
return jsonify({'error': 'Chore already completed today', 'code': 'CHORE_ALREADY_COMPLETED'}), 400
pending_id = existing.get('id')
if pending_id:

View File

@@ -1,5 +1,6 @@
from collections import defaultdict
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from flask import Blueprint, request, jsonify
from tinydb import Query
@@ -24,15 +25,49 @@ from utils.push_sender import send_push_to_user
child_routine_api = Blueprint('child_routine_api', __name__)
def _is_iso_timestamp_today_utc(timestamp: str | None, today_utc: str) -> bool:
return bool(timestamp) and timestamp[:10] == today_utc
def _get_user_timezone(user_id: str) -> str | None:
user = users_db.get(Query().id == user_id)
if not user:
return None
return user.get('timezone')
def _is_epoch_timestamp_today_utc(epoch_ts, today_utc: str) -> bool:
def _get_user_today_local(user_id: str) -> tuple[str, str | None]:
tz_str = _get_user_timezone(user_id)
try:
now_local = datetime.now(ZoneInfo(tz_str)) if tz_str else datetime.now(timezone.utc)
except Exception:
tz_str = None
now_local = datetime.now(timezone.utc)
return now_local.strftime('%Y-%m-%d'), tz_str
def _is_iso_timestamp_on_local_day(timestamp: str | None, local_day: str, tz_str: str | None) -> bool:
if not timestamp:
return False
try:
normalized = timestamp.replace('Z', '+00:00')
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
try:
tz = ZoneInfo(tz_str) if tz_str else timezone.utc
except Exception:
tz = timezone.utc
return parsed.astimezone(tz).strftime('%Y-%m-%d') == local_day
except (TypeError, ValueError):
return False
def _is_epoch_timestamp_on_local_day(epoch_ts, local_day: str, tz_str: str | None) -> bool:
if epoch_ts is None:
return False
try:
return datetime.fromtimestamp(float(epoch_ts), timezone.utc).strftime('%Y-%m-%d') == today_utc
tz = ZoneInfo(tz_str) if tz_str else timezone.utc
except Exception:
tz = timezone.utc
try:
return datetime.fromtimestamp(float(epoch_ts), tz).strftime('%Y-%m-%d') == local_day
except (TypeError, ValueError, OSError):
return False
@@ -201,7 +236,7 @@ def list_child_routines(id):
routine_q = Query()
pending_q = Query()
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
today_local, tz_str = _get_user_today_local(user_id)
child_routines = []
for rid in child.routines:
@@ -235,22 +270,27 @@ def list_child_routines(id):
status = pending.get('status')
approved_at = pending.get('approved_at')
created_at = pending.get('created_at')
confirmation_id = pending.get('id')
if status == 'approved' and _is_iso_timestamp_today_utc(approved_at, today_utc):
if status == 'approved' and _is_iso_timestamp_on_local_day(approved_at, today_local, tz_str):
cr_dict['pending_status'] = 'approved'
cr_dict['approved_at'] = approved_at
elif status == 'pending' and _is_epoch_timestamp_today_utc(created_at, today_utc):
cr_dict['pending_confirmation_id'] = confirmation_id
elif status == 'pending' and _is_epoch_timestamp_on_local_day(created_at, today_local, tz_str):
cr_dict['pending_status'] = 'pending'
cr_dict['approved_at'] = None
cr_dict['pending_confirmation_id'] = confirmation_id
else:
pending_id = pending.get('id')
if pending_id:
pending_confirmations_db.remove(pending_q.id == pending_id)
cr_dict['pending_status'] = None
cr_dict['approved_at'] = None
cr_dict['pending_confirmation_id'] = None
else:
cr_dict['pending_status'] = None
cr_dict['approved_at'] = None
cr_dict['pending_confirmation_id'] = None
child_routines.append(cr_dict)
@@ -319,16 +359,16 @@ def confirm_routine(id):
(pending_q.entity_type == 'routine') & (pending_q.user_id == user_id)
)
if existing:
today_utc = datetime.now(timezone.utc).strftime('%Y-%m-%d')
today_local, tz_str = _get_user_today_local(user_id)
if existing.get('status') == 'pending':
if _is_epoch_timestamp_today_utc(existing.get('created_at'), today_utc):
if _is_epoch_timestamp_on_local_day(existing.get('created_at'), today_local, tz_str):
return jsonify({'error': 'Routine already pending confirmation', 'code': 'ROUTINE_ALREADY_PENDING'}), 400
pending_id = existing.get('id')
if pending_id:
pending_confirmations_db.remove(pending_q.id == pending_id)
if existing.get('status') == 'approved':
approved_at = existing.get('approved_at', '')
if _is_iso_timestamp_today_utc(approved_at, today_utc):
if _is_iso_timestamp_on_local_day(approved_at, today_local, tz_str):
return jsonify({'error': 'Routine already completed today', 'code': 'ROUTINE_ALREADY_COMPLETED'}), 400
pending_id = existing.get('id')
if pending_id:
@@ -526,3 +566,73 @@ def reset_routine(id, confirmation_id):
)
)
return jsonify({'message': 'Routine reset to available.'}), 200
@child_routine_api.route('/child/<id>/trigger-routine', methods=['POST'])
def trigger_child_routine(id):
"""Parent-triggered routine confirmation — directly awards points."""
user_id = get_validated_user_id()
if not user_id:
return jsonify({'error': 'Unauthorized', 'code': ErrorCodes.UNAUTHORIZED}), 401
data = request.get_json() or {}
routine_id = data.get('routine_id')
if not routine_id:
return jsonify({'error': 'routine_id is required', 'code': ErrorCodes.MISSING_FIELD}), 400
child = _validate_child_for_user(id, user_id)
if not child:
return jsonify({'error': 'Child not found', 'code': ErrorCodes.CHILD_NOT_FOUND}), 404
if routine_id not in child.routines:
return jsonify({'error': 'Routine not assigned to child', 'code': ErrorCodes.ENTITY_NOT_ASSIGNED}), 400
routine = _resolve_routine_for_user(routine_id, user_id)
if not routine:
return jsonify({'error': 'Routine not found'}), 404
# Check for override
override = get_override(id, routine_id)
points_value = override.custom_value if override and override.entity_type == 'routine' else routine.points
# Award points
new_points = max(0, child.points + points_value)
child_db.update({'points': new_points}, Query().id == id)
# Create an approved pending confirmation so it shows as completed in the routine list
pending_q = Query()
existing = pending_confirmations_db.get(
(pending_q.child_id == id) & (pending_q.entity_id == routine_id) &
(pending_q.entity_type == 'routine') & (pending_q.user_id == user_id)
)
if existing:
today_local, tz_str = _get_user_today_local(user_id)
# Remove old confirmation if it exists and is not from today
if existing.get('status') == 'approved' and _is_iso_timestamp_on_local_day(existing.get('approved_at'), today_local, tz_str):
return jsonify({'error': 'Routine already completed today', 'code': 'ROUTINE_ALREADY_COMPLETED'}), 400
pending_id = existing.get('id')
if pending_id:
pending_confirmations_db.remove(pending_q.id == pending_id)
confirmation = PendingConfirmation(
child_id=id,
entity_id=routine_id,
entity_type='routine',
user_id=user_id,
status='approved',
approved_at=datetime.now(timezone.utc).isoformat()
)
pending_confirmations_db.insert(confirmation.to_dict())
send_event_for_current_user(
Event(
EventType.CHILD_ROUTINE_CONFIRMATION.value,
ChildRoutineConfirmation(id, routine_id, ChildRoutineConfirmation.OPERATION_APPROVED)
)
)
return jsonify({
'message': f'Routine {routine.name} awarded to {child.name}.',
'points': new_points,
'id': child.id,
}), 200

View File

@@ -5,7 +5,7 @@ from tinydb import Query
from utils.digest_token import validate_and_consume_token, validate_unsubscribe_token, peek_token
from db.db import users_db
from api.child_action_helpers import approve_chore, reject_chore, approve_reward_request, deny_reward
from api.child_action_helpers import approve_chore, reject_chore, approve_reward_request, deny_reward, approve_routine, reject_routine
from api.utils import get_validated_user_id
digest_action_api = Blueprint('digest_action_api', __name__)
@@ -79,6 +79,10 @@ def execute_digest_action(token_id: str):
approve_reward_request(user_id, token.child_id, token.entity_id)
elif token.entity_type == 'reward' and token.action == 'deny':
deny_reward(user_id, token.child_id, token.entity_id)
elif token.entity_type == 'routine' and token.action == 'approve':
approve_routine(user_id, token.child_id, token.entity_id)
elif token.entity_type == 'routine' and token.action == 'deny':
reject_routine(user_id, token.child_id, token.entity_id)
else:
return jsonify({'error': 'Unknown action', 'code': 'INVALID_ACTION'}), 400
except Exception as e:

View File

@@ -4,6 +4,7 @@ import os
from flask import Flask
from api.child_api import child_api
import api.child_api as child_api_module
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
@@ -585,6 +586,39 @@ def test_list_child_tasks_clears_stale_approved_and_pending(client):
assert tasks2[TASK_GOOD_ID]['approved_at'] is None
def test_list_child_tasks_local_day_keeps_approved_across_utc_rollover(client, monkeypatch):
"""Approved chore should remain completed when UTC date differs but user-local day matches.
Example: 2026-05-11T22:30:00Z is 2026-05-12 local day in Pacific/Kiritimati (UTC+14).
"""
_setup_sched_child_and_tasks(task_db, child_db)
# Force deterministic local-day basis for this endpoint call.
monkeypatch.setattr(
child_api_module,
'_get_user_today_local',
lambda user_id: ('2026-05-12', 'Pacific/Kiritimati'),
)
pending_confirmations_db.insert({
'id': 'pend_local_day_approved',
'child_id': CHILD_SCHED_ID,
'entity_id': TASK_GOOD_ID,
'entity_type': 'chore',
'user_id': 'testuserid',
'status': 'approved',
'approved_at': '2026-05-11T22:30:00+00:00',
'created_at': 1778538600,
'updated_at': 1778538600,
})
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'] == 'approved'
assert tasks[TASK_GOOD_ID]['approved_at'] == '2026-05-11T22:30:00+00:00'
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()

View File

@@ -3,6 +3,7 @@ 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
import api.child_api as child_api_module
from flask import Flask
from api.child_api import child_api
@@ -71,6 +72,24 @@ def setup_child_and_chore(child_name='TestChild', age=8, chore_points=10):
return child['id'], 'chore1'
def test_local_day_iso_check_handles_utc_rollover():
# 22:30 UTC on 2026-05-11 is 12:30 local on 2026-05-12 in Pacific/Kiritimati (UTC+14).
assert child_api_module._is_iso_timestamp_on_local_day(
'2026-05-11T22:30:00+00:00',
'2026-05-12',
'Pacific/Kiritimati',
)
def test_local_day_epoch_check_handles_utc_rollover():
# Same instant as above represented as epoch seconds.
assert child_api_module._is_epoch_timestamp_on_local_day(
1778538600,
'2026-05-12',
'Pacific/Kiritimati',
)
# ---------------------------------------------------------------------------
# Child Confirm Flow
# ---------------------------------------------------------------------------

View File

@@ -0,0 +1,296 @@
"""Unit tests for routine API endpoints."""
import pytest
from datetime import datetime, timezone
from tinydb import Query
from models.routine import Routine
from models.routine_item import RoutineItem
from models.routine_schedule import RoutineSchedule
from models.routine_extension import RoutineExtension
from models.pending_confirmation import PendingConfirmation
from api.error_codes import ErrorCodes
from db.db import routine_db, routine_items_db, routine_schedules_db, routine_extensions_db, pending_confirmations_db
class TestRoutineModel:
"""Test Routine model serialization and validation."""
def test_routine_creation(self):
"""Test creating a routine instance."""
routine = Routine(name="Morning Routine", points=50, image_id="img123", user_id="user1")
assert routine.name == "Morning Routine"
assert routine.points == 50
assert routine.image_id == "img123"
assert routine.user_id == "user1"
assert routine.id is not None
def test_routine_to_dict(self):
"""Test routine serialization."""
routine = Routine(name="Test", points=10, image_id=None, user_id="user1")
data = routine.to_dict()
assert data["name"] == "Test"
assert data["points"] == 10
assert data["id"] == routine.id
def test_routine_from_dict(self):
"""Test routine deserialization."""
routine_dict = {"id": "r1", "name": "Test", "points": 20, "image_id": "img1", "user_id": "u1"}
routine = Routine.from_dict(routine_dict)
assert routine.id == "r1"
assert routine.name == "Test"
assert routine.points == 20
class TestRoutineItemModel:
"""Test RoutineItem model."""
def test_routine_item_creation(self):
"""Test creating a routine item."""
item = RoutineItem(routine_id="r1", name="Make Bed", image_id=None, order=0)
assert item.routine_id == "r1"
assert item.name == "Make Bed"
assert item.order == 0
def test_routine_item_to_dict(self):
"""Test item serialization."""
item = RoutineItem(routine_id="r1", name="Get Dressed", image_id="img1", order=1)
data = item.to_dict()
assert data["routine_id"] == "r1"
assert data["name"] == "Get Dressed"
assert data["order"] == 1
class TestRoutineScheduleModel:
"""Test RoutineSchedule model."""
def test_days_mode_schedule_creation(self):
"""Test creating days mode schedule."""
day_configs = [
{"day": 0, "hour": 8, "minute": 0},
{"day": 1, "hour": 9, "minute": 30},
]
schedule = RoutineSchedule(
child_id="c1",
routine_id="r1",
mode="days",
enabled=True,
day_configs=day_configs,
default_hour=8,
default_minute=0,
default_has_deadline=True,
)
assert schedule.mode == "days"
assert len(schedule.day_configs) == 2
assert schedule.enabled is True
def test_interval_mode_schedule_creation(self):
"""Test creating interval mode schedule."""
schedule = RoutineSchedule(
child_id="c1",
routine_id="r1",
mode="interval",
enabled=True,
interval_days=3,
anchor_date="2026-05-01",
interval_has_deadline=True,
interval_hour=10,
interval_minute=30,
)
assert schedule.mode == "interval"
assert schedule.interval_days == 3
assert schedule.interval_hour == 10
class TestRoutineExtensionModel:
"""Test RoutineExtension model."""
def test_extension_creation(self):
"""Test creating a routine extension."""
extension = RoutineExtension(child_id="c1", routine_id="r1", date="2026-05-10")
assert extension.child_id == "c1"
assert extension.routine_id == "r1"
assert extension.date == "2026-05-10"
class TestRoutineDB:
"""Test routine database operations."""
def setup_method(self):
"""Clear routine db before each test."""
routine_db.truncate()
def test_add_routine(self):
"""Test adding a routine to database."""
routine = Routine(name="Test Routine", points=50, image_id=None, user_id="user1")
routine_db.insert(routine.to_dict())
result = routine_db.search(Query().id == routine.id)
assert len(result) == 1
assert result[0]["name"] == "Test Routine"
def test_get_routine_by_id(self):
"""Test fetching routine by ID."""
routine = Routine(name="Fetch Test", points=30, image_id=None, user_id="user1")
routine_db.insert(routine.to_dict())
result = routine_db.get(Query().id == routine.id)
assert result is not None
assert result["name"] == "Fetch Test"
def test_list_user_routines(self):
"""Test listing routines for a user."""
r1 = Routine(name="R1", points=10, image_id=None, user_id="user1")
r2 = Routine(name="R2", points=20, image_id=None, user_id="user1")
r3 = Routine(name="R3", points=15, image_id=None, user_id="user2")
routine_db.insert(r1.to_dict())
routine_db.insert(r2.to_dict())
routine_db.insert(r3.to_dict())
results = routine_db.search(Query().user_id == "user1")
assert len(results) == 2
def test_update_routine(self):
"""Test updating a routine."""
routine = Routine(name="Original", points=50, image_id=None, user_id="user1")
routine_db.insert(routine.to_dict())
routine_db.update({"name": "Updated", "points": 100}, Query().id == routine.id)
result = routine_db.get(Query().id == routine.id)
assert result["name"] == "Updated"
assert result["points"] == 100
class TestRoutineItemDB:
"""Test routine item database operations."""
def setup_method(self):
"""Clear item db before each test."""
routine_items_db.truncate()
def test_add_routine_item(self):
"""Test adding a routine item."""
item = RoutineItem(routine_id="r1", name="Make Bed", image_id=None, order=0)
routine_items_db.insert(item.to_dict())
result = routine_items_db.search(Query().routine_id == "r1")
assert len(result) == 1
assert result[0]["name"] == "Make Bed"
def test_get_items_for_routine(self):
"""Test fetching all items for a routine."""
i1 = RoutineItem(routine_id="r1", name="Item1", image_id=None, order=0)
i2 = RoutineItem(routine_id="r1", name="Item2", image_id=None, order=1)
i3 = RoutineItem(routine_id="r2", name="Item3", image_id=None, order=0)
routine_items_db.insert(i1.to_dict())
routine_items_db.insert(i2.to_dict())
routine_items_db.insert(i3.to_dict())
results = routine_items_db.search(Query().routine_id == "r1")
assert len(results) == 2
def test_delete_item(self):
"""Test deleting a routine item."""
item = RoutineItem(routine_id="r1", name="Test", image_id=None, order=0)
routine_items_db.insert(item.to_dict())
routine_items_db.remove(Query().id == item.id)
result = routine_items_db.search(Query().id == item.id)
assert len(result) == 0
class TestRoutineScheduleDB:
"""Test routine schedule database operations."""
def setup_method(self):
"""Clear schedule db before each test."""
routine_schedules_db.truncate()
def test_upsert_schedule(self):
"""Test upserting a routine schedule."""
day_configs = [{"day": 0, "hour": 8, "minute": 0}]
schedule = RoutineSchedule(
child_id="c1",
routine_id="r1",
mode="days",
enabled=True,
day_configs=day_configs,
default_hour=8,
default_minute=0,
default_has_deadline=True,
)
routine_schedules_db.insert(schedule.to_dict())
result = routine_schedules_db.get(
(Query().child_id == "c1") & (Query().routine_id == "r1")
)
assert result is not None
assert result["mode"] == "days"
def test_delete_schedule(self):
"""Test deleting a schedule."""
schedule = RoutineSchedule(
child_id="c1",
routine_id="r1",
mode="interval",
enabled=True,
interval_days=2,
anchor_date="2026-05-01",
interval_has_deadline=True,
interval_hour=10,
interval_minute=0,
)
routine_schedules_db.insert(schedule.to_dict())
routine_schedules_db.remove(
(Query().child_id == "c1") & (Query().routine_id == "r1")
)
result = routine_schedules_db.search(Query().child_id == "c1")
assert len(result) == 0
class TestPendingRoutineConfirmation:
"""Test pending routine confirmation workflow."""
def setup_method(self):
"""Clear db before each test."""
pending_confirmations_db.truncate()
def test_create_routine_confirmation(self):
"""Test creating a pending routine confirmation."""
confirmation = PendingConfirmation(
child_id="c1",
entity_id="r1",
entity_type="routine",
user_id="u1",
status="pending",
)
pending_confirmations_db.insert(confirmation.to_dict())
result = pending_confirmations_db.get(
(Query().child_id == "c1") & (Query().entity_id == "r1") & (Query().entity_type == "routine")
)
assert result is not None
assert result["status"] == "pending"
def test_approve_routine_confirmation(self):
"""Test approving a routine confirmation."""
confirmation = PendingConfirmation(
child_id="c1",
entity_id="r1",
entity_type="routine",
user_id="u1",
status="pending",
)
pending_confirmations_db.insert(confirmation.to_dict())
today_utc = datetime.now(timezone.utc).isoformat()
pending_confirmations_db.update(
{"status": "approved", "approved_at": today_utc},
Query().id == confirmation.id,
)
result = pending_confirmations_db.get(Query().id == confirmation.id)
assert result["status"] == "approved"
assert result["approved_at"] is not None