376 lines
14 KiB
Python
376 lines
14 KiB
Python
"""
|
|
tests/test_privacy_and_routing.py
|
|
----------------------------------
|
|
Behaviour tests for MT-14c — the non-schema ST catch-ups. No migrations.
|
|
|
|
Runs on the in-memory SQLite app fixture (multi-tenancy inert). Covers:
|
|
|
|
* notify_by_matrix() scopes the INSPECTOR role for 'inspection_completed' to
|
|
the inspection's own inspector, instead of the whole inspector pool
|
|
* custom-email recipients are deduplicated on the normalised address, and
|
|
blank entries are skipped
|
|
* support._redact_pii() scrubs email/phone/SSN/card patterns
|
|
* /auth/my-data/export returns the caller's own records and nobody else's
|
|
* /auth/my-data/delete-request hard-deletes a clean account, and ANONYMIZES
|
|
one with history rather than orphaning records
|
|
* SupportChatSession.message_count / .preview
|
|
"""
|
|
|
|
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
|
|
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):
|
|
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 _facility(name='Main Office'):
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
f = Facility(name=name, active=True)
|
|
db.session.add(f)
|
|
db.session.commit()
|
|
return f
|
|
|
|
|
|
def _template():
|
|
from app import db
|
|
from app.models.inspection import InspectionTemplate
|
|
t = 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(t)
|
|
db.session.commit()
|
|
return t
|
|
|
|
|
|
def _login(client, user):
|
|
return client.post('/auth/login',
|
|
data={'username': user.username, 'password': 'pw-correct1'},
|
|
follow_redirects=True)
|
|
|
|
|
|
# ── notify_by_matrix inspector scoping ───────────────────────────────────────
|
|
|
|
def test_inspection_completed_notifies_only_the_submitting_inspector(app, client):
|
|
"""Without the scoping, switching the Inspector column on for this event
|
|
mails EVERY active inspector on EVERY submitted inspection."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.models.notification import Notification
|
|
from app.models.notification_matrix import NotificationMatrix
|
|
from app.utils.notifications import notify_by_matrix
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
tmpl = _template()
|
|
fac = _facility()
|
|
doer = _user('ivy', 'inspector')
|
|
other1 = _user('otto', 'inspector')
|
|
other2 = _user('opal', 'inspector')
|
|
|
|
db.session.add(NotificationMatrix(event_type='inspection_completed',
|
|
role_key='inspector', enabled=True))
|
|
insp = Inspection(template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=doer.id, inspection_date=now_eastern(),
|
|
status='completed', completed_at=now_eastern())
|
|
db.session.add(insp)
|
|
db.session.commit()
|
|
|
|
notify_by_matrix(event_type='inspection_completed', title='Done',
|
|
body='An inspection was submitted.',
|
|
inspection_id=insp.id, facility_id=fac.id)
|
|
|
|
recipients = {n.user_id for n in
|
|
Notification.query.filter_by(event_type='inspection_completed').all()}
|
|
assert doer.id in recipients
|
|
assert other1.id not in recipients
|
|
assert other2.id not in recipients
|
|
|
|
|
|
def test_unresolvable_inspection_notifies_no_inspector(app, client):
|
|
"""Fail closed: notify nobody rather than everybody."""
|
|
from app import db
|
|
from app.models.notification import Notification
|
|
from app.models.notification_matrix import NotificationMatrix
|
|
from app.utils.notifications import notify_by_matrix
|
|
|
|
_user('ivy', 'inspector')
|
|
_user('otto', 'inspector')
|
|
db.session.add(NotificationMatrix(event_type='inspection_completed',
|
|
role_key='inspector', enabled=True))
|
|
db.session.commit()
|
|
|
|
# No inspection_id at all.
|
|
notify_by_matrix(event_type='inspection_completed', title='Done',
|
|
body='An inspection was submitted.')
|
|
assert Notification.query.filter_by(event_type='inspection_completed').count() == 0
|
|
|
|
|
|
def test_other_events_still_notify_the_whole_inspector_role(app, client):
|
|
"""The scoping is specific to inspection_completed — it must not silently
|
|
narrow every other event."""
|
|
from app import db
|
|
from app.models.notification import Notification
|
|
from app.models.notification_matrix import NotificationMatrix
|
|
from app.utils.notifications import notify_by_matrix
|
|
|
|
a = _user('ivy', 'inspector')
|
|
b = _user('otto', 'inspector')
|
|
db.session.add(NotificationMatrix(event_type='issue_created',
|
|
role_key='inspector', enabled=True))
|
|
db.session.commit()
|
|
|
|
notify_by_matrix(event_type='issue_created', title='New issue',
|
|
body='Something broke.')
|
|
recipients = {n.user_id for n in
|
|
Notification.query.filter_by(event_type='issue_created').all()}
|
|
assert {a.id, b.id} <= recipients
|
|
|
|
|
|
# ── Custom-email dedupe ──────────────────────────────────────────────────────
|
|
|
|
def test_custom_emails_are_deduplicated_and_blanks_skipped(app, client, monkeypatch):
|
|
"""The matrix stores this list as free text, so the same person can appear
|
|
twice with different casing or stray whitespace."""
|
|
from app import db
|
|
from app.models.notification_matrix import NotificationMatrix
|
|
from app.utils import notifications as notif
|
|
|
|
import json
|
|
row = NotificationMatrix(
|
|
event_type='issue_created', role_key='custom', enabled=True,
|
|
# Stored as a JSON list; MT has no setter, only get_custom_emails().
|
|
custom_emails=json.dumps(['Ops@x.com', 'ops@x.com ', ' ', 'other@x.com']),
|
|
)
|
|
db.session.add(row)
|
|
db.session.commit()
|
|
|
|
sent = []
|
|
monkeypatch.setattr(notif, '_send_custom_email',
|
|
lambda email, *a, **kw: sent.append(email))
|
|
|
|
notif.notify_by_matrix(event_type='issue_created', title='T', body='B')
|
|
|
|
assert len(sent) == 2
|
|
assert {e.strip().lower() for e in sent} == {'ops@x.com', 'other@x.com'}
|
|
|
|
|
|
# ── PII redaction ────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.parametrize('raw, expected_marker', [
|
|
('reach me at ops@lts.com', '[redacted-email]'),
|
|
('call 703-555-0142 please', '[redacted-phone]'),
|
|
('ssn is 123-45-6789', '[redacted-ssn]'),
|
|
('card 4111 1111 1111 1111 on file', '[redacted-number]'),
|
|
])
|
|
def test_redact_pii_scrubs_known_shapes(raw, expected_marker):
|
|
from app.routes.support import _redact_pii
|
|
out = _redact_pii(raw)
|
|
assert expected_marker in out
|
|
|
|
|
|
def test_redact_pii_leaves_ordinary_text_alone():
|
|
from app.routes.support import _redact_pii
|
|
text = 'The third floor restroom needs restocking before Monday.'
|
|
assert _redact_pii(text) == text
|
|
|
|
|
|
def test_redact_pii_handles_empty_input():
|
|
from app.routes.support import _redact_pii
|
|
assert _redact_pii('') == ''
|
|
assert _redact_pii(None) is None
|
|
|
|
|
|
# ── Self-service data export ─────────────────────────────────────────────────
|
|
|
|
def test_export_returns_only_the_callers_own_records(app, client):
|
|
import json
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
tmpl = _template()
|
|
fac = _facility()
|
|
me = _user('ivy', 'inspector')
|
|
other = _user('otto', 'inspector')
|
|
|
|
mine = Inspection(template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=me.id, inspection_date=now_eastern(),
|
|
status='completed')
|
|
theirs = Inspection(template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=other.id, inspection_date=now_eastern(),
|
|
status='completed')
|
|
db.session.add_all([mine, theirs])
|
|
db.session.commit()
|
|
mine_id, theirs_id = mine.id, theirs.id
|
|
|
|
_login(client, me)
|
|
resp = client.get('/auth/my-data/export')
|
|
assert resp.status_code == 200
|
|
assert 'application/json' in resp.headers['Content-Type']
|
|
assert 'attachment' in resp.headers['Content-Disposition']
|
|
|
|
data = json.loads(resp.get_data(as_text=True))
|
|
assert data['profile']['username'] == 'ivy'
|
|
ids = {i['id'] for i in data['inspections_performed']}
|
|
assert mine_id in ids
|
|
assert theirs_id not in ids # nobody else's work
|
|
|
|
|
|
def test_export_requires_login(client):
|
|
resp = client.get('/auth/my-data/export', follow_redirects=False)
|
|
assert resp.status_code == 302
|
|
assert '/auth/login' in resp.headers['Location']
|
|
|
|
|
|
# ── Self-service erasure ─────────────────────────────────────────────────────
|
|
|
|
def test_clean_account_is_hard_deleted(app, client):
|
|
from app import db
|
|
from app.models.user import User
|
|
|
|
me = _user('ivy', 'inspector')
|
|
uid = me.id
|
|
|
|
_login(client, me)
|
|
resp = client.post('/auth/my-data/delete-request', follow_redirects=False)
|
|
assert resp.status_code == 302
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(User, uid) is None
|
|
|
|
|
|
def test_account_with_history_is_anonymized_not_deleted(app, client):
|
|
"""Hard-deleting would orphan inspection history that must be kept for
|
|
audit continuity — so erase the identity and keep the ledger."""
|
|
from app import db
|
|
from app.models.user import User
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
tmpl = _template()
|
|
fac = _facility()
|
|
me = _user('ivy', 'inspector')
|
|
uid = me.id
|
|
|
|
insp = Inspection(template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=me.id, inspection_date=now_eastern(),
|
|
status='completed')
|
|
db.session.add(insp)
|
|
db.session.commit()
|
|
insp_id = insp.id
|
|
|
|
_login(client, me)
|
|
client.post('/auth/my-data/delete-request', follow_redirects=False)
|
|
|
|
db.session.expire_all()
|
|
u = db.session.get(User, uid)
|
|
assert u is not None # kept, so the FK is not orphaned
|
|
assert u.username == f'deleted_user_{uid}'
|
|
assert u.full_name is None
|
|
assert u.email == f'deleted_user_{uid}@deleted.local'
|
|
assert u.active is False
|
|
# The old password must no longer work — the hash was replaced with a
|
|
# random secret nobody holds.
|
|
assert u.check_password('pw-correct1') is False
|
|
# The history survives intact.
|
|
assert db.session.get(Inspection, insp_id) is not None
|
|
|
|
|
|
def test_anonymized_account_cannot_log_back_in(app, client):
|
|
from app import db
|
|
|
|
tmpl = _template()
|
|
fac = _facility()
|
|
me = _user('ivy', 'inspector')
|
|
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
db.session.add(Inspection(template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=me.id, inspection_date=now_eastern(),
|
|
status='completed'))
|
|
db.session.commit()
|
|
|
|
_login(client, me)
|
|
client.post('/auth/my-data/delete-request', follow_redirects=True)
|
|
|
|
resp = client.post('/auth/login',
|
|
data={'username': 'ivy', 'password': 'pw-correct1'},
|
|
follow_redirects=False)
|
|
assert resp.status_code == 200 # re-rendered form, not a redirect
|
|
assert client.get('/dashboard', follow_redirects=False).status_code == 302
|
|
|
|
|
|
def test_deletion_requires_login(client):
|
|
resp = client.post('/auth/my-data/delete-request', follow_redirects=False)
|
|
assert resp.status_code == 302
|
|
assert '/auth/login' in resp.headers['Location']
|
|
|
|
|
|
# ── SupportChatSession helpers ───────────────────────────────────────────────
|
|
|
|
def test_support_session_count_and_preview(app, client):
|
|
from app import db
|
|
from app.models.support import SupportChatSession, SupportChatMessage
|
|
|
|
cust = _user('cara', 'customer')
|
|
sess = SupportChatSession(customer_id=cust.id)
|
|
db.session.add(sess)
|
|
db.session.commit()
|
|
|
|
assert sess.message_count == 0
|
|
assert sess.preview == '(no messages)'
|
|
|
|
db.session.add_all([
|
|
SupportChatMessage(session_id=sess.id, role='user',
|
|
content='How do I export a report?'),
|
|
SupportChatMessage(session_id=sess.id, role='assistant',
|
|
content='Open Reports, then Export.'),
|
|
])
|
|
db.session.commit()
|
|
db.session.refresh(sess)
|
|
|
|
assert sess.message_count == 2
|
|
# The opening question, not the assistant's reply.
|
|
assert sess.preview == 'How do I export a report?'
|
|
|
|
|
|
def test_preview_skips_a_leading_assistant_message(app, client):
|
|
from app import db
|
|
from app.models.support import SupportChatSession, SupportChatMessage
|
|
|
|
cust = _user('cara', 'customer')
|
|
sess = SupportChatSession(customer_id=cust.id)
|
|
db.session.add(sess)
|
|
db.session.commit()
|
|
|
|
db.session.add_all([
|
|
SupportChatMessage(session_id=sess.id, role='assistant',
|
|
content='Hi! How can I help?'),
|
|
SupportChatMessage(session_id=sess.id, role='user',
|
|
content='My badge scanner is broken.'),
|
|
])
|
|
db.session.commit()
|
|
db.session.refresh(sess)
|
|
|
|
assert sess.preview == 'My badge scanner is broken.'
|