JQC_multi_tenant/tests/test_privacy_and_routing.py

"""
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.'