342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""
|
|
tests/test_scoped_pages_smoke.py
|
|
--------------------------------
|
|
Smoke coverage for the pages whose queries were rewritten during the database
|
|
optimization pass: the dashboard, the issues list, and the reports overview.
|
|
|
|
These routes had no request-level tests, so a query rewrite that produced
|
|
invalid SQL or a broken scope would only have surfaced in production — and in
|
|
MT that means in one tenant's production. Each test drives the real route
|
|
through the test client for a role that exercises a DIFFERENT branch of the
|
|
scoping code:
|
|
|
|
admin — unscoped branch
|
|
inspector — InspectorAssignment / get_inspector_scope() branch
|
|
customer — CustomerAssignment / get_customer_scope() branch
|
|
|
|
The assertions are deliberately about behaviour that must hold (status code,
|
|
scope isolation), not about markup.
|
|
|
|
What specifically is being guarded
|
|
----------------------------------
|
|
* `with_entities(*_ISSUE_CARD_COLS)` on the dashboard cards — a Row must keep
|
|
answering the same attribute names the tallies and sla_status() read.
|
|
* `facility_filter.isdigit()` on the issues list — `int('abc')` used to raise
|
|
ValueError and return 500 on a hand-edited query string.
|
|
* `scalar_subquery()` in reports `_scope_issue()` — including the empty case,
|
|
where `IN (empty subquery)` must match nothing rather than error.
|
|
* The collapsed scope gate in `issues.view()` (`_issue_readable_by`), which
|
|
the linked-issues panel now shares — see rule 110.
|
|
"""
|
|
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
with app.app_context():
|
|
from app import db
|
|
db.drop_all()
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.session.remove()
|
|
|
|
|
|
# ── Builders ─────────────────────────────────────────────────────────────────
|
|
|
|
_n = {'i': 0}
|
|
|
|
|
|
def _uniq():
|
|
_n['i'] += 1
|
|
return _n['i']
|
|
|
|
|
|
def _user(role='admin'):
|
|
from app import db
|
|
from app.models.user import User
|
|
n = _uniq()
|
|
u = User(username=f'{role}{n}', full_name=f'{role.title()} {n}',
|
|
email=f'{role}{n}@example.com', role=role,
|
|
active=True, password_set=True)
|
|
u.set_password('pw-correct1')
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
return u
|
|
|
|
|
|
def _project(name=None):
|
|
from app import db
|
|
from app.models.project import Project
|
|
p = Project(name=name or f'Contract {_uniq()}', active=True)
|
|
db.session.add(p)
|
|
db.session.commit()
|
|
return p
|
|
|
|
|
|
def _facility(project=None, name=None):
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
f = Facility(name=name or f'Facility {_uniq()}', active=True,
|
|
project_id=project.id if project else None)
|
|
db.session.add(f)
|
|
db.session.commit()
|
|
return f
|
|
|
|
|
|
def _area(facility):
|
|
from app import db
|
|
from app.models.facility import Area
|
|
a = Area(facility_id=facility.id, name=f'Area {_uniq()}',
|
|
area_type='restroom')
|
|
db.session.add(a)
|
|
db.session.commit()
|
|
return a
|
|
|
|
|
|
def _issue(facility=None, area=None, severity='high', status='open',
|
|
hours_ago=1, **kw):
|
|
from app import db
|
|
from app.models.issue import Issue
|
|
from app.utils.time_utils import now_eastern
|
|
issue = Issue(
|
|
facility_id=facility.id if facility else None,
|
|
area_id=area.id if area else None,
|
|
severity=severity,
|
|
description=kw.pop('description', 'test issue'),
|
|
status=status,
|
|
reported_at=now_eastern() - timedelta(hours=hours_ago),
|
|
)
|
|
for k, v in kw.items():
|
|
setattr(issue, k, v)
|
|
db.session.add(issue)
|
|
db.session.commit()
|
|
return issue
|
|
|
|
|
|
def _template():
|
|
from app import db
|
|
from app.models.inspection import InspectionTemplate
|
|
t = InspectionTemplate(name=f'Template {_uniq()}', active=True,
|
|
form_schema=[])
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
return t
|
|
|
|
|
|
def _inspection(facility, inspector, template, status='completed',
|
|
overall_score=90.0, days_ago=5, **kw):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
insp = Inspection(
|
|
template_id=template.id,
|
|
facility_id=facility.id,
|
|
inspector_id=inspector.id,
|
|
inspection_date=now_eastern() - timedelta(days=days_ago),
|
|
overall_score=overall_score,
|
|
status=status,
|
|
)
|
|
for k, v in kw.items():
|
|
setattr(insp, k, v)
|
|
db.session.add(insp)
|
|
db.session.commit()
|
|
return insp
|
|
|
|
|
|
def _assign_inspector(user, project):
|
|
from app import db
|
|
from app.models.inspector_assignment import InspectorAssignment
|
|
db.session.add(InspectorAssignment(user_id=user.id, project_id=project.id))
|
|
db.session.commit()
|
|
|
|
|
|
def _assign_customer(user, project):
|
|
from app import db
|
|
from app.models.project import CustomerAssignment
|
|
db.session.add(CustomerAssignment(user_id=user.id, project_id=project.id))
|
|
db.session.commit()
|
|
|
|
|
|
def _login(client, user):
|
|
return client.post('/auth/login',
|
|
data={'username': user.username, 'password': 'pw-correct1'},
|
|
follow_redirects=False)
|
|
|
|
|
|
# ── Dashboard ────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.parametrize('role', ['admin', 'director', 'project_manager', 'auditor'])
|
|
def test_dashboard_renders_for_staff_roles(client, role):
|
|
"""The card queries (severity / handler / SLA splits) build and run."""
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
facility = _facility(_project())
|
|
_issue(facility=facility, severity='critical', hours_ago=100)
|
|
_issue(facility=facility, severity='low', status='in_progress')
|
|
_issue(facility=facility, status='resolved', resolved_at=now_eastern())
|
|
|
|
_login(client, _user(role))
|
|
assert client.get('/').status_code == 200
|
|
|
|
|
|
def test_dashboard_renders_for_scoped_inspector(client):
|
|
project = _project()
|
|
inspector = _user('inspector')
|
|
_assign_inspector(inspector, project)
|
|
|
|
facility = _facility(project)
|
|
_issue(area=_area(facility), severity='high', hours_ago=50) # via area
|
|
_issue(facility=facility, severity='medium') # directly
|
|
|
|
_login(client, inspector)
|
|
assert client.get('/').status_code == 200
|
|
|
|
|
|
def test_dashboard_renders_for_inspector_with_no_assignments(client):
|
|
"""Strict scoping: no assignments must render an empty dashboard, not 500."""
|
|
_login(client, _user('inspector'))
|
|
assert client.get('/').status_code == 200
|
|
|
|
|
|
def test_dashboard_renders_for_customer(client):
|
|
project = _project()
|
|
customer = _user('customer')
|
|
_assign_customer(customer, project)
|
|
_issue(facility=_facility(project))
|
|
|
|
_login(client, customer)
|
|
assert client.get('/').status_code == 200
|
|
|
|
|
|
# ── Issues list ──────────────────────────────────────────────────────────────
|
|
|
|
def test_issues_list_renders(client):
|
|
_issue(facility=_facility(_project()))
|
|
_login(client, _user('admin'))
|
|
assert client.get('/issues/').status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize('query', [
|
|
'?facility_id=abc', # non-numeric — used to raise ValueError -> 500
|
|
'?facility_id=',
|
|
'?facility_id=999999', # numeric but nonexistent
|
|
'?contract_id=abc',
|
|
'?issue_id=abc',
|
|
'?severity=high&status=open',
|
|
'?handler_type=vendor',
|
|
'?unassigned=1',
|
|
'?sla=breached',
|
|
'?date_from=notadate&date_to=alsonot',
|
|
])
|
|
def test_issues_list_survives_malformed_filters(client, query):
|
|
"""A hand-edited or stale query string must never 500 the list."""
|
|
_issue(facility=_facility(_project()))
|
|
_login(client, _user('admin'))
|
|
assert client.get('/issues/' + query).status_code == 200
|
|
|
|
|
|
def test_issues_list_scopes_a_customer_to_their_own_facilities(client):
|
|
"""The scope filter still isolates customers after the query rewrite."""
|
|
mine_project, other_project = _project('Contract A'), _project('Contract B')
|
|
customer = _user('customer')
|
|
_assign_customer(customer, mine_project)
|
|
|
|
_issue(facility=_facility(mine_project, 'Mine'), description='visible to me')
|
|
_issue(facility=_facility(other_project, 'Theirs'),
|
|
description='other customer only')
|
|
|
|
_login(client, customer)
|
|
body = client.get('/issues/').get_data(as_text=True)
|
|
assert 'visible to me' in body
|
|
assert 'other customer only' not in body
|
|
|
|
|
|
# ── Issue detail scope gate ──────────────────────────────────────────────────
|
|
# issues.view() had two inline scope blocks; they were collapsed into
|
|
# _issue_readable_by() so the linked-issues panel could reuse the same rule
|
|
# (rule 110). These pin the behaviour that gate must keep.
|
|
|
|
def test_issue_view_allows_an_inspector_inside_their_contract(client):
|
|
project = _project()
|
|
inspector = _user('inspector')
|
|
_assign_inspector(inspector, project)
|
|
issue = _issue(facility=_facility(project))
|
|
|
|
_login(client, inspector)
|
|
assert client.get(f'/issues/{issue.id}').status_code == 200
|
|
|
|
|
|
def test_issue_view_denies_an_inspector_outside_their_contract(client):
|
|
mine, theirs = _project(), _project()
|
|
inspector = _user('inspector')
|
|
_assign_inspector(inspector, mine)
|
|
issue = _issue(facility=_facility(theirs), description='not yours')
|
|
|
|
_login(client, inspector)
|
|
res = client.get(f'/issues/{issue.id}', follow_redirects=True)
|
|
assert 'not yours' not in res.get_data(as_text=True)
|
|
|
|
|
|
def test_issue_view_denies_an_inspector_with_no_assignments(client):
|
|
"""Strict scoping: no assignments means no access, not full access."""
|
|
issue = _issue(facility=_facility(_project()), description='strictly scoped')
|
|
_login(client, _user('inspector'))
|
|
res = client.get(f'/issues/{issue.id}', follow_redirects=True)
|
|
assert 'strictly scoped' not in res.get_data(as_text=True)
|
|
|
|
|
|
def test_issue_view_denies_a_customer_outside_their_contract(client):
|
|
mine, theirs = _project(), _project()
|
|
customer = _user('customer')
|
|
_assign_customer(customer, mine)
|
|
issue = _issue(facility=_facility(theirs), description='other customer only')
|
|
|
|
_login(client, customer)
|
|
res = client.get(f'/issues/{issue.id}', follow_redirects=True)
|
|
assert 'other customer only' not in res.get_data(as_text=True)
|
|
|
|
|
|
def test_issue_view_allows_a_customer_inside_their_contract(client):
|
|
project = _project()
|
|
customer = _user('customer')
|
|
_assign_customer(customer, project)
|
|
issue = _issue(facility=_facility(project), description='mine to read')
|
|
|
|
_login(client, customer)
|
|
res = client.get(f'/issues/{issue.id}')
|
|
assert res.status_code == 200
|
|
assert 'mine to read' in res.get_data(as_text=True)
|
|
|
|
|
|
# ── Reports overview ─────────────────────────────────────────────────────────
|
|
|
|
def test_reports_overview_renders_for_admin(client):
|
|
project = _project()
|
|
_inspection(_facility(project), _user('inspector'), _template())
|
|
|
|
_login(client, _user('admin'))
|
|
assert client.get('/reports/').status_code == 200
|
|
|
|
|
|
def test_reports_overview_inspector_scope_uses_subquery(client):
|
|
"""_scope_issue() now filters via a subquery instead of an id list."""
|
|
project = _project()
|
|
inspector = _user('inspector')
|
|
_assign_inspector(inspector, project)
|
|
|
|
facility = _facility(project)
|
|
insp = _inspection(facility, inspector, _template())
|
|
_issue(facility=facility, inspection_id=insp.id)
|
|
|
|
_login(client, inspector)
|
|
assert client.get('/reports/').status_code == 200
|
|
|
|
|
|
def test_reports_overview_inspector_with_no_inspections(client):
|
|
"""IN (empty subquery) must match nothing rather than error."""
|
|
_login(client, _user('inspector'))
|
|
assert client.get('/reports/').status_code == 200
|