116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""
|
|
tests/test_inspection_schedules.py
|
|
----------------------------------
|
|
End-to-end behaviour test for phase34 recurring inspection schedules.
|
|
|
|
Runs on the in-memory SQLite app fixture (multi-tenancy inert). Drives the real
|
|
token-protected cron endpoint (POST /inspection-schedules/run) so the full path
|
|
— including request-context url_for link building inside notify() — is exercised
|
|
exactly as it runs in production. Asserts:
|
|
|
|
* a due schedule materialises a real in_progress Inspection for the inspector
|
|
* an in-app Notification is fired for the assigned inspector
|
|
* a not-yet-due schedule is left untouched
|
|
* a bad/missing token is rejected with 403
|
|
"""
|
|
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Fresh schema + test client for each test (isolated in-memory DB)."""
|
|
app.config['DIGEST_SECRET'] = 'test-digest'
|
|
with app.app_context():
|
|
from app import db
|
|
db.drop_all()
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.session.remove()
|
|
|
|
|
|
def _seed(suffix='a'):
|
|
from app import db
|
|
from app.models.user import User
|
|
from app.models.facility import Facility
|
|
from app.models.inspection import InspectionTemplate
|
|
|
|
inspector = User(username=f'insp_{suffix}', full_name='Ivy Inspector',
|
|
email=f'insp_{suffix}@example.com', role='inspector')
|
|
inspector.set_password('x')
|
|
tmpl = InspectionTemplate(name='Restroom Check', active=True,
|
|
form_schema=[{'id': 'f1', 'type': 'rating_5', 'label': 'Clean'}])
|
|
fac = Facility(name='Main Office', active=True)
|
|
db.session.add_all([inspector, tmpl, fac])
|
|
db.session.commit()
|
|
return inspector, tmpl, fac
|
|
|
|
|
|
def test_cron_materialises_due_schedule_and_notifies(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.models.inspection import Inspection
|
|
from app.models.notification import Notification
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
inspector, tmpl, fac = _seed('due')
|
|
sched = InspectionSchedule(
|
|
name='Weekly restroom', template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=inspector.id, frequency='weekly', active=True,
|
|
created_at=now_eastern(), next_run_at=now_eastern() - timedelta(days=1),
|
|
)
|
|
db.session.add(sched)
|
|
db.session.commit()
|
|
sched_id, insp_user_id, fac_id, tmpl_id = sched.id, inspector.id, fac.id, tmpl.id
|
|
|
|
resp = client.post('/inspection-schedules/run', data={'token': 'test-digest'})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {'ok': True, 'due': 1, 'created': 1}
|
|
|
|
# A real in_progress inspection now exists for the assigned inspector.
|
|
insp = Inspection.query.filter_by(inspector_id=insp_user_id).first()
|
|
assert insp is not None
|
|
assert insp.status == 'in_progress'
|
|
assert insp.facility_id == fac_id
|
|
assert insp.template_id == tmpl_id
|
|
|
|
# The inspector got an in-app notification linked to the new inspection.
|
|
notif = Notification.query.filter_by(user_id=insp_user_id,
|
|
inspection_id=insp.id).first()
|
|
assert notif is not None
|
|
assert notif.event_type == 'inspection_scheduled'
|
|
|
|
# The schedule's cadence advanced into the future.
|
|
sched = db.session.get(InspectionSchedule, sched_id)
|
|
assert sched.last_run_at is not None
|
|
assert sched.next_run_at > now_eastern()
|
|
|
|
|
|
def test_cron_skips_future_schedule(client):
|
|
from app import db
|
|
from app.models.inspection_schedule import InspectionSchedule
|
|
from app.models.inspection import Inspection
|
|
from app.utils.time_utils import now_eastern
|
|
|
|
inspector, tmpl, fac = _seed('future')
|
|
sched = InspectionSchedule(
|
|
name='Not yet due', template_id=tmpl.id, facility_id=fac.id,
|
|
inspector_id=inspector.id, frequency='weekly', active=True,
|
|
created_at=now_eastern(), next_run_at=now_eastern() + timedelta(days=3),
|
|
)
|
|
db.session.add(sched)
|
|
db.session.commit()
|
|
|
|
resp = client.post('/inspection-schedules/run', data={'token': 'test-digest'})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['created'] == 0
|
|
assert Inspection.query.count() == 0
|
|
|
|
|
|
def test_cron_rejects_bad_token(client):
|
|
resp = client.post('/inspection-schedules/run', data={'token': 'wrong'})
|
|
assert resp.status_code == 403
|
|
assert resp.get_json()['ok'] is False
|