474 lines
18 KiB
Python
474 lines
18 KiB
Python
"""
|
|
tests/test_schedule_recurrence.py
|
|
----------------------------------
|
|
Behaviour tests for phase45 (frequency ENUM), phase46 (day-of-week /
|
|
day-of-month recurrence) and phase47 (end date) on `inspection_schedules`.
|
|
|
|
Runs on the in-memory SQLite app fixture (multi-tenancy inert). Covers:
|
|
|
|
* pure date maths — weekly weekday sets, monthly day-of-month with short-month
|
|
clamping, monthly nth-weekday, the longer month-stepping cadences
|
|
* align_due_date() snapping a picked start date forward onto the rule
|
|
* fulfill() rolling forward on the rule, deactivating a 'once' schedule, and
|
|
deactivating when the next occurrence crosses the end date
|
|
* fulfill() with NO next_run_fn argument still advances the due date — the
|
|
pre-phase46 trap where omitting it left the schedule perpetually due
|
|
* legacy rows (all recurrence columns NULL) keeping their exact old cadence
|
|
* the cron endpoint's expiry sweep deactivating a schedule that passed its end
|
|
date without ever running, and reporting it as "expired"
|
|
"""
|
|
|
|
from datetime import date, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Fresh schema + test client for each test (isolated in-memory DB)."""
|
|
app.config['DIGEST_SECRET'] = 'test-digest'
|
|
with app.app_context():
|
|
from app import db
|
|
db.drop_all()
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.session.remove()
|
|
|
|
|
|
def _seed(suffix='a'):
|
|
from app import db
|
|
from app.models.user import User
|
|
from app.models.facility import Facility
|
|
from app.models.inspection import InspectionTemplate
|
|
|
|
inspector = User(username=f'insp_{suffix}', full_name='Ivy Inspector',
|
|
email=f'insp_{suffix}@example.com', role='inspector')
|
|
inspector.set_password('x')
|
|
tmpl = InspectionTemplate(name='Restroom Check', active=True,
|
|
form_schema=[{'id': 'f1', 'type': 'rating_5', 'label': 'Clean'}])
|
|
fac = Facility(name='Main Office', active=True)
|
|
db.session.add_all([inspector, tmpl, fac])
|
|
db.session.commit()
|
|
return inspector, tmpl, fac
|
|
|
|
|
|
def _sched(suffix, **kw):
|
|
"""Persist an InspectionSchedule with sensible defaults."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed(suffix)
|
|
kw.setdefault('name', 'Test schedule')
|
|
kw.setdefault('mode', 'plan')
|
|
kw.setdefault('active', True)
|
|
s = InspectionSchedule(template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=inspector.id, **kw)
|
|
db.session.add(s)
|
|
db.session.commit()
|
|
return s
|
|
|
|
|
|
# ── Pure date maths (phase46) ────────────────────────────────────────────────
|
|
|
|
def test_weekly_weekday_set_advances_within_the_week(client):
|
|
# Mon=0, Wed=2, Fri=4
|
|
s = _sched('wk', frequency='weekly')
|
|
s.set_weekdays([0, 2, 4])
|
|
assert s.weekday_list == [0, 2, 4]
|
|
assert s.weekdays == '0,2,4'
|
|
|
|
monday = date(2026, 8, 3) # a Monday
|
|
assert monday.weekday() == 0
|
|
assert s.next_occurrence_after(monday) == date(2026, 8, 5) # Wed
|
|
assert s.next_occurrence_after(date(2026, 8, 5)) == date(2026, 8, 7) # Fri
|
|
# Friday wraps to the following Monday, not +7 days.
|
|
assert s.next_occurrence_after(date(2026, 8, 7)) == date(2026, 8, 10)
|
|
|
|
|
|
def test_align_due_date_snaps_forward_onto_the_rule(client):
|
|
s = _sched('align', frequency='weekly')
|
|
s.set_weekdays([0, 2, 4])
|
|
tuesday = date(2026, 8, 4)
|
|
assert tuesday.weekday() == 1
|
|
# Picking a Tuesday on a Mon/Wed/Fri schedule yields that Wednesday.
|
|
assert s.align_due_date(tuesday) == date(2026, 8, 5)
|
|
# A date already on the rule is left alone.
|
|
assert s.align_due_date(date(2026, 8, 5)) == date(2026, 8, 5)
|
|
|
|
|
|
def test_monthly_day_of_month_clamps_to_short_months(client):
|
|
from app.models.inspection_schedule import MONTH_MODE_DAY
|
|
s = _sched('dom', frequency='monthly')
|
|
s.month_mode = MONTH_MODE_DAY
|
|
s.day_of_month = 31
|
|
assert s.next_occurrence_after(date(2026, 1, 31)) == date(2026, 2, 28)
|
|
assert s.next_occurrence_after(date(2026, 3, 31)) == date(2026, 4, 30)
|
|
|
|
|
|
def test_monthly_nth_weekday(client):
|
|
from app.models.inspection_schedule import MONTH_MODE_NTH
|
|
s = _sched('nth', frequency='monthly')
|
|
s.month_mode = MONTH_MODE_NTH
|
|
s.nth_week = 2
|
|
s.nth_weekday = 1 # Tuesday
|
|
nxt = s.next_occurrence_after(date(2026, 8, 20))
|
|
assert nxt == date(2026, 9, 8)
|
|
assert nxt.weekday() == 1
|
|
|
|
s.nth_week = -1 # last Tuesday
|
|
assert s.next_occurrence_after(date(2026, 8, 20)) == date(2026, 9, 29)
|
|
|
|
|
|
def test_longer_cadences_step_whole_months(client):
|
|
s = _sched('long', frequency='quarterly')
|
|
assert s.next_occurrence_after(date(2026, 1, 15)) == date(2026, 4, 15)
|
|
s.frequency = 'bi-annually'
|
|
assert s.next_occurrence_after(date(2026, 1, 15)) == date(2026, 7, 15)
|
|
s.frequency = 'annually'
|
|
assert s.next_occurrence_after(date(2026, 1, 15)) == date(2027, 1, 15)
|
|
|
|
|
|
def test_legacy_row_with_null_recurrence_keeps_old_cadence(client):
|
|
"""A pre-phase46 row has every recurrence column NULL and must not change."""
|
|
s = _sched('legacy', frequency='weekly')
|
|
assert s.weekdays is None and s.month_mode is None
|
|
assert s.next_occurrence_after(date(2026, 8, 4)) == date(2026, 8, 11)
|
|
s.frequency = 'monthly'
|
|
assert s.next_occurrence_after(date(2026, 8, 4)) == date(2026, 9, 4)
|
|
|
|
|
|
# ── fulfill() (phase46/47) ───────────────────────────────────────────────────
|
|
|
|
def test_fulfill_without_next_run_fn_still_advances(client):
|
|
"""The pre-phase46 trap: omitting next_run_fn used to leave next_run_at
|
|
untouched, leaving the schedule perpetually due."""
|
|
from app import db
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
due = now_eastern().replace(hour=6, minute=0, second=0, microsecond=0)
|
|
s = _sched('nofn', frequency='daily', next_run_at=due)
|
|
before = s.next_run_at
|
|
|
|
s.fulfill() # NO next_run_fn
|
|
db.session.commit()
|
|
|
|
assert s.next_run_at > before
|
|
assert s.next_run_at.date() > now_eastern().date()
|
|
assert s.active is True
|
|
assert s.last_completed_at is not None
|
|
# Time-of-day preserved.
|
|
assert s.next_run_at.hour == 6
|
|
|
|
|
|
def test_fulfill_rolls_forward_on_the_weekday_rule(client):
|
|
from app import db
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
today = now_eastern().date()
|
|
s = _sched('roll', frequency='weekly',
|
|
next_run_at=datetime.combine(today, datetime.min.time()).replace(hour=6))
|
|
# Every weekday, so the next occurrence is simply tomorrow.
|
|
s.set_weekdays([0, 1, 2, 3, 4, 5, 6])
|
|
db.session.commit()
|
|
|
|
s.fulfill()
|
|
db.session.commit()
|
|
assert s.next_run_at.date() == today + timedelta(days=1)
|
|
assert s.active is True
|
|
|
|
|
|
def test_fulfill_deactivates_a_once_schedule(client):
|
|
from app import db
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
due = now_eastern().replace(hour=6, minute=0, second=0, microsecond=0)
|
|
s = _sched('once', frequency='once', next_run_at=due)
|
|
s.fulfill()
|
|
db.session.commit()
|
|
assert s.active is False
|
|
# The due date is left where it was — the row still shows the occurrence.
|
|
assert s.next_run_at == due
|
|
|
|
|
|
def test_fulfill_deactivates_when_next_occurrence_passes_end_date(client):
|
|
from app import db
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
today = now_eastern().date()
|
|
due = datetime.combine(today, datetime.min.time()).replace(hour=6)
|
|
# Daily schedule whose end date is today: tomorrow's occurrence is past it.
|
|
s = _sched('end', frequency='daily', next_run_at=due, end_date=today)
|
|
assert s.is_within_end_date(today) is True
|
|
assert s.is_within_end_date(today + timedelta(days=1)) is False
|
|
|
|
s.fulfill()
|
|
db.session.commit()
|
|
assert s.active is False
|
|
assert s.next_run_at.date() == today + timedelta(days=1)
|
|
|
|
|
|
def test_no_end_date_repeats_indefinitely(client):
|
|
from app import db
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
due = now_eastern().replace(hour=6, minute=0, second=0, microsecond=0)
|
|
s = _sched('noend', frequency='daily', next_run_at=due)
|
|
assert s.end_date is None
|
|
assert s.is_expired is False
|
|
for _ in range(5):
|
|
s.fulfill()
|
|
db.session.commit()
|
|
assert s.active is True
|
|
|
|
|
|
# ── Expiry sweep (phase47) ───────────────────────────────────────────────────
|
|
|
|
def test_expire_if_past_end_date_is_idempotent(client):
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
today = now_eastern().date()
|
|
s = _sched('exp', frequency='daily', end_date=today - timedelta(days=1))
|
|
assert s.is_expired is True
|
|
assert s.expire_if_past_end_date(today) is True
|
|
assert s.active is False
|
|
# Second call changes nothing.
|
|
assert s.expire_if_past_end_date(today) is False
|
|
|
|
|
|
def test_cron_expires_a_schedule_that_never_ran(client):
|
|
"""A schedule can reach its end date without ever producing an occurrence,
|
|
so fulfill() never runs and the boundary is checked nowhere else."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
today = now_eastern().date()
|
|
s = _sched('cron', frequency='daily', mode='plan',
|
|
next_run_at=datetime.combine(today - timedelta(days=10),
|
|
datetime.min.time()).replace(hour=6),
|
|
end_date=today - timedelta(days=5))
|
|
sid = s.id
|
|
|
|
resp = client.post('/inspection-schedules/run', data={'token': 'test-digest'})
|
|
assert resp.status_code == 200
|
|
payload = resp.get_json()
|
|
assert payload['ok'] is True
|
|
assert payload['expired'] == 1
|
|
|
|
db.session.expire_all()
|
|
reloaded = db.session.get(InspectionSchedule, sid)
|
|
assert reloaded.active is False
|
|
# And it must not have generated an overdue alert on the way out.
|
|
assert reloaded.overdue_notified is False
|
|
|
|
|
|
def test_cron_leaves_a_live_schedule_alone(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
today = now_eastern().date()
|
|
s = _sched('live', frequency='daily', mode='plan',
|
|
next_run_at=datetime.combine(today + timedelta(days=3),
|
|
datetime.min.time()).replace(hour=6),
|
|
end_date=today + timedelta(days=30))
|
|
sid = s.id
|
|
|
|
resp = client.post('/inspection-schedules/run', data={'token': 'test-digest'})
|
|
assert resp.get_json()['expired'] == 0
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(InspectionSchedule, sid).active is True
|
|
|
|
|
|
# ── Routes: create / edit form handling ──────────────────────────────────────
|
|
|
|
def _seed_manager(username='mgr'):
|
|
from app import db
|
|
from app.models.user import User
|
|
u = User(username=username, full_name='Mo Manager',
|
|
email=f'{username}@example.com', role='admin', active=True)
|
|
u.set_password('pw-correct1')
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
return u
|
|
|
|
|
|
def _login(client, username='mgr'):
|
|
return client.post('/auth/login',
|
|
data={'username': username, 'password': 'pw-correct1'},
|
|
follow_redirects=True)
|
|
|
|
|
|
def test_create_route_stores_recurrence_and_end_date(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed('crt')
|
|
_seed_manager()
|
|
_login(client)
|
|
|
|
start = date(2026, 8, 4) # a Tuesday
|
|
end = date(2026, 12, 31)
|
|
resp = client.post('/inspection-schedules/new', data={
|
|
'name': 'MWF restrooms', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': inspector.id, 'frequency': 'weekly', 'mode': 'plan',
|
|
'weekdays': ['0', '2', '4'],
|
|
'next_due_date': start.isoformat(), 'end_date': end.isoformat(),
|
|
}, follow_redirects=True)
|
|
assert resp.status_code == 200
|
|
|
|
s = InspectionSchedule.query.filter_by(name='MWF restrooms').one()
|
|
assert s.weekday_list == [0, 2, 4]
|
|
assert s.end_date == end
|
|
# Tuesday snapped forward onto the Wednesday.
|
|
assert s.due_date == date(2026, 8, 5)
|
|
assert s.recurrence_label == 'Weekly · Mon, Wed, Fri'
|
|
|
|
|
|
def test_create_rejects_weekly_with_no_weekdays(client):
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed('nowd')
|
|
_seed_manager()
|
|
_login(client)
|
|
|
|
client.post('/inspection-schedules/new', data={
|
|
'name': 'No days', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': inspector.id, 'frequency': 'weekly', 'mode': 'plan',
|
|
}, follow_redirects=True)
|
|
assert InspectionSchedule.query.filter_by(name='No days').first() is None
|
|
|
|
|
|
def test_create_rejects_end_date_on_a_one_time_schedule(client):
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed('onceend')
|
|
_seed_manager()
|
|
_login(client)
|
|
|
|
client.post('/inspection-schedules/new', data={
|
|
'name': 'Once with end', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': inspector.id, 'frequency': 'once', 'mode': 'plan',
|
|
'next_due_date': '2026-09-01', 'end_date': '2026-10-01',
|
|
}, follow_redirects=True)
|
|
assert InspectionSchedule.query.filter_by(name='Once with end').first() is None
|
|
|
|
|
|
def test_create_rejects_first_occurrence_past_the_end_date(client):
|
|
"""The end date clears the *picked* date but not the *aligned* one: a
|
|
Mon/Wed/Fri schedule started on a Tuesday first runs on the Wednesday."""
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed('past')
|
|
_seed_manager()
|
|
_login(client)
|
|
|
|
client.post('/inspection-schedules/new', data={
|
|
'name': 'Impossible', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': inspector.id, 'frequency': 'weekly', 'mode': 'plan',
|
|
'weekdays': ['0', '2', '4'],
|
|
'next_due_date': '2026-08-04', # Tuesday -> aligns to Wed the 5th
|
|
'end_date': '2026-08-04', # ...which is past this
|
|
}, follow_redirects=True)
|
|
assert InspectionSchedule.query.filter_by(name='Impossible').first() is None
|
|
|
|
|
|
def test_edit_does_not_move_the_due_date_on_an_unrelated_change(client):
|
|
"""Root-cause regression: edit() used to recompute next_run_at from now on
|
|
every save, discarding the manager's chosen due date and re-arming every
|
|
reminder."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed('edit')
|
|
_seed_manager()
|
|
_login(client)
|
|
|
|
s = InspectionSchedule(
|
|
name='Original', template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=inspector.id, frequency='weekly', mode='plan', active=True,
|
|
next_run_at=datetime(2026, 9, 2, 6, 0), weekdays='0,2,4',
|
|
advance_notified=True, due_notified=True,
|
|
)
|
|
db.session.add(s)
|
|
db.session.commit()
|
|
sid, original_due = s.id, s.next_run_at
|
|
|
|
# Rename only — resubmit the same recurrence and due date.
|
|
client.post(f'/inspection-schedules/{sid}/edit', data={
|
|
'name': 'Renamed', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': inspector.id, 'frequency': 'weekly', 'mode': 'plan',
|
|
'weekdays': ['0', '2', '4'],
|
|
'next_due_date': original_due.date().isoformat(),
|
|
'active': 'on',
|
|
}, follow_redirects=True)
|
|
|
|
db.session.expire_all()
|
|
reloaded = db.session.get(InspectionSchedule, sid)
|
|
assert reloaded.name == 'Renamed'
|
|
assert reloaded.next_run_at == original_due # unchanged
|
|
assert reloaded.advance_notified is True # reminders NOT re-armed
|
|
assert reloaded.due_notified is True
|
|
|
|
|
|
def test_edit_resets_reminders_when_the_due_date_moves(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
inspector, tmpl, fac = _seed('move')
|
|
_seed_manager()
|
|
_login(client)
|
|
|
|
s = InspectionSchedule(
|
|
name='Movable', template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=inspector.id, frequency='daily', mode='plan', active=True,
|
|
next_run_at=datetime(2026, 9, 2, 6, 0),
|
|
advance_notified=True, due_notified=True, overdue_notified=True,
|
|
)
|
|
db.session.add(s)
|
|
db.session.commit()
|
|
sid = s.id
|
|
|
|
client.post(f'/inspection-schedules/{sid}/edit', data={
|
|
'name': 'Movable', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': inspector.id, 'frequency': 'daily', 'mode': 'plan',
|
|
'next_due_date': '2026-09-20', 'active': 'on',
|
|
}, follow_redirects=True)
|
|
|
|
db.session.expire_all()
|
|
reloaded = db.session.get(InspectionSchedule, sid)
|
|
assert reloaded.next_run_at.date() == date(2026, 9, 20)
|
|
assert reloaded.next_run_at.hour == 6 # time-of-day preserved
|
|
assert reloaded.advance_notified is False
|
|
assert reloaded.due_notified is False
|
|
assert reloaded.overdue_notified is False
|
|
|
|
|
|
# ── Labels (phase45/46) ──────────────────────────────────────────────────────
|
|
|
|
def test_recurrence_label(client):
|
|
from app.models.inspection_schedule import MONTH_MODE_DAY, MONTH_MODE_NTH
|
|
|
|
s = _sched('lbl', frequency='weekly')
|
|
s.set_weekdays([0, 2, 4])
|
|
assert s.recurrence_label == 'Weekly · Mon, Wed, Fri'
|
|
|
|
s.frequency = 'monthly'
|
|
s.weekdays = None
|
|
s.month_mode = MONTH_MODE_DAY
|
|
s.day_of_month = 15
|
|
assert s.recurrence_label == 'Monthly · day 15'
|
|
|
|
s.month_mode = MONTH_MODE_NTH
|
|
s.day_of_month = None
|
|
s.nth_week = 2
|
|
s.nth_weekday = 1
|
|
assert s.recurrence_label == 'Monthly · 2nd Tuesday'
|
|
|
|
s.frequency = 'bi-annually'
|
|
assert s.recurrence_label.startswith('Every 6 months')
|
|
|
|
s.frequency = 'once'
|
|
assert s.recurrence_label == 'One-time'
|