337 lines
13 KiB
Python
337 lines
13 KiB
Python
"""
|
|
tests/test_followup_requests.py
|
|
--------------------------------
|
|
Behaviour tests for phase49 — follow-up request attribution and
|
|
customer-raised follow-ups.
|
|
|
|
Runs on the in-memory SQLite app fixture (multi-tenancy inert). Covers:
|
|
|
|
* a customer can request a follow-up on a completed inspection at THEIR
|
|
facility, and both attribution columns are set
|
|
* a customer cannot reach another client's inspection with a crafted POST,
|
|
cannot request on a draft, and cannot overwrite a pending request
|
|
* a customer cannot CLEAR a follow-up — request-only
|
|
* admin/director keep their existing flag behaviour, now attributed
|
|
* inspector and auditor are refused (they were before too, via
|
|
@supervisor_required; phase49 must not widen access to them)
|
|
* clear_followup() nulls the attribution along with the flag
|
|
* the request routes through notify_by_matrix as EVENT_FOLLOWUP_REQUESTED,
|
|
reaching the managers who action it while excluding the actor and the
|
|
inspection's own inspector (who is notified directly instead)
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Fresh schema + test client for each test (isolated in-memory DB)."""
|
|
with app.app_context():
|
|
from app import db
|
|
# get_inspector_scope() imports this model lazily, so the mapper is not
|
|
# registered at create_all() time and the table is missing when a
|
|
# logged-in inspector hits the dashboard. Import it up front.
|
|
from app.models import inspector_assignment # noqa: F401
|
|
db.drop_all()
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.session.remove()
|
|
|
|
|
|
def _user(username, role, **kw):
|
|
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, **kw)
|
|
u.set_password('pw-correct1')
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
return u
|
|
|
|
|
|
def _seed():
|
|
"""Two facilities, an inspector, a manager pool, and two customers."""
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
from app.models.inspection import InspectionTemplate
|
|
from app.models.project import Project, CustomerAssignment
|
|
|
|
# MT scopes customers through a Project; customer_assignments.project_id is
|
|
# NOT NULL, so each facility needs one even for a facility-level assignment.
|
|
proj_a = Project(name='Contract A', active=True)
|
|
proj_b = Project(name='Contract B', active=True)
|
|
tmpl = InspectionTemplate(name='Restroom Check', active=True,
|
|
form_schema=[{'id': 'f1', 'type': 'rating_5', 'label': 'Clean',
|
|
'row': 0, 'col': 0,
|
|
'rowSpan': 1, 'colSpan': 1}])
|
|
db.session.add_all([proj_a, proj_b, tmpl])
|
|
db.session.commit()
|
|
|
|
fac_a = Facility(name='Client A Site', active=True, project_id=proj_a.id)
|
|
fac_b = Facility(name='Client B Site', active=True, project_id=proj_b.id)
|
|
db.session.add_all([fac_a, fac_b])
|
|
db.session.commit()
|
|
|
|
inspector = _user('ivy', 'inspector')
|
|
admin = _user('ada', 'admin')
|
|
director = _user('dan', 'director')
|
|
pm = _user('pat', 'project_manager')
|
|
cust_a = _user('cara', 'customer')
|
|
cust_b = _user('carl', 'customer')
|
|
|
|
db.session.add_all([
|
|
CustomerAssignment(user_id=cust_a.id, project_id=proj_a.id,
|
|
facility_id=fac_a.id),
|
|
CustomerAssignment(user_id=cust_b.id, project_id=proj_b.id,
|
|
facility_id=fac_b.id),
|
|
])
|
|
db.session.commit()
|
|
|
|
return dict(tmpl=tmpl, fac_a=fac_a, fac_b=fac_b, inspector=inspector,
|
|
admin=admin, director=director, pm=pm,
|
|
cust_a=cust_a, cust_b=cust_b)
|
|
|
|
|
|
def _inspection(env, facility, status='completed'):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
insp = Inspection(template_id=env['tmpl'].id, facility_id=facility.id,
|
|
inspector_id=env['inspector'].id,
|
|
inspection_date=now_eastern(), status=status,
|
|
completed_at=now_eastern() if status == 'completed' else None,
|
|
overall_score=71.0)
|
|
db.session.add(insp)
|
|
db.session.commit()
|
|
return insp
|
|
|
|
|
|
def _login(client, user):
|
|
return client.post('/auth/login',
|
|
data={'username': user.username, 'password': 'pw-correct1'},
|
|
follow_redirects=True)
|
|
|
|
|
|
def _flag(client, inspection_id, note=None):
|
|
data = {'follow_up_note': note} if note else {}
|
|
return client.post(f'/inspections/{inspection_id}/flag-followup',
|
|
data=data, follow_redirects=False)
|
|
|
|
|
|
# ── Customer requests ────────────────────────────────────────────────────────
|
|
|
|
def test_customer_can_request_follow_up_at_their_own_facility(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env['cust_a'])
|
|
|
|
resp = _flag(client, insp.id, note='Stalls still dirty')
|
|
assert resp.status_code == 302
|
|
|
|
db.session.expire_all()
|
|
insp = db.session.get(Inspection, insp.id)
|
|
assert insp.follow_up_required is True
|
|
assert insp.follow_up_note == 'Stalls still dirty'
|
|
assert insp.follow_up_requested_by == env['cust_a'].id
|
|
assert insp.follow_up_requested_at is not None
|
|
# The relationship is what the template renders the badge from.
|
|
assert insp.follow_up_requester.role == 'customer'
|
|
|
|
|
|
def test_customer_cannot_reach_another_clients_inspection(client):
|
|
"""A crafted POST must not cross the facility scope."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_b']) # Client B's facility
|
|
_login(client, env['cust_a']) # Client A's customer
|
|
|
|
assert _flag(client, insp.id).status_code == 403
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, insp.id).follow_up_required is False
|
|
|
|
|
|
def test_customer_cannot_request_on_a_draft(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'], status='in_progress')
|
|
_login(client, env['cust_a'])
|
|
|
|
_flag(client, insp.id)
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, insp.id).follow_up_required is False
|
|
|
|
|
|
def test_repeat_customer_request_does_not_overwrite_the_pending_one(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env['cust_a'])
|
|
|
|
_flag(client, insp.id, note='First note')
|
|
db.session.expire_all()
|
|
first_at = db.session.get(Inspection, insp.id).follow_up_requested_at
|
|
|
|
_flag(client, insp.id, note='Second note')
|
|
db.session.expire_all()
|
|
reloaded = db.session.get(Inspection, insp.id)
|
|
assert reloaded.follow_up_note == 'First note'
|
|
assert reloaded.follow_up_requested_at == first_at
|
|
|
|
|
|
def test_customer_cannot_clear_a_follow_up(client):
|
|
"""Customers may REQUEST only — clearing stays @supervisor_required."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env['cust_a'])
|
|
_flag(client, insp.id)
|
|
|
|
resp = client.post(f'/inspections/{insp.id}/clear-followup',
|
|
follow_redirects=False)
|
|
assert resp.status_code in (302, 403)
|
|
|
|
db.session.expire_all()
|
|
# Whether it redirected or 403'd, the flag must still be up.
|
|
assert db.session.get(Inspection, insp.id).follow_up_required is True
|
|
|
|
|
|
# ── Staff behaviour is preserved ─────────────────────────────────────────────
|
|
|
|
def test_admin_flag_still_works_and_is_now_attributed(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env['admin'])
|
|
|
|
_flag(client, insp.id, note='Rework required')
|
|
db.session.expire_all()
|
|
insp = db.session.get(Inspection, insp.id)
|
|
assert insp.follow_up_required is True
|
|
assert insp.follow_up_requested_by == env['admin'].id
|
|
assert insp.follow_up_requester.role == 'admin'
|
|
|
|
|
|
@pytest.mark.parametrize('role_key', ['inspector', 'pm'])
|
|
def test_roles_without_permission_are_refused(client, role_key):
|
|
"""phase49 removed @supervisor_required from this route. It must still
|
|
refuse everyone who could not flag before."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env[role_key])
|
|
|
|
assert _flag(client, insp.id).status_code == 403
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, insp.id).follow_up_required is False
|
|
|
|
|
|
def test_auditor_is_refused(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
auditor = _user('aud', 'auditor')
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, auditor)
|
|
|
|
assert _flag(client, insp.id).status_code == 403
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, insp.id).follow_up_required is False
|
|
|
|
|
|
def test_clear_followup_nulls_the_attribution(client):
|
|
"""Leaving it behind would make the NEXT unattributed follow-up appear to
|
|
have been requested by whoever raised the previous one."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
|
|
_login(client, env['cust_a'])
|
|
_flag(client, insp.id, note='Please recheck')
|
|
client.get('/auth/logout', follow_redirects=True)
|
|
|
|
_login(client, env['admin'])
|
|
client.post(f'/inspections/{insp.id}/clear-followup', follow_redirects=True)
|
|
|
|
db.session.expire_all()
|
|
insp = db.session.get(Inspection, insp.id)
|
|
assert insp.follow_up_required is False
|
|
assert insp.follow_up_note is None
|
|
assert insp.follow_up_requested_by is None
|
|
assert insp.follow_up_requested_at is None
|
|
assert insp.follow_up_requester is None
|
|
|
|
|
|
# ── Notification routing ─────────────────────────────────────────────────────
|
|
|
|
def test_request_notifies_managers_via_the_matrix(client):
|
|
from app import db
|
|
from app.models.notification import Notification, EVENT_FOLLOWUP_REQUESTED
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env['cust_a'])
|
|
_flag(client, insp.id, note='Still dirty')
|
|
|
|
rows = Notification.query.filter_by(event_type=EVENT_FOLLOWUP_REQUESTED).all()
|
|
recipients = {n.user_id for n in rows}
|
|
|
|
# Defaults: admin ✓ director ✓ project_manager ✓
|
|
assert env['admin'].id in recipients
|
|
assert env['director'].id in recipients
|
|
assert env['pm'].id in recipients
|
|
# The actor never notifies themselves.
|
|
assert env['cust_a'].id not in recipients
|
|
# The inspection's own inspector is notified DIRECTLY instead, so the matrix
|
|
# must exclude them rather than double-notifying.
|
|
assert env['inspector'].id not in recipients
|
|
# ...and that direct notification did happen.
|
|
direct = Notification.query.filter_by(user_id=env['inspector'].id).all()
|
|
assert any('Follow-Up Required' in n.title for n in direct)
|
|
|
|
|
|
def test_matrix_defaults_include_followup_requested(client):
|
|
from app.models.notification_matrix import MATRIX_EVENTS, MATRIX_DEFAULTS
|
|
|
|
assert 'followup_requested' in MATRIX_EVENTS
|
|
assert MATRIX_DEFAULTS[('followup_requested', 'admin')] is True
|
|
assert MATRIX_DEFAULTS[('followup_requested', 'director')] is True
|
|
assert MATRIX_DEFAULTS[('followup_requested', 'project_manager')] is True
|
|
# Off, or every request would alert the entire inspector pool.
|
|
assert MATRIX_DEFAULTS[('followup_requested', 'inspector')] is False
|
|
assert MATRIX_DEFAULTS[('followup_requested', 'customer')] is False
|
|
|
|
|
|
def test_notification_body_marks_a_customer_request_as_such(client):
|
|
"""Staff must be able to tell a client request from an internal one at a
|
|
glance, not just from the badge on the detail page."""
|
|
from app.models.notification import Notification, EVENT_FOLLOWUP_REQUESTED
|
|
|
|
env = _seed()
|
|
insp = _inspection(env, env['fac_a'])
|
|
_login(client, env['cust_a'])
|
|
_flag(client, insp.id)
|
|
|
|
row = Notification.query.filter_by(
|
|
event_type=EVENT_FOLLOWUP_REQUESTED, user_id=env['admin'].id).first()
|
|
assert row is not None
|
|
assert 'customer' in row.body.lower()
|