382 lines
14 KiB
Python
382 lines
14 KiB
Python
"""
|
|
tests/test_open_items.py
|
|
-------------------------
|
|
Tests for the two open items closed in this pass.
|
|
|
|
1. PATCH /api/v1/inspections/<id> now clears the parent's
|
|
follow_up_required on the draft → completed transition, and inherits
|
|
parent_inspection_id from a schedule attached at PATCH time. Previously
|
|
only the POST path did either, so an iPad that created a follow-up as a
|
|
draft and submitted it via PATCH left the parent flagged forever.
|
|
|
|
2. The schedule list is split into Pending / Completed tabs, partitioned on
|
|
`active` so the split is exhaustive and nothing can be lost between them.
|
|
"""
|
|
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
app.config['DIGEST_SECRET'] = 'test-digest'
|
|
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 _today():
|
|
from app.utils.time_utils import now_eastern
|
|
return now_eastern().date()
|
|
|
|
|
|
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 _seed():
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
from app.models.inspection import InspectionTemplate
|
|
tmpl = InspectionTemplate(name='Restroom Check', active=True,
|
|
form_schema=[{'id': 'f1', 'type': 'rating_5',
|
|
'label': 'Clean', 'row': 0, 'col': 0,
|
|
'rowSpan': 1, 'colSpan': 1}])
|
|
fac = Facility(name='Main Office', active=True)
|
|
db.session.add_all([tmpl, fac])
|
|
db.session.commit()
|
|
return tmpl, fac
|
|
|
|
|
|
def _auth(user):
|
|
from app.api.jwt_utils import generate_access_token
|
|
return {'Authorization': f'Bearer {generate_access_token(user)}'}
|
|
|
|
|
|
def _completed(tmpl, fac, user, follow_up=True):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
i = Inspection(template_id=tmpl.id, facility_id=fac.id, inspector_id=user.id,
|
|
inspection_date=now_eastern(), status='completed',
|
|
completed_at=now_eastern(), overall_score=60.0,
|
|
follow_up_required=follow_up)
|
|
db.session.add(i)
|
|
db.session.commit()
|
|
return i
|
|
|
|
|
|
def _draft(tmpl, fac, user, **kw):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
i = Inspection(template_id=tmpl.id, facility_id=fac.id, inspector_id=user.id,
|
|
inspection_date=now_eastern(), status='in_progress', **kw)
|
|
db.session.add(i)
|
|
db.session.commit()
|
|
return i
|
|
|
|
|
|
# ── Item 1: PATCH clears the parent's follow-up flag ─────────────────────────
|
|
|
|
def test_patch_to_completed_clears_the_parents_follow_up(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
tmpl, fac = _seed()
|
|
user = _user('ivy', 'admin')
|
|
parent = _completed(tmpl, fac, user)
|
|
draft = _draft(tmpl, fac, user, parent_inspection_id=parent.id)
|
|
|
|
resp = client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'status': 'completed', 'form_data': {'f1': 5}})
|
|
assert resp.status_code == 200
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, parent.id).follow_up_required is False
|
|
|
|
|
|
def test_patch_inherits_the_parent_from_a_schedule_attached_at_patch_time(client):
|
|
"""A draft created before the schedule was known, then linked on submit."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
user = _user('ivy', 'admin')
|
|
parent = _completed(tmpl, fac, user)
|
|
|
|
sched = InspectionSchedule(name='Follow-up', template_id=tmpl.id,
|
|
facility_id=fac.id, inspector_id=user.id,
|
|
frequency='once', mode='plan', active=True,
|
|
parent_inspection_id=parent.id)
|
|
db.session.add(sched)
|
|
db.session.commit()
|
|
|
|
draft = _draft(tmpl, fac, user) # no parent yet
|
|
resp = client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'status': 'completed', 'form_data': {'f1': 5},
|
|
'scheduled_inspection_id': sched.id})
|
|
assert resp.status_code == 200
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, draft.id).parent_inspection_id == parent.id
|
|
assert db.session.get(Inspection, parent.id).follow_up_required is False
|
|
|
|
|
|
def test_patch_explicit_parent_wins_over_the_schedule(client):
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
user = _user('ivy', 'admin')
|
|
parent_a = _completed(tmpl, fac, user)
|
|
parent_b = _completed(tmpl, fac, user)
|
|
|
|
sched = InspectionSchedule(name='Follow-up', template_id=tmpl.id,
|
|
facility_id=fac.id, inspector_id=user.id,
|
|
frequency='once', mode='plan', active=True,
|
|
parent_inspection_id=parent_a.id)
|
|
db.session.add(sched)
|
|
db.session.commit()
|
|
|
|
draft = _draft(tmpl, fac, user)
|
|
client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'status': 'completed', 'form_data': {'f1': 5},
|
|
'scheduled_inspection_id': sched.id,
|
|
'parent_inspection_id': parent_b.id})
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, draft.id).parent_inspection_id == parent_b.id
|
|
assert db.session.get(Inspection, parent_b.id).follow_up_required is False
|
|
# The schedule's own parent must NOT have been touched.
|
|
assert db.session.get(Inspection, parent_a.id).follow_up_required is True
|
|
|
|
|
|
def test_patch_that_does_not_complete_leaves_the_flag_alone(client):
|
|
"""Only the draft → completed transition clears it."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
tmpl, fac = _seed()
|
|
user = _user('ivy', 'admin')
|
|
parent = _completed(tmpl, fac, user)
|
|
draft = _draft(tmpl, fac, user, parent_inspection_id=parent.id)
|
|
|
|
client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'notes': 'still working on it'})
|
|
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, parent.id).follow_up_required is True
|
|
|
|
|
|
def test_patch_on_an_already_completed_inspection_is_not_a_transition(client):
|
|
"""A second PATCH must not re-fire the side effects."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
tmpl, fac = _seed()
|
|
user = _user('ivy', 'admin')
|
|
parent = _completed(tmpl, fac, user)
|
|
draft = _draft(tmpl, fac, user, parent_inspection_id=parent.id)
|
|
|
|
client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'status': 'completed', 'form_data': {'f1': 5}})
|
|
db.session.expire_all()
|
|
|
|
# Re-raise the parent's flag, then PATCH the already-completed child again.
|
|
p = db.session.get(Inspection, parent.id)
|
|
p.follow_up_required = True
|
|
db.session.commit()
|
|
|
|
client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'status': 'completed'})
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, parent.id).follow_up_required is True
|
|
|
|
|
|
def test_patch_without_a_parent_is_unaffected(client):
|
|
"""The ordinary case: no parent link, nothing to clear, no error."""
|
|
from app import db
|
|
from app.models.inspection import Inspection
|
|
|
|
tmpl, fac = _seed()
|
|
user = _user('ivy', 'admin')
|
|
draft = _draft(tmpl, fac, user)
|
|
|
|
resp = client.patch(f'/api/v1/inspections/{draft.id}', headers=_auth(user),
|
|
json={'status': 'completed', 'form_data': {'f1': 5}})
|
|
assert resp.status_code == 200
|
|
db.session.expire_all()
|
|
assert db.session.get(Inspection, draft.id).status == 'completed'
|
|
|
|
|
|
# ── Item 2: Pending / Completed tabs ─────────────────────────────────────────
|
|
|
|
def _schedule(tmpl, fac, user, name, active=True, **kw):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.utils.time_utils import now_eastern
|
|
s = InspectionSchedule(name=name, template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=user.id, frequency='weekly',
|
|
mode='plan', active=active,
|
|
next_run_at=now_eastern() + timedelta(days=2), **kw)
|
|
db.session.add(s)
|
|
db.session.commit()
|
|
return s
|
|
|
|
|
|
def _login(client, user):
|
|
return client.post('/auth/login',
|
|
data={'username': user.username, 'password': 'pw-correct1'},
|
|
follow_redirects=True)
|
|
|
|
|
|
def test_tabs_partition_schedules_exhaustively(client):
|
|
tmpl, fac = _seed()
|
|
mgr = _user('mona', 'admin')
|
|
_schedule(tmpl, fac, mgr, 'Live one', active=True)
|
|
_schedule(tmpl, fac, mgr, 'Closed one', active=False)
|
|
_login(client, mgr)
|
|
|
|
pending = client.get('/inspection-schedules/?tab=pending').get_data(as_text=True)
|
|
assert 'Live one' in pending
|
|
assert 'Closed one' not in pending
|
|
|
|
done = client.get('/inspection-schedules/?tab=completed').get_data(as_text=True)
|
|
assert 'Closed one' in done
|
|
assert 'Live one' not in done
|
|
|
|
|
|
def test_pending_is_the_default_tab(client):
|
|
tmpl, fac = _seed()
|
|
mgr = _user('mona', 'admin')
|
|
_schedule(tmpl, fac, mgr, 'Live one', active=True)
|
|
_schedule(tmpl, fac, mgr, 'Closed one', active=False)
|
|
_login(client, mgr)
|
|
|
|
body = client.get('/inspection-schedules/').get_data(as_text=True)
|
|
assert 'Live one' in body
|
|
assert 'Closed one' not in body
|
|
|
|
|
|
def test_an_unknown_tab_falls_back_to_pending(client):
|
|
tmpl, fac = _seed()
|
|
mgr = _user('mona', 'admin')
|
|
_schedule(tmpl, fac, mgr, 'Live one', active=True)
|
|
_login(client, mgr)
|
|
|
|
resp = client.get('/inspection-schedules/?tab=nonsense')
|
|
assert resp.status_code == 200
|
|
assert 'Live one' in resp.get_data(as_text=True)
|
|
|
|
|
|
def test_counts_are_scoped_to_what_the_viewer_can_see(client):
|
|
"""An inspector's badges must count their own schedules, not everyone's."""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
|
|
tmpl, fac = _seed()
|
|
mgr = _user('mona', 'admin')
|
|
mine = _user('ivy', 'inspector')
|
|
other = _user('otto', 'inspector')
|
|
|
|
_schedule(tmpl, fac, mine, 'Mine live', active=True)
|
|
_schedule(tmpl, fac, mine, 'Mine closed', active=False)
|
|
_schedule(tmpl, fac, other, 'Theirs live', active=True)
|
|
|
|
_login(client, mine)
|
|
body = client.get('/inspection-schedules/?tab=pending').get_data(as_text=True)
|
|
assert 'Mine live' in body
|
|
assert 'Theirs live' not in body
|
|
|
|
# Scoped counts: one pending, one completed — not the tenant-wide 2 and 1.
|
|
from app.models.user import User
|
|
q = InspectionSchedule.query.filter(InspectionSchedule.inspector_id == mine.id)
|
|
assert q.filter(InspectionSchedule.active.is_(True)).count() == 1
|
|
assert q.filter(InspectionSchedule.active.is_(False)).count() == 1
|
|
|
|
|
|
def test_empty_pending_with_history_points_at_the_completed_tab(client):
|
|
"""Inviting someone to 'Create First Schedule' when they have history would
|
|
prompt a duplicate of something already closed."""
|
|
tmpl, fac = _seed()
|
|
mgr = _user('mona', 'admin')
|
|
_schedule(tmpl, fac, mgr, 'Closed one', active=False)
|
|
_login(client, mgr)
|
|
|
|
body = client.get('/inspection-schedules/?tab=pending').get_data(as_text=True)
|
|
assert 'all schedules are complete' in body.lower()
|
|
assert 'Create First Schedule' not in body
|
|
|
|
|
|
def test_truly_empty_state_still_offers_creation(client):
|
|
tmpl, fac = _seed()
|
|
mgr = _user('mona', 'admin')
|
|
_login(client, mgr)
|
|
|
|
body = client.get('/inspection-schedules/').get_data(as_text=True)
|
|
assert 'Create First Schedule' in body
|
|
|
|
|
|
def test_inspector_is_not_offered_schedule_creation(client):
|
|
"""Inspectors cannot create schedules — the route is
|
|
@project_manager_required, so the empty state must not dangle the button."""
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
_login(client, insp)
|
|
|
|
body = client.get('/inspection-schedules/').get_data(as_text=True)
|
|
assert 'Create First Schedule' not in body
|
|
|
|
|
|
def test_customer_directors_may_reach_the_schedule_list(client):
|
|
"""A Customer Director plans work at their OWN facilities (Aug 2026).
|
|
|
|
They reach the list — scoped to their assignments, which for an
|
|
unassigned account like this one means an empty list, not a 403.
|
|
"""
|
|
tmpl, fac = _seed()
|
|
cust = _user('cara', 'customer')
|
|
_login(client, cust)
|
|
assert client.get('/inspection-schedules/?tab=completed').status_code == 200
|
|
|
|
|
|
def test_customers_still_cannot_start_a_scheduled_inspection(client):
|
|
"""Planning is not performing — the boundary that replaced the old 403.
|
|
|
|
A Customer Director may create and edit a schedule; Start belongs to the
|
|
assigned inspector, and a customer is never one.
|
|
"""
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
tmpl, fac = _seed()
|
|
insp = _user('ivy', 'inspector')
|
|
sched = InspectionSchedule(
|
|
name='Weekly check', template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=insp.id, frequency='weekly', mode='plan', active=True,
|
|
created_at=now_eastern(), next_run_at=now_eastern(),
|
|
)
|
|
db.session.add(sched)
|
|
db.session.commit()
|
|
sid = sched.id
|
|
|
|
cust = _user('cara', 'customer')
|
|
_login(client, cust)
|
|
assert client.get(f'/inspection-schedules/{sid}/start').status_code == 403
|