Aug 5 - Update code to follow up ST - MT14c
This commit is contained in:
@@ -0,0 +1,375 @@
|
||||
"""
|
||||
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.'
|
||||
Reference in New Issue
Block a user