139 lines
5.1 KiB
Python
139 lines
5.1 KiB
Python
"""
|
|
tests/test_work_orders.py
|
|
-------------------------
|
|
End-to-end tests for the phase36 vendor work-order workflow.
|
|
|
|
* a manager dispatches an issue to a contractor (issue moves to in_progress)
|
|
* the contractor opens the tokenized public page (no login) and acknowledges
|
|
* the contractor marks it complete → issue moves to pending_verification
|
|
* staff (the reporter) get notified on each vendor action
|
|
* an unknown token 404s
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
from app import db, limiter
|
|
limiter.enabled = False
|
|
app.config['WTF_CSRF_ENABLED'] = False
|
|
app.config['SQLALCHEMY_ECHO'] = False
|
|
with app.app_context():
|
|
db.drop_all()
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.session.remove()
|
|
limiter.enabled = True
|
|
|
|
|
|
def _seed():
|
|
from app import db
|
|
from app.models.user import User
|
|
from app.models.facility import Facility
|
|
from app.models.issue import Issue
|
|
|
|
admin = User(username='mgr', full_name='Manager', email='mgr@x.com',
|
|
role='admin', active=True, password_set=True)
|
|
admin.set_password('pw-correct1')
|
|
reporter = User(username='insp', full_name='Reporter', email='insp@x.com',
|
|
role='inspector', active=True, password_set=True)
|
|
reporter.set_password('pw-correct1')
|
|
fac = Facility(name='Main Office', active=True)
|
|
db.session.add_all([admin, reporter, fac])
|
|
db.session.commit()
|
|
|
|
issue = Issue(facility_id=fac.id, severity='high', description='Leaking faucet',
|
|
status='open', reported_by=reporter.id)
|
|
db.session.add(issue)
|
|
db.session.commit()
|
|
return admin.id, reporter.id, issue.id
|
|
|
|
|
|
def _login(client, username):
|
|
return client.post('/auth/login', data={'username': username, 'password': 'pw-correct1'})
|
|
|
|
|
|
def test_full_work_order_lifecycle(client):
|
|
from app import db
|
|
from app.models.work_order import IssueWorkOrder
|
|
from app.models.issue import Issue
|
|
from app.models.notification import Notification
|
|
|
|
admin_id, reporter_id, issue_id = _seed()
|
|
|
|
# 1. Manager dispatches the work order.
|
|
_login(client, 'mgr')
|
|
resp = client.post(f'/issues/{issue_id}/work-order',
|
|
data={'vendor_name': 'Ace Plumbing',
|
|
'vendor_email': 'ace@plumbing.com',
|
|
'message': 'Please fix ASAP'})
|
|
assert resp.status_code == 302 # redirect back to issue view
|
|
|
|
wo = IssueWorkOrder.query.filter_by(issue_id=issue_id).first()
|
|
assert wo is not None and wo.status == 'sent'
|
|
assert db.session.get(Issue, issue_id).status == 'in_progress'
|
|
token = wo.token
|
|
client.get('/auth/logout')
|
|
|
|
# 2. Contractor opens the public page (no login) and acknowledges.
|
|
assert client.get(f'/work-orders/{token}').status_code == 200
|
|
ack = client.post(f'/work-orders/{token}', data={'action': 'acknowledge'})
|
|
assert ack.status_code == 302
|
|
assert db.session.get(IssueWorkOrder, wo.id).status == 'acknowledged'
|
|
|
|
# Reporter was notified.
|
|
n1 = Notification.query.filter_by(user_id=reporter_id,
|
|
event_type='work_order_update').count()
|
|
assert n1 >= 1
|
|
|
|
# 3. Contractor marks it complete with a note.
|
|
done = client.post(f'/work-orders/{token}',
|
|
data={'action': 'complete', 'vendor_note': 'Replaced washer'})
|
|
assert done.status_code == 302
|
|
wo2 = db.session.get(IssueWorkOrder, wo.id)
|
|
assert wo2.status == 'completed'
|
|
assert wo2.vendor_note == 'Replaced washer'
|
|
# Issue is now awaiting staff verification.
|
|
assert db.session.get(Issue, issue_id).status == 'pending_verification'
|
|
# Reporter notified again.
|
|
assert Notification.query.filter_by(user_id=reporter_id,
|
|
event_type='work_order_update').count() >= 2
|
|
|
|
|
|
def test_unknown_token_404(client):
|
|
_seed()
|
|
assert client.get('/work-orders/definitely-not-a-real-token').status_code == 404
|
|
|
|
|
|
def test_dispatch_requires_valid_email(client):
|
|
from app.models.work_order import IssueWorkOrder
|
|
_, _, issue_id = _seed()
|
|
_login(client, 'mgr')
|
|
client.post(f'/issues/{issue_id}/work-order',
|
|
data={'vendor_name': 'Ace', 'vendor_email': 'not-an-email'})
|
|
assert IssueWorkOrder.query.count() == 0 # rejected, nothing created
|
|
|
|
|
|
def test_completed_order_ignores_further_actions(client):
|
|
from app import db
|
|
from app.models.work_order import IssueWorkOrder
|
|
|
|
_, _, issue_id = _seed()
|
|
_login(client, 'mgr')
|
|
client.post(f'/issues/{issue_id}/work-order',
|
|
data={'vendor_name': 'Ace', 'vendor_email': 'ace@x.com'})
|
|
token = IssueWorkOrder.query.first().token
|
|
client.get('/auth/logout')
|
|
|
|
client.post(f'/work-orders/{token}', data={'action': 'complete'})
|
|
wo = IssueWorkOrder.query.first()
|
|
assert wo.status == 'completed'
|
|
completed_at = wo.completed_at
|
|
|
|
# A second 'acknowledge' must not resurrect a completed order.
|
|
client.post(f'/work-orders/{token}', data={'action': 'acknowledge'})
|
|
wo = db.session.get(IssueWorkOrder, wo.id)
|
|
assert wo.status == 'completed'
|
|
assert wo.completed_at == completed_at
|