Ryan Kegel
6982fa561f
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 2m40s
251 lines
9.7 KiB
Python
251 lines
9.7 KiB
Python
"""Tests for the state expiry scheduler."""
|
|
import os
|
|
import time
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from flask import Flask
|
|
from tinydb import Query
|
|
|
|
from utils.state_expiry_scheduler import (
|
|
_get_today_start_timestamp,
|
|
expire_stale_pending_for_user,
|
|
run_state_expiry_check,
|
|
)
|
|
from db.db import users_db, pending_confirmations_db
|
|
from tests.conftest import TEST_SECRET_KEY
|
|
|
|
EXPIRY_USER_ID = "expiry_test_user"
|
|
EXPIRY_USER_ID_2 = "expiry_test_user_2"
|
|
EXPIRY_CHILD_ID = "expiry_child_id"
|
|
EXPIRY_CHILD_ID_2 = "expiry_child_id_2"
|
|
EXPIRY_TASK_ID = "expiry_task_id"
|
|
EXPIRY_REWARD_ID = "expiry_reward_id"
|
|
|
|
|
|
def _yesterday_timestamp() -> float:
|
|
"""Return a Unix timestamp from 25 hours ago (safely before today's midnight)."""
|
|
return time.time() - (25 * 3600)
|
|
|
|
|
|
def _seed_user(user_id: str = EXPIRY_USER_ID, timezone: str = "UTC", verified: bool = True):
|
|
users_db.remove(Query().id == user_id)
|
|
users_db.insert({
|
|
"id": user_id,
|
|
"first_name": "Expiry",
|
|
"last_name": "Tester",
|
|
"email": f"{user_id}@example.com",
|
|
"password": "hashed",
|
|
"verified": verified,
|
|
"role": "user",
|
|
"image_id": None,
|
|
"marked_for_deletion": False,
|
|
"timezone": timezone,
|
|
"email_digest_enabled": False,
|
|
})
|
|
|
|
|
|
def _seed_pending(
|
|
pending_id: str,
|
|
user_id: str = EXPIRY_USER_ID,
|
|
child_id: str = EXPIRY_CHILD_ID,
|
|
entity_id: str = EXPIRY_TASK_ID,
|
|
entity_type: str = "chore",
|
|
status: str = "pending",
|
|
created_at: float = None,
|
|
):
|
|
pending_confirmations_db.remove(Query().id == pending_id)
|
|
pending_confirmations_db.insert({
|
|
"id": pending_id,
|
|
"user_id": user_id,
|
|
"child_id": child_id,
|
|
"entity_id": entity_id,
|
|
"entity_type": entity_type,
|
|
"status": status,
|
|
"approved_at": None,
|
|
"created_at": created_at if created_at is not None else _yesterday_timestamp(),
|
|
"updated_at": time.time(),
|
|
})
|
|
|
|
|
|
def _cleanup():
|
|
for uid in (EXPIRY_USER_ID, EXPIRY_USER_ID_2):
|
|
users_db.remove(Query().id == uid)
|
|
pending_confirmations_db.remove(Query().user_id == uid)
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
flask_app = Flask(__name__)
|
|
flask_app.config['TESTING'] = True
|
|
flask_app.config['SECRET_KEY'] = TEST_SECRET_KEY
|
|
return flask_app
|
|
|
|
|
|
class TestGetTodayStartTimestamp:
|
|
def test_returns_float(self):
|
|
ts = _get_today_start_timestamp("UTC")
|
|
assert isinstance(ts, float)
|
|
|
|
def test_today_start_is_before_now(self):
|
|
ts = _get_today_start_timestamp("UTC")
|
|
assert ts <= time.time()
|
|
|
|
def test_falls_back_to_utc_on_none(self):
|
|
ts_utc = _get_today_start_timestamp("UTC")
|
|
ts_none = _get_today_start_timestamp(None)
|
|
assert abs(ts_utc - ts_none) < 1 # within 1 second
|
|
|
|
def test_falls_back_to_utc_on_invalid_timezone(self):
|
|
ts_utc = _get_today_start_timestamp("UTC")
|
|
ts_bad = _get_today_start_timestamp("Invalid/Timezone")
|
|
assert abs(ts_utc - ts_bad) < 1
|
|
|
|
def test_timezone_shifts_midnight(self):
|
|
"""UTC-12 midnight is 12 hours later than UTC midnight in absolute terms."""
|
|
ts_utc = _get_today_start_timestamp("UTC")
|
|
ts_west = _get_today_start_timestamp("Etc/GMT+12")
|
|
# The western timezone's "today start" is at most 12 hours away from UTC's
|
|
assert abs(ts_utc - ts_west) <= 12 * 3600 + 1
|
|
|
|
|
|
class TestExpireStaleForUser:
|
|
def setup_method(self):
|
|
_cleanup()
|
|
|
|
def teardown_method(self):
|
|
_cleanup()
|
|
|
|
def test_pending_chore_from_yesterday_is_deleted(self, app):
|
|
_seed_pending("exp_chore_1", entity_type="chore", created_at=_yesterday_timestamp())
|
|
with app.app_context():
|
|
with patch('events.sse.send_event_to_user') as mock_send:
|
|
count = expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
assert count == 1
|
|
assert pending_confirmations_db.get(Query().id == "exp_chore_1") is None
|
|
|
|
def test_pending_chore_fires_reset_sse(self, app):
|
|
_seed_pending("exp_chore_2", entity_type="chore", created_at=_yesterday_timestamp())
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
mock_send.assert_called_once()
|
|
event = mock_send.call_args[0][1]
|
|
assert event.type == "child_chore_confirmation"
|
|
assert event.payload.data['operation'] == "RESET"
|
|
assert event.payload.data['task_id'] == EXPIRY_TASK_ID
|
|
|
|
def test_pending_reward_from_yesterday_is_deleted(self, app):
|
|
_seed_pending(
|
|
"exp_reward_1",
|
|
entity_id=EXPIRY_REWARD_ID,
|
|
entity_type="reward",
|
|
created_at=_yesterday_timestamp(),
|
|
)
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user'):
|
|
count = expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
assert count == 1
|
|
assert pending_confirmations_db.get(Query().id == "exp_reward_1") is None
|
|
|
|
def test_pending_reward_fires_cancelled_sse(self, app):
|
|
_seed_pending(
|
|
"exp_reward_2",
|
|
entity_id=EXPIRY_REWARD_ID,
|
|
entity_type="reward",
|
|
created_at=_yesterday_timestamp(),
|
|
)
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
mock_send.assert_called_once()
|
|
event = mock_send.call_args[0][1]
|
|
assert event.type == "child_reward_request"
|
|
assert event.payload.data['operation'] == "CANCELLED"
|
|
assert event.payload.data['reward_id'] == EXPIRY_REWARD_ID
|
|
|
|
def test_pending_record_from_today_is_not_deleted(self, app):
|
|
_seed_pending("exp_today_1", created_at=time.time())
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
count = expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
assert count == 0
|
|
assert pending_confirmations_db.get(Query().id == "exp_today_1") is not None
|
|
mock_send.assert_not_called()
|
|
|
|
def test_approved_record_from_yesterday_is_not_deleted(self, app):
|
|
_seed_pending("exp_approved_1", status="approved", created_at=_yesterday_timestamp())
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
count = expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
assert count == 0
|
|
assert pending_confirmations_db.get(Query().id == "exp_approved_1") is not None
|
|
mock_send.assert_not_called()
|
|
|
|
def test_no_pending_records_returns_zero(self, app):
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
count = expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
assert count == 0
|
|
mock_send.assert_not_called()
|
|
|
|
def test_multiple_stale_records_all_expired(self, app):
|
|
_seed_pending("exp_multi_1", entity_id=EXPIRY_TASK_ID, entity_type="chore",
|
|
created_at=_yesterday_timestamp())
|
|
_seed_pending("exp_multi_2", entity_id=EXPIRY_REWARD_ID, entity_type="reward",
|
|
created_at=_yesterday_timestamp())
|
|
with app.app_context():
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
count = expire_stale_pending_for_user(EXPIRY_USER_ID, "UTC")
|
|
assert count == 2
|
|
assert mock_send.call_count == 2
|
|
|
|
|
|
class TestRunStateExpiryCheck:
|
|
def setup_method(self):
|
|
_cleanup()
|
|
|
|
def teardown_method(self):
|
|
_cleanup()
|
|
|
|
def test_expires_stale_records_for_all_verified_users(self, app):
|
|
_seed_user(EXPIRY_USER_ID)
|
|
_seed_user(EXPIRY_USER_ID_2)
|
|
_seed_pending("exp_u1", user_id=EXPIRY_USER_ID, created_at=_yesterday_timestamp())
|
|
_seed_pending("exp_u2", user_id=EXPIRY_USER_ID_2, child_id=EXPIRY_CHILD_ID_2,
|
|
created_at=_yesterday_timestamp())
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user'):
|
|
run_state_expiry_check(app)
|
|
assert pending_confirmations_db.get(Query().id == "exp_u1") is None
|
|
assert pending_confirmations_db.get(Query().id == "exp_u2") is None
|
|
|
|
def test_only_expires_stale_records_not_todays(self, app):
|
|
_seed_user(EXPIRY_USER_ID)
|
|
_seed_pending("exp_stale", created_at=_yesterday_timestamp())
|
|
_seed_pending("exp_fresh", created_at=time.time())
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user'):
|
|
run_state_expiry_check(app)
|
|
assert pending_confirmations_db.get(Query().id == "exp_stale") is None
|
|
assert pending_confirmations_db.get(Query().id == "exp_fresh") is not None
|
|
|
|
def test_skips_when_db_env_is_e2e(self, app):
|
|
_seed_user(EXPIRY_USER_ID)
|
|
_seed_pending("exp_e2e", created_at=_yesterday_timestamp())
|
|
original = os.environ.get('DB_ENV')
|
|
try:
|
|
os.environ['DB_ENV'] = 'e2e'
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
run_state_expiry_check(app)
|
|
mock_send.assert_not_called()
|
|
assert pending_confirmations_db.get(Query().id == "exp_e2e") is not None
|
|
finally:
|
|
if original is not None:
|
|
os.environ['DB_ENV'] = original
|
|
else:
|
|
os.environ.pop('DB_ENV', None)
|
|
|
|
def test_no_error_when_user_has_no_pending_records(self, app):
|
|
_seed_user(EXPIRY_USER_ID)
|
|
with patch('utils.state_expiry_scheduler.send_event_to_user') as mock_send:
|
|
run_state_expiry_check(app)
|
|
mock_send.assert_not_called()
|