459 lines
17 KiB
Python
459 lines
17 KiB
Python
"""
|
|
tests/test_schedule_acknowledge.py
|
|
-----------------------------------
|
|
Behaviour tests for phase50 — scheduled inspection receipt acknowledgement.
|
|
|
|
Runs on the in-memory SQLite app fixture (multi-tenancy inert). Covers:
|
|
|
|
* the assigned inspector can confirm receipt; the stamp and creator
|
|
notification land, and a second confirm is an idempotent no-op
|
|
* a manager cannot confirm on the inspector's behalf (assignee-only)
|
|
* the login-free token route confirms without a session, and is safe to
|
|
re-request (email client prefetch)
|
|
* a token issued to a previous assignee stops working once the schedule is
|
|
reassigned — the property that makes a stateless token safe
|
|
* tampered, expired and dangling tokens each render their own status page
|
|
rather than a bare error
|
|
* reassignment resets the acknowledgement; an unrelated edit does NOT
|
|
* acknowledgement is per ASSIGNMENT, not per occurrence — rolling the schedule
|
|
forward must not re-open it
|
|
* notify(extra_action=...) renders the confirm button in the email, and stops
|
|
once confirmed
|
|
"""
|
|
|
|
from datetime import 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
|
|
from app.models import inspector_assignment # noqa: F401
|
|
db.drop_all()
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.session.remove()
|
|
|
|
|
|
def _today():
|
|
"""Today in the app's timezone, not the machine's."""
|
|
from app.utils.time_utils import now_eastern
|
|
return now_eastern().date()
|
|
|
|
|
|
def _user(username, role):
|
|
from app import db
|
|
from app.models.user import User
|
|
u = User(username=username, full_name=username.title(), role=role,
|
|
email=f'{username}@example.com', active=True)
|
|
u.set_password('pw-correct1')
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
return u
|
|
|
|
|
|
def _seed():
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
from app.models.inspection import InspectionTemplate
|
|
|
|
tmpl = InspectionTemplate(name='Restroom Check', active=True,
|
|
form_schema=[{'id': 'f1', 'type': 'rating_5',
|
|
'label': 'Clean', 'row': 0, 'col': 0,
|
|
'rowSpan': 1, 'colSpan': 1}])
|
|
fac = Facility(name='Main Office', active=True)
|
|
db.session.add_all([tmpl, fac])
|
|
db.session.commit()
|
|
return tmpl, fac
|
|
|
|
|
|
def _schedule(tmpl, fac, inspector, creator, **kw):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
kw.setdefault('mode', 'plan')
|
|
kw.setdefault('frequency', 'weekly')
|
|
kw.setdefault('active', True)
|
|
s = InspectionSchedule(name='Weekly restrooms', template_id=tmpl.id,
|
|
facility_id=fac.id, inspector_id=inspector.id,
|
|
created_by=creator.id,
|
|
next_run_at=now_eastern() + timedelta(days=3), **kw)
|
|
db.session.add(s)
|
|
db.session.commit()
|
|
return s
|
|
|
|
|
|
def _login(client, user):
|
|
return client.post('/auth/login',
|
|
data={'username': user.username, 'password': 'pw-correct1'},
|
|
follow_redirects=True)
|
|
|
|
|
|
# ── Logged-in acknowledgement ────────────────────────────────────────────────
|
|
|
|
def test_assigned_inspector_can_confirm(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.models.notification import Notification
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
assert s.is_acknowledged is False
|
|
|
|
_login(client, insp)
|
|
resp = client.post(f'/inspection-schedules/{s.id}/acknowledge')
|
|
assert resp.status_code == 302
|
|
|
|
db.session.expire_all()
|
|
s = db.session.get(InspectionSchedule, s.id)
|
|
assert s.is_acknowledged is True
|
|
assert s.acknowledged_at is not None
|
|
|
|
# The creator is told, since they are the one waiting on the confirmation.
|
|
assert any('confirmed receipt' in n.title.lower()
|
|
for n in Notification.query.filter_by(user_id=mgr.id).all())
|
|
|
|
|
|
def test_confirming_twice_is_idempotent(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.models.notification import Notification
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
_login(client, insp)
|
|
client.post(f'/inspection-schedules/{s.id}/acknowledge')
|
|
db.session.expire_all()
|
|
first = db.session.get(InspectionSchedule, s.id).acknowledged_at
|
|
|
|
client.post(f'/inspection-schedules/{s.id}/acknowledge')
|
|
db.session.expire_all()
|
|
assert db.session.get(InspectionSchedule, s.id).acknowledged_at == first
|
|
# And the creator is not pestered a second time.
|
|
assert len([n for n in Notification.query.filter_by(user_id=mgr.id).all()
|
|
if 'confirmed receipt' in n.title.lower()]) == 1
|
|
|
|
|
|
def test_manager_cannot_confirm_on_the_inspectors_behalf(client):
|
|
"""The record means "this person saw it", so only the assignee may set it."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
_login(client, mgr)
|
|
assert client.post(f'/inspection-schedules/{s.id}/acknowledge').status_code == 403
|
|
db.session.expire_all()
|
|
assert db.session.get(InspectionSchedule, s.id).is_acknowledged is False
|
|
|
|
|
|
def test_other_inspector_cannot_confirm(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
other = _user('otto', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
_login(client, other)
|
|
assert client.post(f'/inspection-schedules/{s.id}/acknowledge').status_code == 403
|
|
db.session.expire_all()
|
|
assert db.session.get(InspectionSchedule, s.id).is_acknowledged is False
|
|
|
|
|
|
# ── Login-free email token ───────────────────────────────────────────────────
|
|
|
|
def _token_for(app, schedule):
|
|
from app.routes.inspection_schedules import _make_ack_token
|
|
with app.test_request_context():
|
|
return _make_ack_token(schedule)
|
|
|
|
|
|
def test_email_token_confirms_without_a_session(client, app):
|
|
"""Inspectors read this on a phone that is usually not logged in — a login
|
|
wall is exactly what stops them confirming."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
token = _token_for(app, s)
|
|
|
|
resp = client.get(f'/inspection-schedules/confirm/{token}')
|
|
assert resp.status_code == 200
|
|
assert b'Receipt confirmed' in resp.data
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(InspectionSchedule, s.id).is_acknowledged is True
|
|
|
|
|
|
def test_email_token_reclick_is_harmless(client, app):
|
|
"""A re-click, or an email client prefetching the link, must not error."""
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
token = _token_for(app, s)
|
|
|
|
client.get(f'/inspection-schedules/confirm/{token}')
|
|
resp = client.get(f'/inspection-schedules/confirm/{token}')
|
|
assert resp.status_code == 200
|
|
assert b'Already confirmed' in resp.data
|
|
|
|
|
|
def test_reassignment_invalidates_a_previously_emailed_token(client, app):
|
|
"""This is what makes a stateless token safe: the old link dies at
|
|
redemption, with nothing tracked anywhere."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
other = _user('otto', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
token = _token_for(app, s) # emailed to Ivy
|
|
|
|
s.inspector_id = other.id # reassigned to Otto
|
|
db.session.commit()
|
|
|
|
resp = client.get(f'/inspection-schedules/confirm/{token}')
|
|
assert resp.status_code == 409
|
|
assert b'No longer assigned to you' in resp.data
|
|
db.session.expire_all()
|
|
assert db.session.get(InspectionSchedule, s.id).is_acknowledged is False
|
|
|
|
|
|
def test_tampered_token_renders_the_invalid_page(client):
|
|
resp = client.get('/inspection-schedules/confirm/not-a-real-token')
|
|
assert resp.status_code == 400
|
|
assert b"isn't valid" in resp.data
|
|
|
|
|
|
def test_expired_token_renders_the_expired_page(client, app):
|
|
from itsdangerous import URLSafeTimedSerializer
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
with app.test_request_context():
|
|
from app.routes.inspection_schedules import _ACK_SALT
|
|
ser = URLSafeTimedSerializer(app.config['SECRET_KEY'], salt=_ACK_SALT)
|
|
# Sign it 31 days ago — past the 30-day _ACK_MAX_AGE.
|
|
import itsdangerous.timed
|
|
old = ser.dumps({'sid': s.id, 'iid': insp.id})
|
|
|
|
# Re-sign with a backdated timestamp by monkeypatching the serializer clock.
|
|
import time as _time
|
|
real = _time.time
|
|
try:
|
|
_time.time = lambda: real() - (60 * 60 * 24 * 31)
|
|
with app.test_request_context():
|
|
old = URLSafeTimedSerializer(
|
|
app.config['SECRET_KEY'], salt=_ACK_SALT
|
|
).dumps({'sid': s.id, 'iid': insp.id})
|
|
finally:
|
|
_time.time = real
|
|
|
|
resp = client.get(f'/inspection-schedules/confirm/{old}')
|
|
assert resp.status_code == 400
|
|
assert b'expired' in resp.data.lower()
|
|
|
|
|
|
def test_token_for_a_deleted_schedule_renders_missing(client, app):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
token = _token_for(app, s)
|
|
|
|
db.session.delete(db.session.get(InspectionSchedule, s.id))
|
|
db.session.commit()
|
|
|
|
resp = client.get(f'/inspection-schedules/confirm/{token}')
|
|
assert resp.status_code == 404
|
|
assert b'not found' in resp.data.lower()
|
|
|
|
|
|
def test_inactive_schedule_renders_inactive(client, app):
|
|
from app import db
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
token = _token_for(app, s)
|
|
|
|
s.active = False
|
|
db.session.commit()
|
|
|
|
resp = client.get(f'/inspection-schedules/confirm/{token}')
|
|
assert resp.status_code == 200
|
|
assert b'no longer active' in resp.data.lower()
|
|
|
|
|
|
# ── Reset semantics ──────────────────────────────────────────────────────────
|
|
|
|
def test_reassignment_resets_the_acknowledgement(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
other = _user('otto', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
_login(client, insp)
|
|
client.post(f'/inspection-schedules/{s.id}/acknowledge')
|
|
client.get('/auth/logout', follow_redirects=True)
|
|
|
|
_login(client, mgr)
|
|
client.post(f'/inspection-schedules/{s.id}/edit', data={
|
|
'name': 'Weekly restrooms', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': other.id, 'frequency': 'weekly', 'mode': 'plan',
|
|
'weekdays': ['0'], 'active': 'on',
|
|
}, follow_redirects=True)
|
|
|
|
db.session.expire_all()
|
|
s = db.session.get(InspectionSchedule, s.id)
|
|
assert s.inspector_id == other.id
|
|
assert s.is_acknowledged is False
|
|
|
|
|
|
def test_unrelated_edit_does_not_reset_the_acknowledgement(client):
|
|
"""Only a real reassignment re-opens it — otherwise every rename would make
|
|
the inspector confirm again."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
_login(client, insp)
|
|
client.post(f'/inspection-schedules/{s.id}/acknowledge')
|
|
client.get('/auth/logout', follow_redirects=True)
|
|
db.session.expire_all()
|
|
stamped = db.session.get(InspectionSchedule, s.id).acknowledged_at
|
|
|
|
_login(client, mgr)
|
|
client.post(f'/inspection-schedules/{s.id}/edit', data={
|
|
'name': 'Renamed round', 'template_id': tmpl.id, 'facility_id': fac.id,
|
|
'inspector_id': insp.id, 'frequency': 'weekly', 'mode': 'plan',
|
|
'weekdays': ['0'], 'active': 'on',
|
|
}, follow_redirects=True)
|
|
|
|
db.session.expire_all()
|
|
s = db.session.get(InspectionSchedule, s.id)
|
|
assert s.name == 'Renamed round'
|
|
assert s.acknowledged_at == stamped
|
|
|
|
|
|
def test_acknowledgement_survives_rolling_the_schedule_forward(client):
|
|
"""Per ASSIGNMENT, not per occurrence — an inspector who confirmed "this
|
|
weekly round is mine" must not be asked again every week."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr, frequency='daily')
|
|
|
|
_login(client, insp)
|
|
client.post(f'/inspection-schedules/{s.id}/acknowledge')
|
|
db.session.expire_all()
|
|
|
|
s = db.session.get(InspectionSchedule, s.id)
|
|
stamped = s.acknowledged_at
|
|
s.fulfill()
|
|
db.session.commit()
|
|
|
|
assert s.acknowledged_at == stamped
|
|
assert s.is_acknowledged is True
|
|
|
|
|
|
# ── notify(extra_action=...) ─────────────────────────────────────────────────
|
|
|
|
def test_confirm_action_is_offered_only_when_there_is_something_to_confirm(client, app):
|
|
from app import db
|
|
from app.routes.inspection_schedules import _confirm_action
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
mgr = _user('mona', 'admin')
|
|
s = _schedule(tmpl, fac, insp, mgr)
|
|
|
|
with app.test_request_context():
|
|
action = _confirm_action(s)
|
|
assert action is not None
|
|
assert action['label'] == 'Confirm receipt'
|
|
assert '/inspection-schedules/confirm/' in action['url']
|
|
|
|
# Auto mode materialises its own inspection — nothing to receive.
|
|
s.mode = 'auto'
|
|
assert _confirm_action(s) is None
|
|
s.mode = 'plan'
|
|
|
|
# Already confirmed — the button stops appearing in later reminders.
|
|
s.acknowledged_at = datetime.now()
|
|
assert _confirm_action(s) is None
|
|
s.acknowledged_at = None
|
|
|
|
# Unassigned — nobody to confirm.
|
|
s.inspector_id = None
|
|
assert _confirm_action(s) is None
|
|
|
|
|
|
def test_email_renders_the_extra_action_button(client, app):
|
|
"""notify(extra_action=...) is new plumbing; assert it reaches the body."""
|
|
from flask import render_template_string
|
|
from app.utils.notifications import _EMAIL_HTML_SINGLE, _EMAIL_TEXT_SINGLE
|
|
|
|
action = {'label': 'Confirm receipt', 'url': 'https://lts.jqc.app/x/abc'}
|
|
with app.test_request_context():
|
|
html = render_template_string(_EMAIL_HTML_SINGLE, title='T', body='B',
|
|
link='/inspection-schedules',
|
|
base_url='https://lts.jqc.app',
|
|
extra_action=action)
|
|
text = render_template_string(_EMAIL_TEXT_SINGLE, title='T', body='B',
|
|
link='/inspection-schedules',
|
|
base_url='https://lts.jqc.app',
|
|
extra_action=action)
|
|
assert 'Confirm receipt' in html and action['url'] in html
|
|
assert 'Confirm receipt' in text and action['url'] in text
|
|
# The generic link still renders alongside it.
|
|
assert 'View Details' in html
|
|
|
|
# And without extra_action nothing changes for every other caller.
|
|
with app.test_request_context():
|
|
plain = render_template_string(_EMAIL_HTML_SINGLE, title='T', body='B',
|
|
link='/x', base_url='http://h',
|
|
extra_action=None)
|
|
assert 'Confirm receipt' not in plain
|
|
assert 'View Details' in plain
|