"""Tests for the chore expiry notification scheduler.""" import os import pytest from datetime import datetime, timedelta, timezone from unittest.mock import patch, MagicMock from flask import Flask from tinydb import Query from utils.chore_expiry_notification_scheduler import ( get_expiring_chores_for_user, _build_push_payload, send_chore_expiry_notifications_for_user, run_chore_expiry_check, ) from utils.schedule_utils import interval_hits_today, is_scheduled_today, get_due_time_today from db.db import users_db, child_db, task_db, chore_schedules_db, pending_confirmations_db from tests.conftest import TEST_SECRET_KEY USER_ID = "expiry_notif_user" CHILD_ID = "expiry_notif_child" CHILD_ID_2 = "expiry_notif_child_2" TASK_ID = "expiry_notif_task" TASK_ID_2 = "expiry_notif_task_2" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _seed_user(push_enabled: bool = True, verified: bool = True): users_db.remove(Query().id == USER_ID) users_db.insert({ "id": USER_ID, "first_name": "Notify", "last_name": "Tester", "email": f"{USER_ID}@example.com", "password": "hashed", "verified": verified, "role": "user", "image_id": None, "marked_for_deletion": False, "timezone": "UTC", "email_digest_enabled": False, "push_notifications_enabled": push_enabled, }) def _seed_child(child_id: str = CHILD_ID, task_ids: list = None): if task_ids is None: task_ids = [TASK_ID] child_db.remove(Query().id == child_id) child_db.insert({ "id": child_id, "user_id": USER_ID, "name": "Alex" if child_id == CHILD_ID else "Sam", "age": 8, "points": 0, "tasks": task_ids, "rewards": [], "image_id": None, }) def _seed_task(task_id: str = TASK_ID, name: str = "Clean Room"): task_db.remove(Query().id == task_id) task_db.insert({ "id": task_id, "user_id": USER_ID, "name": name, "points": 10, "type": "chore", "image_id": None, }) def _seed_schedule( child_id: str = CHILD_ID, task_id: str = TASK_ID, mode: str = "days", day_configs: list = None, default_has_deadline: bool = True, default_hour: int = 21, default_minute: int = 0, enabled: bool = True, interval_days: int = 1, anchor_date: str = "", interval_has_deadline: bool = True, interval_hour: int = 21, interval_minute: int = 0, ): """Seed a chore schedule. For 'days' mode, day_configs defaults to all 7 days.""" if day_configs is None: day_configs = [ {"day": d, "hour": default_hour, "minute": default_minute} for d in range(7) ] chore_schedules_db.remove( (Query().child_id == child_id) & (Query().task_id == task_id) ) chore_schedules_db.insert({ "id": f"sched_{child_id}_{task_id}", "child_id": child_id, "task_id": task_id, "mode": mode, "day_configs": day_configs, "default_hour": default_hour, "default_minute": default_minute, "default_has_deadline": default_has_deadline, "interval_days": interval_days, "anchor_date": anchor_date, "interval_has_deadline": interval_has_deadline, "interval_hour": interval_hour, "interval_minute": interval_minute, "enabled": enabled, }) def _seed_confirmation(child_id: str = CHILD_ID, task_id: str = TASK_ID, status: str = "pending"): pending_confirmations_db.remove( (Query().child_id == child_id) & (Query().entity_id == task_id) ) pending_confirmations_db.insert({ "id": f"conf_{child_id}_{task_id}", "user_id": USER_ID, "child_id": child_id, "entity_id": task_id, "entity_type": "chore", "status": status, "approved_at": None, "created_at": 0, "updated_at": 0, }) def _cleanup(): for cid in (CHILD_ID, CHILD_ID_2): child_db.remove(Query().id == cid) chore_schedules_db.remove(Query().child_id == cid) pending_confirmations_db.remove(Query().child_id == cid) for tid in (TASK_ID, TASK_ID_2): task_db.remove(Query().id == tid) users_db.remove(Query().id == USER_ID) def _now_with_deadline_in_window(minutes_ahead: int = 30) -> tuple[datetime, int, int]: """Return (now_dt, hour, minute) so that deadline = now + minutes_ahead.""" now = datetime.now(timezone.utc) deadline = now + timedelta(minutes=minutes_ahead) return now, deadline.hour, deadline.minute @pytest.fixture def app(): flask_app = Flask(__name__) flask_app.config['TESTING'] = True flask_app.config['SECRET_KEY'] = TEST_SECRET_KEY return flask_app # --------------------------------------------------------------------------- # schedule_utils unit tests # --------------------------------------------------------------------------- class TestIntervalHitsToday: def test_same_day_as_anchor_hits(self): from datetime import date d = date(2026, 4, 22) assert interval_hits_today("2026-04-22", 3, d) is True def test_interval_day_hits(self): from datetime import date assert interval_hits_today("2026-04-22", 3, date(2026, 4, 25)) is True def test_non_interval_day_misses(self): from datetime import date assert interval_hits_today("2026-04-22", 3, date(2026, 4, 24)) is False def test_before_anchor_misses(self): from datetime import date assert interval_hits_today("2026-04-22", 1, date(2026, 4, 21)) is False def test_empty_anchor_hits_today(self): from datetime import date d = date(2026, 4, 22) assert interval_hits_today("", 1, d) is True class TestIsScheduledToday: def test_days_mode_matching_weekday(self): from datetime import date # 2026-04-22 is Wednesday → JS weekday 3 (Sun=0..Sat=6) schedule = {"mode": "days", "enabled": True, "day_configs": [{"day": 3}]} assert is_scheduled_today(schedule, date(2026, 4, 22)) is True def test_days_mode_non_matching_weekday(self): from datetime import date schedule = {"mode": "days", "enabled": True, "day_configs": [{"day": 1}]} assert is_scheduled_today(schedule, date(2026, 4, 22)) is False def test_paused_schedule_always_true(self): from datetime import date schedule = {"mode": "days", "enabled": False, "day_configs": []} assert is_scheduled_today(schedule, date(2026, 4, 22)) is True def test_interval_mode_hit(self): from datetime import date schedule = { "mode": "interval", "enabled": True, "interval_days": 1, "anchor_date": "2026-04-22", } assert is_scheduled_today(schedule, date(2026, 4, 22)) is True def test_interval_mode_miss(self): from datetime import date schedule = { "mode": "interval", "enabled": True, "interval_days": 3, "anchor_date": "2026-04-22", } assert is_scheduled_today(schedule, date(2026, 4, 23)) is False class TestGetDueTimeToday: def test_days_mode_returns_due_time(self): from datetime import date # Wednesday → JS 3 (Sun=0..Sat=6) schedule = { "mode": "days", "enabled": True, "default_has_deadline": True, "day_configs": [{"day": 3, "hour": 21, "minute": 0}], } assert get_due_time_today(schedule, date(2026, 4, 22)) == (21, 0) def test_days_mode_anytime_returns_none(self): from datetime import date schedule = { "mode": "days", "enabled": True, "default_has_deadline": False, "day_configs": [{"day": 4, "hour": 21, "minute": 0}], } assert get_due_time_today(schedule, date(2026, 4, 22)) is None def test_days_mode_wrong_day_returns_none(self): from datetime import date schedule = { "mode": "days", "enabled": True, "default_has_deadline": True, "day_configs": [{"day": 1, "hour": 21, "minute": 0}], } assert get_due_time_today(schedule, date(2026, 4, 22)) is None def test_paused_returns_none(self): from datetime import date schedule = { "mode": "days", "enabled": False, "default_has_deadline": True, "day_configs": [{"day": 4, "hour": 21, "minute": 0}], } assert get_due_time_today(schedule, date(2026, 4, 22)) is None def test_interval_mode_returns_due_time(self): from datetime import date schedule = { "mode": "interval", "enabled": True, "interval_days": 1, "anchor_date": "2026-04-22", "interval_has_deadline": True, "interval_hour": 20, "interval_minute": 30, } assert get_due_time_today(schedule, date(2026, 4, 22)) == (20, 30) def test_interval_mode_anytime_returns_none(self): from datetime import date schedule = { "mode": "interval", "enabled": True, "interval_days": 1, "anchor_date": "2026-04-22", "interval_has_deadline": False, "interval_hour": 20, "interval_minute": 30, } assert get_due_time_today(schedule, date(2026, 4, 22)) is None # --------------------------------------------------------------------------- # get_expiring_chores_for_user tests # --------------------------------------------------------------------------- class TestGetExpiringChoresForUser: def setup_method(self): _cleanup() _seed_user() _seed_child() _seed_task() def teardown_method(self): _cleanup() def test_chore_in_window_included(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert len(result) == 1 assert result[0]['task_id'] == TASK_ID assert result[0]['child_id'] == CHILD_ID def test_chore_deadline_past_excluded(self, app): # Deadline 10 minutes in the past now = datetime.now(timezone.utc) past = now - timedelta(minutes=10) _seed_schedule(default_hour=past.hour, default_minute=past.minute) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_chore_deadline_beyond_window_excluded(self, app): # Deadline 80 minutes ahead (beyond 75-min window) now, h, m = _now_with_deadline_in_window(minutes_ahead=80) _seed_schedule(default_hour=h, default_minute=m) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_anytime_chore_excluded(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_has_deadline=False, default_hour=h, default_minute=m) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_wrong_day_excluded(self, app): from datetime import date now = datetime.now(timezone.utc) # Schedule only on a day that is NOT today today_js = (now.weekday() + 1) % 7 wrong_day = (today_js + 1) % 7 h = (now + timedelta(minutes=30)).hour m = (now + timedelta(minutes=30)).minute _seed_schedule( day_configs=[{"day": wrong_day, "hour": h, "minute": m}], default_hour=h, default_minute=m, ) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_pending_confirmation_excludes_chore(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) _seed_confirmation(status="pending") with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_approved_confirmation_excludes_chore(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) _seed_confirmation(status="approved") with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_no_schedule_excluded(self, app): now = datetime.now(timezone.utc) # No schedule seeded with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_paused_schedule_excluded(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(enabled=False, default_hour=h, default_minute=m) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_interval_mode_hit_included(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) today_iso = now.strftime("%Y-%m-%d") _seed_schedule( mode="interval", interval_days=1, anchor_date=today_iso, interval_has_deadline=True, interval_hour=h, interval_minute=m, ) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert len(result) == 1 def test_interval_mode_miss_excluded(self, app): now = datetime.now(timezone.utc) h = (now + timedelta(minutes=30)).hour m = (now + timedelta(minutes=30)).minute # Anchor yesterday with interval_days=2 → not today yesterday = (now - timedelta(days=1)).strftime("%Y-%m-%d") _seed_schedule( mode="interval", interval_days=2, anchor_date=yesterday, interval_has_deadline=True, interval_hour=h, interval_minute=m, ) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert result == [] def test_multiple_children_grouped(self, app): _seed_child(CHILD_ID_2, task_ids=[TASK_ID_2]) _seed_task(TASK_ID_2, name="Take Out Trash") now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(child_id=CHILD_ID, task_id=TASK_ID, default_hour=h, default_minute=m) _seed_schedule(child_id=CHILD_ID_2, task_id=TASK_ID_2, default_hour=h, default_minute=m) with app.app_context(): result = get_expiring_chores_for_user(USER_ID, "UTC", now) assert len(result) == 2 child_ids = {r['child_id'] for r in result} assert CHILD_ID in child_ids assert CHILD_ID_2 in child_ids # --------------------------------------------------------------------------- # _build_push_payload tests # --------------------------------------------------------------------------- class TestBuildPushPayload: def test_single_chore_sets_title_and_deep_link(self): expiring = [{"child_id": "c1", "child_name": "Alex", "task_id": "t1", "task_name": "Clean Room"}] payload = _build_push_payload(expiring) assert payload['title'] == "Chore ending soon: Clean Room" assert payload['child_id'] == "c1" assert payload['entity_id'] == "t1" assert payload['type'] == 'chore_expiring_soon' def test_multiple_chores_uses_generic_title(self): expiring = [ {"child_id": "c1", "child_name": "Alex", "task_id": "t1", "task_name": "Clean Room"}, {"child_id": "c1", "child_name": "Alex", "task_id": "t2", "task_name": "Make Bed"}, ] payload = _build_push_payload(expiring) assert payload['title'] == "Chores ending soon" assert payload['child_id'] is None assert payload['entity_id'] is None def test_body_groups_by_child(self): expiring = [ {"child_id": "c1", "child_name": "Alex", "task_id": "t1", "task_name": "Clean Room"}, {"child_id": "c1", "child_name": "Alex", "task_id": "t2", "task_name": "Make Bed"}, {"child_id": "c2", "child_name": "Sam", "task_id": "t3", "task_name": "Trash"}, ] payload = _build_push_payload(expiring) assert "Alex: Clean Room, Make Bed" in payload['body'] assert "Sam: Trash" in payload['body'] def test_no_approve_deny_tokens(self): expiring = [{"child_id": "c1", "child_name": "Alex", "task_id": "t1", "task_name": "Clean Room"}] payload = _build_push_payload(expiring) assert 'approve_token' not in payload assert 'deny_token' not in payload # --------------------------------------------------------------------------- # send_chore_expiry_notifications_for_user tests # --------------------------------------------------------------------------- class TestSendChoreExpiryNotificationsForUser: def setup_method(self): _cleanup() _seed_user() _seed_child() _seed_task() def teardown_method(self): _cleanup() def test_sends_push_when_chore_expiring(self, app): now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) with app.app_context(): with patch('utils.chore_expiry_notification_scheduler.send_push_to_user') as mock_push: count = send_chore_expiry_notifications_for_user(USER_ID, "UTC") assert count == 1 mock_push.assert_called_once() payload = mock_push.call_args[0][1] assert payload['type'] == 'chore_expiring_soon' def test_no_push_when_no_expiring_chores(self, app): # No schedule seeded → nothing expiring with app.app_context(): with patch('utils.chore_expiry_notification_scheduler.send_push_to_user') as mock_push: count = send_chore_expiry_notifications_for_user(USER_ID, "UTC") assert count == 0 mock_push.assert_not_called() # --------------------------------------------------------------------------- # run_chore_expiry_check tests # --------------------------------------------------------------------------- class TestRunChoreExpiryCheck: def setup_method(self): _cleanup() def teardown_method(self): _cleanup() def test_skips_when_db_env_is_e2e(self, app): with patch.dict(os.environ, {'DB_ENV': 'e2e'}): with patch('utils.chore_expiry_notification_scheduler.send_push_to_user') as mock_push: run_chore_expiry_check(app) mock_push.assert_not_called() def test_skips_user_with_push_disabled(self, app): _seed_user(push_enabled=False) _seed_child() _seed_task() now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) with app.app_context(): with patch('utils.chore_expiry_notification_scheduler.send_push_to_user') as mock_push: run_chore_expiry_check(app) mock_push.assert_not_called() def test_skips_unverified_user(self, app): _seed_user(verified=False) _seed_child() _seed_task() now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) with app.app_context(): with patch('utils.chore_expiry_notification_scheduler.send_push_to_user') as mock_push: run_chore_expiry_check(app) mock_push.assert_not_called() def test_sends_push_for_eligible_user(self, app): _seed_user(push_enabled=True, verified=True) _seed_child() _seed_task() now, h, m = _now_with_deadline_in_window(minutes_ahead=30) _seed_schedule(default_hour=h, default_minute=m) with app.app_context(): with patch('utils.chore_expiry_notification_scheduler.send_push_to_user') as mock_push: run_chore_expiry_check(app) mock_push.assert_called_once()