From 6ca30c0dea9e8645d3fc67d82e5eb140bf54857c Mon Sep 17 00:00:00 2001 From: NguyenND Date: Fri, 7 Aug 2026 16:08:10 -0400 Subject: [PATCH] Aug 7 - Update: add external inspector --- app/api/comments.py | 5 +- app/api/inspections.py | 9 +- app/api/issues.py | 17 +- app/api/photos.py | 3 +- app/api/scheduled.py | 10 +- app/api/stats.py | 5 +- app/api/templates.py | 3 +- app/models/notification_matrix.py | 21 ++ app/models/user.py | 52 ++- app/routes/auth.py | 95 +++++- app/routes/broadcast.py | 6 +- app/routes/dashboard.py | 4 +- app/routes/facilities.py | 12 +- app/routes/facility_qr.py | 2 +- app/routes/inspection_schedules.py | 7 +- app/routes/inspections.py | 46 +-- app/routes/issues.py | 37 ++- app/routes/reports.py | 31 +- app/templates/auth/user_form.html | 38 ++- app/templates/auth/users.html | 19 +- app/templates/inspections/execute.html | 2 +- app/templates/issues/list.html | 2 +- .../reports/inspector_performance.html | 6 + app/utils/forms.py | 4 + app/utils/notifications.py | 11 +- app/utils/scope.py | 9 +- .../versions/phase51_external_inspector.py | 64 ++++ tests/test_external_inspector.py | 303 ++++++++++++++++++ 28 files changed, 727 insertions(+), 96 deletions(-) create mode 100644 migrations/versions/phase51_external_inspector.py create mode 100644 tests/test_external_inspector.py diff --git a/app/api/comments.py b/app/api/comments.py index cd5ddcc..a57d64d 100644 --- a/app/api/comments.py +++ b/app/api/comments.py @@ -29,7 +29,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_comments', __name__) -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} def _comment_payload(comment: IssueComment) -> dict: @@ -47,7 +48,7 @@ def _comment_payload(comment: IssueComment) -> dict: def _check_issue_access(issue: Issue, user) -> bool: """Return True if user may read/write this issue. False = 403.""" - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: diff --git a/app/api/inspections.py b/app/api/inspections.py index 00bf6ac..2a763f1 100644 --- a/app/api/inspections.py +++ b/app/api/inspections.py @@ -35,7 +35,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_inspections', __name__) -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} def _merge_form_data(existing: dict, incoming: dict) -> dict: @@ -135,7 +136,7 @@ def _resolve_schedule(schedule_id, user): logger.warning('API INSPECTIONS | unknown schedule id=%s from user=%s ' '— submitting unlinked', schedule_id, user.username) return None - if user.role == 'inspector' and sched.inspector_id != user.id: + if user.is_inspector and sched.inspector_id != user.id: logger.warning('API INSPECTIONS | schedule id=%s not assigned to user=%s ' '— submitting unlinked', schedule_id, user.username) return None @@ -282,7 +283,7 @@ def list_inspections(): query = Inspection.query # Inspectors only see their own inspections - if user.role == 'inspector': + if user.is_inspector: query = query.filter(Inspection.inspector_id == user.id) # Optional filters @@ -613,7 +614,7 @@ def update_inspection(inspection_id): if inspection is None: return api_error('Inspection not found', 404) - if user.role == 'inspector' and inspection.inspector_id != user.id: + if user.is_inspector and inspection.inspector_id != user.id: return api_error('Access denied', 403) data = request.get_json(silent=True) or {} diff --git a/app/api/issues.py b/app/api/issues.py index fac1d77..8437dbe 100644 --- a/app/api/issues.py +++ b/app/api/issues.py @@ -42,7 +42,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_issues', __name__) -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} _VALID_SEVERITY = {'low', 'medium', 'high', 'critical'} _VALID_STATUSES = {'open', 'in_progress', 'resolved', 'pending_verification'} _VALID_HANDLERS = {'internal', 'facility', 'vendor'} @@ -156,7 +157,7 @@ def list_issues(): query = Issue.query - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) if not fids: return api_ok({'issues': [], 'total': 0, 'limit': limit, 'offset': offset}) @@ -250,7 +251,7 @@ def create_issue(): if facility is None: return api_error('Facility not found', 404) - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) if not fids or facility_id not in fids: return api_error('Access denied — facility is not in your assigned contracts', 403) @@ -339,7 +340,7 @@ def get_issue(issue_id): if issue is None: return api_error('Issue not found', 404) - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: @@ -372,7 +373,7 @@ def update_issue_status(issue_id): if issue is None: return api_error('Issue not found', 404) - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: @@ -437,7 +438,7 @@ def update_issue_photos(issue_id): if issue is None: return api_error('Issue not found', 404) - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: @@ -498,7 +499,7 @@ def update_issue_result_photos(issue_id): if issue is None: return api_error('Issue not found', 404) - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: @@ -575,7 +576,7 @@ def update_issue_handler(issue_id): if issue is None: return api_error('Issue not found', 404) - if user.role == 'inspector': + if user.is_inspector: fids = get_inspector_scope(user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: diff --git a/app/api/photos.py b/app/api/photos.py index 7bacafd..b2e80e8 100644 --- a/app/api/photos.py +++ b/app/api/photos.py @@ -25,7 +25,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_photos', __name__) _ALLOWED_EXTENSIONS = {'jpg', 'jpeg', 'png', 'gif'} -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} def _allowed_file(filename: str) -> bool: diff --git a/app/api/scheduled.py b/app/api/scheduled.py index 48d5765..e4847a0 100644 --- a/app/api/scheduled.py +++ b/app/api/scheduled.py @@ -40,7 +40,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_scheduled', __name__) -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} def _scheduled_payload(s): @@ -127,7 +128,7 @@ def list_scheduled(): InspectionSchedule.mode == 'plan', ) - if user.role == 'inspector': + if user.is_inspector: # Inspectors only see schedules assigned directly to them. query = query.filter(InspectionSchedule.inspector_id == user.id) @@ -185,7 +186,8 @@ def create_follow_up(): user = g.api_user # Auditor is read-only everywhere else; keep it that way here. - if user.role not in {'admin', 'director', 'inspector', 'project_manager'}: + if user.role not in {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager'}: return api_error('Access denied', 403) body = request.get_json(silent=True) or {} @@ -201,7 +203,7 @@ def create_follow_up(): # An inspector may only schedule a follow-up of their own work, and only # within their assigned contracts — the same two gates the rest of the # mobile API applies. Managers are unrestricted, matching the web. - if user.role == 'inspector': + if user.is_inspector: if parent.inspector_id != user.id: return api_error('Access denied', 403) fids = get_inspector_scope(user) diff --git a/app/api/stats.py b/app/api/stats.py index 41abe55..fc877a3 100644 --- a/app/api/stats.py +++ b/app/api/stats.py @@ -40,7 +40,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_stats', __name__) -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} @bp.route('/stats/dashboard', methods=['GET']) @@ -74,7 +75,7 @@ def dashboard_stats(): today_end = today_start + timedelta(days=1) thirty_days_ago = now - timedelta(days=30) - is_inspector = user.role == 'inspector' + is_inspector = user.is_inspector fids = get_inspector_scope(user) if is_inspector else None # None = no scoping # ── Today's inspections ─────────────────────────────────────────────── diff --git a/app/api/templates.py b/app/api/templates.py index b1f3f1c..195de63 100644 --- a/app/api/templates.py +++ b/app/api/templates.py @@ -27,7 +27,8 @@ logger = logging.getLogger(__name__) bp = Blueprint('api_templates', __name__) # Customer role cannot access template data — inspectors and above only -_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager', 'auditor'} +_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'external_inspector', + 'project_manager', 'auditor'} def _template_summary_payload(template: InspectionTemplate) -> dict: diff --git a/app/models/notification_matrix.py b/app/models/notification_matrix.py index 40cb5e5..d6512e3 100644 --- a/app/models/notification_matrix.py +++ b/app/models/notification_matrix.py @@ -10,6 +10,14 @@ role_key values admin — all users with role='admin' director — all users with role='director' inspector — all users with role='inspector' + EXCEPTION: for event 'inspection_completed', the inspector + column notifies ONLY the inspection's own inspector + (the submitter), not the whole inspector pool. Scoping is + applied in notify_by_matrix() via the inspection_id. +external_inspector — all users with role='external_inspector' (customer / + third-party inspectors). Separate column so third parties + can be routed differently from the tenant's own crew; the + 'inspection_completed' scoping above applies here too. project_manager — all users with role='project_manager' customer — all customer-portal users assigned to the relevant facility assignee — the specific user the issue/inspection is assigned to @@ -40,6 +48,7 @@ MATRIX_ROLES = [ ('admin', 'Admin'), ('director', 'Director'), ('inspector', 'Inspector'), + ('external_inspector', 'External Inspector'), ('project_manager', 'Project Manager'), ('auditor', 'Auditor'), ('customer', 'Customer'), @@ -171,6 +180,18 @@ MATRIX_DEFAULTS = { ('score_alert', 'custom'): False, } +# MT-15 — the External Inspector column defaults to whatever the internal +# Inspector column defaults to, for every event. Mirroring rather than listing +# 14 more literals means a future event added for 'inspector' automatically +# gets a matching external default instead of silently falling back to the +# is_enabled() fallback. Admins can diverge the two columns in the UI at any +# time; this only seeds rows that do not exist yet. +MATRIX_DEFAULTS.update({ + (_event, 'external_inspector'): _enabled + for (_event, _role), _enabled in list(MATRIX_DEFAULTS.items()) + if _role == 'inspector' +}) + class NotificationMatrix(db.Model): """Admin-controlled per-event notification routing.""" diff --git a/app/models/user.py b/app/models/user.py index a145edf..77f0a99 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -3,6 +3,20 @@ from flask_login import UserMixin from werkzeug.security import generate_password_hash, check_password_hash from app.utils.time_utils import now_eastern +# Display labels for the role ENUM. 'external_inspector' would otherwise title +# case to "External Inspector" anyway, but the map keeps every label in one +# place for templates that show a role name. +ROLE_LABELS = { + 'admin': 'Admin', + 'director': 'Director', + 'project_manager': 'Project Manager', + 'auditor': 'Auditor', + 'inspector': 'Inspector', + 'external_inspector': 'External Inspector', + 'customer': 'Customer', +} + + @login_manager.user_loader def load_user(user_id): from app import db @@ -11,6 +25,19 @@ def load_user(user_id): class User(UserMixin, db.Model): __tablename__ = 'users' + # ── Inspector roles (MT-15) ─────────────────────────────────────────────── + # 'external_inspector' is an inspector employed by the customer or a third + # party rather than by the tenant. It has exactly the same capabilities as + # the internal 'inspector' role and is scoped the same way — through + # InspectorAssignment rows, via get_inspector_scope(). + # + # Every place that used to test `role == 'inspector'` must test membership + # of this tuple instead, or external inspectors silently fall into the + # privileged (org-wide) branch and see every contract. Use the + # `is_inspector` property below — it is an ordinary attribute, so it reads + # the same way in Python and in Jinja (`current_user.is_inspector`). + INSPECTOR_ROLES = ('inspector', 'external_inspector') + id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(100), unique=True, nullable=False, index=True) full_name = db.Column(db.String(150), nullable=True) @@ -19,7 +46,10 @@ class User(UserMixin, db.Model): role = db.Column( # Phase 11 migration complete — 'supervisor' removed from both the DB # ENUM and this Python-side declaration. Director is the canonical role. - db.Enum('admin', 'director', 'inspector', 'project_manager', 'customer', 'auditor'), + # MT-15 — 'external_inspector' added: a customer / third-party + # inspector with identical capabilities to 'inspector'. + db.Enum('admin', 'director', 'inspector', 'project_manager', 'customer', + 'auditor', 'external_inspector'), nullable=False ) created_at = db.Column(db.DateTime, default=now_eastern) @@ -64,6 +94,26 @@ class User(UserMixin, db.Model): def check_password(self, password): return check_password_hash(self.password_hash, password) + @property + def is_inspector(self): + """True for both the internal and the external inspector role. + + Prefer this over `role == 'inspector'` for capability and scoping + checks. Use an explicit `role == 'external_inspector'` test only where + the two genuinely differ (currently: display labelling only). + """ + return self.role in self.INSPECTOR_ROLES + + @property + def is_external_inspector(self): + """True only for third-party / customer-employed inspectors.""" + return self.role == 'external_inspector' + + @property + def role_label(self): + """Human-readable role name, used in staff-facing lists.""" + return ROLE_LABELS.get(self.role, (self.role or '').replace('_', ' ').title()) + @property def display_name(self): """Return full name if set, otherwise fall back to username.""" diff --git a/app/routes/auth.py b/app/routes/auth.py index cec5abd..b031719 100644 --- a/app/routes/auth.py +++ b/app/routes/auth.py @@ -431,20 +431,64 @@ def create_user(): if form.validate_on_submit(): role = 'inspector' if director_editing else form.role.data + + # MT-15 — an external inspector works for the customer or a third + # party, so we never set a password on their behalf. They are invited + # exactly like a customer: created with password_set=False (which the + # login route refuses until they finish), given a one-time token, and + # emailed a link to choose their own password. + invite = (role == 'external_inspector') + + # UserForm.password is Optional() because the same form is used for + # EDIT, where blank means "keep current". On CREATE a blank password + # would otherwise store the hash of an empty string, so require one + # unless the account is being invited to choose their own. + if not invite and not form.password.data: + flash('Please set a password, or choose the External Inspector role ' + 'to send an invitation instead.', 'danger') + return render_template('auth/user_form.html', form=form, user=None, + title='Create User', + director_editing=director_editing) + user = User( username=form.username.data, full_name=form.full_name.data.strip() or None, email=form.email.data.strip().lower(), - role=role + role=role, + password_set=not invite, ) - user.set_password(form.password.data) + if invite: + # A random unguessable placeholder — password_set=False already + # blocks login, but never leave an account holding a known or + # empty-string hash. + import secrets + user.set_password(secrets.token_hex(32)) + else: + user.set_password(form.password.data) db.session.add(user) + db.session.flush() # need user.id before minting the token + + token = user.generate_set_password_token(expires_hours=72) if invite else None db.session.commit() - logger.info('AUTH | user_create | admin_id=%s admin=%s new_user=%s role=%s', - current_user.id, current_user.username, user.username, user.role) + + logger.info('AUTH | user_create | admin_id=%s admin=%s new_user=%s role=%s invite=%s', + current_user.id, current_user.username, user.username, + user.role, invite) log_action(ACTION_CREATE, 'User', user.id, user.username, - f'role={user.role}; email={user.email}') - flash(f'User {user.username} created successfully.', 'success') + f'role={user.role}; email={user.email}; invite_sent={invite}') + + if invite: + # Reuses the customer invitation email — the copy ("an account has + # been created for you… set your password") is already correct for + # any invited account. Imported inside the function to keep the + # auth ↔ customers import graph acyclic. + from app.routes.customers import _send_invite_email + _send_invite_email(user, token, base_url=request.host_url) + flash(f'External inspector {user.display_name} created. An invitation ' + f'email has been sent to {user.email} with a link to set their ' + f'password.', 'success') + else: + flash(f'User {user.username} created successfully.', 'success') return redirect(url_for('auth.list_users')) return render_template('auth/user_form.html', form=form, title='Create User', @@ -488,12 +532,49 @@ def edit_user(user_id): title='Edit User', director_editing=director_editing) +@bp.route('/users//resend-invite', methods=['POST']) +@login_required +@admin_required +def resend_invite(user_id): + """Re-send the set-password invitation for an account still awaiting setup. + + Without this an invitation that bounces, is deleted or expires leaves the + account permanently unusable — password_set=False blocks login and only a + valid token can clear it. Mirrors customers.resend_invite for staff-side + accounts (currently only external inspectors are ever invited this way). + """ + user = db.session.get(User, user_id) + if user is None: + abort(404) + if user.password_set: + flash(f'{user.display_name} has already completed their account setup.', + 'info') + return redirect(url_for('auth.list_users')) + + # A fresh token invalidates the previous link. + token = user.generate_set_password_token(expires_hours=72) + db.session.commit() + + logger.info('AUTH | resend_invite | admin=%s user=%s', + current_user.username, user.username) + log_action(ACTION_UPDATE, 'User', user.id, user.username, + 'invitation email resent') + + from app.routes.customers import _send_invite_email + _send_invite_email(user, token, base_url=request.host_url) + + flash(f'Invitation resent to {user.email}.', 'success') + return redirect(url_for('auth.list_users')) + + @bp.route('/users//assign-contracts', methods=['GET', 'POST']) @login_required @admin_required def assign_inspector_contracts(user_id): user = db.session.get(User, user_id) - if user is None or user.role != 'inspector': + # MT-15: external inspectors are scoped by the same InspectorAssignment + # rows, so this page must accept them too. + if user is None or not user.is_inspector: abort(404) from app.models.project import Project diff --git a/app/routes/broadcast.py b/app/routes/broadcast.py index 3d5e608..32e7f0c 100644 --- a/app/routes/broadcast.py +++ b/app/routes/broadcast.py @@ -22,10 +22,12 @@ logger = logging.getLogger(__name__) bp = Blueprint('broadcast', __name__, url_prefix='/admin/broadcast') # All roles that can hold an active iOS session -BROADCAST_ROLES = ['inspector', 'project_manager', 'director', 'admin'] +BROADCAST_ROLES = ['inspector', 'external_inspector', 'project_manager', + 'director', 'admin'] ROLE_LABELS = { - 'inspector': 'Inspectors', + 'inspector': 'Inspectors', + 'external_inspector': 'External Inspectors', 'project_manager': 'Project Managers', 'director': 'Directors', 'admin': 'Admins', diff --git a/app/routes/dashboard.py b/app/routes/dashboard.py index d41d22a..df389fc 100644 --- a/app/routes/dashboard.py +++ b/app/routes/dashboard.py @@ -25,7 +25,7 @@ def index(): today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) today_end = today_start + timedelta(days=1) - is_inspector = current_user.role == 'inspector' + is_inspector = current_user.is_inspector is_privileged = current_user.role in ['admin', 'director'] is_customer = current_user.role == 'customer' is_project_manager = current_user.role == 'project_manager' @@ -305,7 +305,7 @@ def index(): if is_privileged or is_project_manager or is_auditor: active_inspectors = ( User.query - .filter_by(role='inspector', active=True) + .filter(User.role.in_(User.INSPECTOR_ROLES), User.active == True) .order_by(User.full_name, User.username) .all() ) diff --git a/app/routes/facilities.py b/app/routes/facilities.py index b5fbaf8..7e55e05 100644 --- a/app/routes/facilities.py +++ b/app/routes/facilities.py @@ -23,7 +23,7 @@ def list_facilities(): facilities = Facility.query.filter( Facility.id.in_(cids), Facility.active == True ).order_by(Facility.name).all() - elif current_user.role == 'inspector': + elif current_user.is_inspector: fids = get_inspector_scope(current_user) or [] facilities = Facility.query.filter( Facility.id.in_(fids), Facility.active == True @@ -262,7 +262,7 @@ def _facility_for_qr_or_403(facility_id): if facility is None: abort(404) # QR management is not an inspector task (matches qr_print_all/qr_export_pdf). - if current_user.role == 'inspector': + if current_user.is_inspector: abort(403) if current_user.role == 'customer': cids = get_customer_scope(current_user) or [] @@ -329,7 +329,7 @@ def qr_sheet(): """Bulk print sheet — one labeled QR card per active facility.""" from app.utils.qr import qr_svg - if current_user.role == 'inspector': + if current_user.is_inspector: abort(403) if current_user.role == 'customer': @@ -395,7 +395,7 @@ def _area_for_qr_or_403(area_id): if area is None: abort(404) # QR management is not an inspector task (matches qr_print_all/qr_export_pdf). - if current_user.role == 'inspector': + if current_user.is_inspector: abort(403) if current_user.role == 'customer': cids = get_customer_scope(current_user) or [] @@ -483,7 +483,7 @@ def qr_print_all(): Inspectors have no QR management (403); customers are scoped to their assigned facilities; managers see all active facilities. """ - if current_user.role == 'inspector': + if current_user.is_inspector: abort(403) contract_id = request.args.get('contract_id', type=int) @@ -552,7 +552,7 @@ def qr_export_pdf(): Scope is enforced per-id via the same helpers as the QR pages, so a customer can never export a code outside their assigned facilities. """ - if current_user.role == 'inspector': + if current_user.is_inspector: abort(403) facility_ids = request.form.getlist('facility_ids', type=int) diff --git a/app/routes/facility_qr.py b/app/routes/facility_qr.py index c3d33e0..520a07c 100644 --- a/app/routes/facility_qr.py +++ b/app/routes/facility_qr.py @@ -125,7 +125,7 @@ def _can_view_full(facility): return False if current_user.role in ('admin', 'director', 'project_manager'): return True - if current_user.role == 'inspector': + if current_user.is_inspector: return facility.id in (get_inspector_scope(current_user) or []) if current_user.role == 'customer': return facility.id in (get_customer_scope(current_user) or []) diff --git a/app/routes/inspection_schedules.py b/app/routes/inspection_schedules.py index 602bd82..213e6ea 100644 --- a/app/routes/inspection_schedules.py +++ b/app/routes/inspection_schedules.py @@ -189,7 +189,8 @@ def _active_inspectors(): """Users who can be assigned inspections (inspector-capable roles).""" return User.query.filter( User.active.is_(True), - User.role.in_(['inspector', 'project_manager', 'director', 'admin']), + User.role.in_(['inspector', 'external_inspector', 'project_manager', + 'director', 'admin']), ).order_by(User.full_name, User.username).all() @@ -356,7 +357,7 @@ def index(): base = InspectionSchedule.query # Inspectors see only their own assignments; managers see everything. - if current_user.role == 'inspector': + if current_user.is_inspector: base = base.filter(InspectionSchedule.inspector_id == current_user.id) # Counts are computed on the same scoped query, so the badges match what the @@ -645,7 +646,7 @@ def start(schedule_id): abort(404) if current_user.role == 'customer': abort(403) - if current_user.role == 'inspector' and schedule.inspector_id != current_user.id: + if current_user.is_inspector and schedule.inspector_id != current_user.id: abort(403) if not schedule.active: diff --git a/app/routes/inspections.py b/app/routes/inspections.py index 9f50176..4a30a7c 100644 --- a/app/routes/inspections.py +++ b/app/routes/inspections.py @@ -216,7 +216,7 @@ def index(): joinedload(Inspection.area), ).order_by(Inspection.inspection_date.desc()) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) if not fids: q = q.filter(False) @@ -285,12 +285,12 @@ def index(): q = q.filter(Inspection.overall_score <= float(score_max_filter)) except ValueError: score_max_filter = '' - if inspector_filter.isdigit() and current_user.role != 'inspector': + if inspector_filter.isdigit() and not current_user.is_inspector: q = q.filter(Inspection.inspector_id == int(inspector_filter)) inspections = q.paginate(page=page, per_page=20, error_out=False) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) or [] _fq = Facility.query.filter(Facility.id.in_(fids), Facility.active == True) elif current_user.role == 'customer': @@ -317,9 +317,9 @@ def index(): projects = Project.query.filter_by(active=True).order_by(Project.name).all() # Inspector dropdown — shown to all roles except inspector (they only see their own) - if current_user.role != 'inspector': + if not current_user.is_inspector: inspectors = (User.query - .filter(User.role == 'inspector', User.active == True) + .filter(User.role.in_(User.INSPECTOR_ROLES), User.active == True) .order_by(User.full_name, User.username).all()) else: inspectors = [] @@ -354,7 +354,7 @@ def start(): projects = Project.query.filter_by(active=True).order_by(Project.name).all() # Scope projects to inspector's assigned contracts - if current_user.role == 'inspector': + if current_user.is_inspector: from app.models.inspector_assignment import InspectorAssignment assigned_pids = { a.project_id for a in @@ -410,7 +410,7 @@ def start(): # Inspector facility scope check — prevent crafted POST from selecting # a facility outside their assigned contracts. - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) if not fids or form.facility_id.data not in fids: abort(403) @@ -474,7 +474,7 @@ def execute(inspection_id): if inspection is None: abort(404) - if current_user.role == 'inspector' and inspection.inspector_id != current_user.id: + if current_user.is_inspector and inspection.inspector_id != current_user.id: flash('Access denied.', 'danger') return redirect(url_for('inspections.index')) @@ -610,7 +610,8 @@ def execute(inspection_id): return redirect(url_for('inspections.execute', inspection_id=inspection_id)) staff_for_flag_issue = User.query.filter( - User.role.in_(['director', 'inspector', 'project_manager', 'auditor']), + User.role.in_(['director', 'inspector', 'external_inspector', + 'project_manager', 'auditor']), User.active == True, ).order_by(User.full_name, User.username).all() @@ -650,7 +651,7 @@ def save_draft_ajax(inspection_id): if inspection is None: abort(404) - if current_user.role == 'inspector' and inspection.inspector_id != current_user.id: + if current_user.is_inspector and inspection.inspector_id != current_user.id: return jsonify({'ok': False, 'error': 'Access denied'}), 403 if inspection.status == 'completed': @@ -689,7 +690,7 @@ def upload_photo_ajax(inspection_id): if inspection is None: return jsonify({'ok': False, 'error': 'Not found'}), 404 - if current_user.role == 'inspector' and inspection.inspector_id != current_user.id: + if current_user.is_inspector and inspection.inspector_id != current_user.id: return jsonify({'ok': False, 'error': 'Access denied'}), 403 if inspection.status == 'completed': @@ -716,7 +717,7 @@ def view(inspection_id): if inspection is None: abort(404) - if current_user.role == 'inspector' and inspection.inspector_id != current_user.id: + if current_user.is_inspector and inspection.inspector_id != current_user.id: flash('Access denied.', 'danger') return redirect(url_for('inspections.index')) if current_user.role == 'customer': @@ -930,15 +931,22 @@ def flag_issue(inspection_id): if inspection is None: abort(404) - if current_user.role == 'inspector' and inspection.inspector_id != current_user.id: + if current_user.is_inspector and inspection.inspector_id != current_user.id: flash('Access denied.', 'danger') return redirect(url_for('inspections.index')) form = IssueForm() - staff = User.query.filter(User.role.in_(['director', 'inspector'])).order_by(User.username).all() + staff = User.query.filter( + User.role.in_(['director', 'inspector', 'external_inspector']) + ).order_by(User.username).all() form.facility_id.choices = [(inspection.facility_id, inspection.facility.name)] - form.assigned_to.choices = [(0, '— Unassigned —')] + [(u.id, u.username) for u in staff] + # MT-15: suffix external (customer / third-party) inspectors so whoever is + # triaging can see the work is going outside the company. Display only. + form.assigned_to.choices = [(0, '— Unassigned —')] + [ + (u.id, u.username + (' (External)' if u.is_external_inspector else '')) + for u in staff + ] if form.validate_on_submit(): photo_path = _save_photo(form.photo.data, subfolder='issue_photos') @@ -1020,7 +1028,7 @@ def export_list_pdf(): joinedload(Inspection.area), ).order_by(Inspection.inspection_date.desc()) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) if not fids: q = q.filter(False) @@ -1083,7 +1091,7 @@ def export_list_pdf(): q = q.filter(Inspection.overall_score <= float(score_max_filter)) except ValueError: pass - if inspector_filter.isdigit() and current_user.role != 'inspector': + if inspector_filter.isdigit() and not current_user.is_inspector: q = q.filter(Inspection.inspector_id == int(inspector_filter)) inspections = q.all() @@ -1113,7 +1121,7 @@ def export_list_pdf(): filter_parts.append(f'Min score: {score_min_filter}%') if score_max_filter: filter_parts.append(f'Max score: {score_max_filter}%') - if inspector_filter.isdigit() and current_user.role != 'inspector': + if inspector_filter.isdigit() and not current_user.is_inspector: u = db.session.get(User, int(inspector_filter)) if u: filter_parts.append(f'Inspector: {u.display_name}') @@ -1143,7 +1151,7 @@ def export_pdf(inspection_id): if inspection is None: abort(404) - if current_user.role == 'inspector' and inspection.inspector_id != current_user.id: + if current_user.is_inspector and inspection.inspector_id != current_user.id: flash('Access denied.', 'danger') return redirect(url_for('inspections.index')) if current_user.role == 'customer': diff --git a/app/routes/issues.py b/app/routes/issues.py index 0465a5e..381d190 100644 --- a/app/routes/issues.py +++ b/app/routes/issues.py @@ -71,6 +71,18 @@ class _SLAFilteredPage: return iter([1]) +def _assignee_label(user): + """Dropdown label for an assignee. + + MT-15 — external (customer / third-party) inspectors are assignable just + like the tenant's own crew, but are suffixed so whoever is triaging can see + at a glance that the work is going outside the company. Display only; the + stored value is still the user id. + """ + return (f'{user.display_name} (External)' + if user.is_external_inspector else user.display_name) + + @bp.route('/export-list-pdf') @login_required def export_list_pdf(): @@ -86,7 +98,7 @@ def export_list_pdf(): .order_by(Issue.reported_at.desc()) ) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) if not fids: q = q.filter(False) @@ -211,7 +223,7 @@ def index(): .order_by(Issue.reported_at.desc()) ) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) if not fids: q = q.filter(False) @@ -319,7 +331,7 @@ def index(): # Facilities for the filter dropdown — scoped for inspectors/customers, # then narrowed to the selected contract when contract_filter is active. - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) or [] _fq = Facility.query.filter(Facility.id.in_(fids), Facility.active == True) elif current_user.role == 'customer': @@ -347,7 +359,8 @@ def index(): # Staff for quick-assign dropdown — same roles as the full issue form staff = User.query.filter( - User.role.in_(['director', 'inspector', 'auditor']), User.active == True + User.role.in_(['director', 'inspector', 'external_inspector', 'auditor']), + User.active == True ).order_by(User.username).all() # Reporters dropdown — users who have actually filed at least one issue @@ -385,7 +398,7 @@ def view(issue_id): if issue is None: abort(404) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: @@ -422,7 +435,7 @@ def view(issue_id): return redirect(url_for('issues.view', issue_id=issue_id)) form = IssueUpdateForm(obj=issue) - staff = User.query.filter(User.role.in_(['director', 'inspector', 'auditor'])).order_by(User.username).all() + staff = User.query.filter(User.role.in_(['director', 'inspector', 'external_inspector', 'auditor'])).order_by(User.username).all() # Preserve any pre-existing assignee who is no longer in the assignable set # (e.g. an admin assigned before admins were removed from the dropdown) so # saving the form doesn't silently unassign them. @@ -430,7 +443,9 @@ def view(issue_id): current_assignee = db.session.get(User, issue.assigned_to) if current_assignee: staff.append(current_assignee) - form.assigned_to.choices = [(0, '— Unassigned —')] + [(u.id, u.display_name) for u in staff] + form.assigned_to.choices = [(0, '— Unassigned —')] + [ + (u.id, _assignee_label(u)) for u in staff + ] form.status.data = form.status.data or issue.status if form.validate_on_submit(): @@ -743,10 +758,12 @@ def create(): else: facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all() projects = Project.query.filter_by(active=True).order_by(Project.name).all() - staff = User.query.filter(User.role.in_(['director', 'inspector', 'auditor'])).order_by(User.username).all() + staff = User.query.filter(User.role.in_(['director', 'inspector', 'external_inspector', 'auditor'])).order_by(User.username).all() form.facility_id.choices = [(f.id, f.name) for f in facilities] - form.assigned_to.choices = [(0, '— Unassigned —')] + [(u.id, u.display_name) for u in staff] + form.assigned_to.choices = [(0, '— Unassigned —')] + [ + (u.id, _assignee_label(u)) for u in staff + ] # On POST validation error: identify which contract the submitted facility # belongs to so the contract selector can be restored on re-render. @@ -1129,7 +1146,7 @@ def export_pdf(issue_id): if issue is None: abort(404) - if current_user.role == 'inspector': + if current_user.is_inspector: fids = get_inspector_scope(current_user) facility = issue.resolved_facility if not fids or not facility or facility.id not in fids: diff --git a/app/routes/reports.py b/app/routes/reports.py index 362da50..e3a3da7 100644 --- a/app/routes/reports.py +++ b/app/routes/reports.py @@ -57,7 +57,8 @@ def index(): # Inspectors get a scoped view of their own inspections and related issues. # Customers get a facility-scoped view. # Internal management roles (director+) get the full unscoped view. - if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']: + if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', + 'external_inspector', 'customer']: from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('dashboard.index')) @@ -66,7 +67,7 @@ def index(): # Resolve scoping for customers (facility list) and inspectors (inspector_id) customer_facility_ids = get_customer_scope(current_user) # None = unrestricted - is_inspector = current_user.role == 'inspector' + is_inspector = current_user.is_inspector # Inspector filter — admin / director / project_manager only inspector_filter = None @@ -262,7 +263,8 @@ def index(): inspectors = [] if current_user.role in ('admin', 'director', 'project_manager'): - inspectors = User.query.filter_by(role='inspector', active=True)\ + inspectors = User.query.filter(User.role.in_(User.INSPECTOR_ROLES), + User.active == True)\ .order_by(User.full_name, User.username).all() facility_scores_list = [{ @@ -307,7 +309,8 @@ def index(): @bp.route('/facility/') @login_required def facility_report(facility_id): - if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']: + if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', + 'external_inspector', 'customer']: from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('dashboard.index')) @@ -320,7 +323,7 @@ def facility_report(facility_id): from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('reports.index')) - if current_user.role == 'inspector': + if current_user.is_inspector: # Inspectors may only view the facility report for facilities where # they have personally conducted at least one inspection. has_access = Inspection.query.filter_by( @@ -369,7 +372,8 @@ def facility_report(facility_id): def facility_scorecard(facility_id): """Comprehensive per-facility scorecard: score trend, SLA compliance, issue breakdown by severity, inspection frequency.""" - if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']: + if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', + 'external_inspector', 'customer']: from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('dashboard.index')) @@ -385,7 +389,7 @@ def facility_scorecard(facility_id): flash('Access denied.', 'danger') return redirect(url_for('reports.index')) - if current_user.role == 'inspector': + if current_user.is_inspector: has_access = Inspection.query.filter_by( facility_id=facility_id, inspector_id=current_user.id, @@ -701,7 +705,8 @@ def _build_inspector_stats(start, end): all_ids = set(total_map.keys()) active_inspectors = ( User.query - .filter(User.id.in_(all_ids), User.active == True, User.role == 'inspector') + .filter(User.id.in_(all_ids), User.active == True, + User.role.in_(User.INSPECTOR_ROLES)) .order_by(User.full_name, User.username) .all() ) if all_ids else [] @@ -717,6 +722,10 @@ def _build_inspector_stats(start, end): inspector_stats.append({ 'id': u.id, 'display_name': u.display_name, + # MT-15 — customer / third-party inspectors appear in the same + # table as the tenant's own crew, badged so the numbers can be read + # in context. Consumed by the HTML table and the export. + 'external': u.is_external_inspector, 'total': tot, 'completed': comp, 'completion_rate': round(comp / tot * 100) if tot else 0, @@ -759,7 +768,7 @@ def inspector_performance(): if selected_id: selected_inspector = db.session.get(User, selected_id) - if selected_inspector and selected_inspector.role == 'inspector': + if selected_inspector and selected_inspector.is_inspector: selected_kpis = next((s for s in inspector_stats if s['id'] == selected_id), None) trend_rows = db.session.query( @@ -857,7 +866,7 @@ def export_inspector_performance(): .filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, - User.role == 'inspector', + User.role.in_(User.INSPECTOR_ROLES), ) if selected_id: detail_q = detail_q.filter(Inspection.inspector_id == selected_id) @@ -1628,7 +1637,7 @@ def facility_summary_pdf(facility_id): cids = get_customer_scope(current_user) or [] if facility_id not in cids: abort(403) - elif current_user.role == 'inspector': + elif current_user.is_inspector: has = Inspection.query.filter_by(facility_id=facility_id, inspector_id=current_user.id).first() if not has: diff --git a/app/templates/auth/user_form.html b/app/templates/auth/user_form.html index ed04820..7c60799 100644 --- a/app/templates/auth/user_form.html +++ b/app/templates/auth/user_form.html @@ -80,7 +80,20 @@ -
+ {# MT-15 — an External Inspector is invited by email and chooses + their own password, so the admin never sets one. The JS at the + foot of this page swaps these two blocks when the role changes; + the server decides independently of the JS. #} +
+ + This account will be invited by email. + External inspectors work outside the business, so we do not set + a password for them. On save, an invitation is sent to the email + address above with a link to choose their own password. The link + is valid for 72 hours. +
+ +
{{ form.password.label(class="form-label") }} {{ form.password(class="form-control", placeholder="Leave blank to keep current" if user else "") }} @@ -128,4 +141,27 @@
+ + {% endblock %} \ No newline at end of file diff --git a/app/templates/auth/users.html b/app/templates/auth/users.html index 21e8f97..99c6a38 100644 --- a/app/templates/auth/users.html +++ b/app/templates/auth/users.html @@ -37,12 +37,12 @@ {{ user.full_name or '—' }} {{ user.email }} - - {{ user.role.replace('_',' ')|title }} + + {{ user.role_label }} - {% if user.role == 'inspector' %} + {% if user.is_inspector %} {% set cnt = inspector_contract_counts.get(user.id, 0) %} {% if cnt > 0 %} {{ cnt }} contract{{ 's' if cnt != 1 else '' }} @@ -65,12 +65,23 @@ - {% if user.role == 'inspector' %} + {% if user.is_inspector %} {% endif %} + {% if not user.password_set %} +
+ + +
+ {% endif %} {% if user.id != current_user.id %}
diff --git a/app/templates/inspections/execute.html b/app/templates/inspections/execute.html index 42ab574..b9f494e 100644 --- a/app/templates/inspections/execute.html +++ b/app/templates/inspections/execute.html @@ -564,7 +564,7 @@ {% set staff = staff_for_flag_issue %} {% if staff %}{% for u in staff %} - + {% endfor %}{% endif %} diff --git a/app/templates/issues/list.html b/app/templates/issues/list.html index 1a7113e..c51a119 100644 --- a/app/templates/issues/list.html +++ b/app/templates/issues/list.html @@ -181,7 +181,7 @@ diff --git a/app/templates/reports/inspector_performance.html b/app/templates/reports/inspector_performance.html index a85ddf7..64648a5 100644 --- a/app/templates/reports/inspector_performance.html +++ b/app/templates/reports/inspector_performance.html @@ -115,6 +115,9 @@ data-inspector-id="{{ s.id }}"> {{ s.display_name }} + {% if s.external %} + External + {% endif %} {{ s.total }} {{ s.completed }} @@ -189,6 +192,9 @@
{{ selected_inspector.display_name }} + {% if selected_inspector.is_external_inspector %} + External + {% endif %}
diff --git a/app/utils/forms.py b/app/utils/forms.py index b7d403e..640057d 100644 --- a/app/utils/forms.py +++ b/app/utils/forms.py @@ -91,6 +91,10 @@ class UserForm(FlaskForm): ('admin', 'Administrator'), ('director', 'Director'), ('inspector', 'Inspector'), + # MT-15 — an inspector employed by the customer or a third party. + # Same capabilities as 'inspector'; scoped to the contracts assigned on + # the Assign Contracts page (see User.INSPECTOR_ROLES). + ('external_inspector', 'External Inspector'), ('project_manager', 'Project Manager'), ('auditor', 'Auditor'), # 'customer' is intentionally excluded — customer accounts are managed via /customers diff --git a/app/utils/notifications.py b/app/utils/notifications.py index d399dc7..23c5f6b 100644 --- a/app/utils/notifications.py +++ b/app/utils/notifications.py @@ -570,7 +570,8 @@ def notify_by_matrix( role_to_db = { 'admin': 'admin', 'director': 'director', - 'inspector': 'inspector', + 'inspector': 'inspector', + 'external_inspector': 'external_inspector', 'project_manager': 'project_manager', 'auditor': 'auditor', 'customer': 'customer', @@ -602,15 +603,17 @@ def notify_by_matrix( # a tenant with a dozen inspectors is a mail storm and trains people to # ignore notifications. Falls back to notifying nobody when the # inspection cannot be resolved, rather than notifying everybody. - if role_key == 'inspector' and event_type == 'inspection_completed': + if (role_key in ('inspector', 'external_inspector') + and event_type == 'inspection_completed'): target_id = None if inspection_id: from app.models.inspection import Inspection insp = db.session.get(Inspection, inspection_id) target_id = insp.inspector_id if insp else None users = [u for u in users if u.id == target_id] if target_id else [] - logger.info('MATRIX NOTIFY | event=%s | role=inspector scoped to ' - 'submitting inspector_id=%s', event_type, target_id) + logger.info('MATRIX NOTIFY | event=%s | role=%s scoped to ' + 'submitting inspector_id=%s', + event_type, role_key, target_id) # Scope customer role to facility if provided if role_key == 'customer' and facility_id: diff --git a/app/utils/scope.py b/app/utils/scope.py index 2748b47..7d4952d 100644 --- a/app/utils/scope.py +++ b/app/utils/scope.py @@ -8,6 +8,8 @@ Facility-scoping utilities for the Janitorial QC portal. get_inspector_scope(user) -> list[int] | None Facility IDs an inspector may access via InspectorAssignment rows. + Applies to BOTH 'inspector' (internal) and 'external_inspector' + (customer / third-party) — see User.INSPECTOR_ROLES. Returns [] (empty list) when the inspector has no contract assignments, meaning they see nothing (strict mode). @@ -90,7 +92,12 @@ def get_inspector_scope(user) -> list[int] | None: None Returned for non-inspector roles, indicating unrestricted access. """ - if user.role != 'inspector': + # MT-15: covers BOTH 'inspector' and 'external_inspector'. An external + # (customer / third-party) inspector is scoped by exactly the same + # InspectorAssignment rows — the contracts an admin grants them. + from app.models.user import User + + if user.role not in User.INSPECTOR_ROLES: return None from app.models.inspector_assignment import InspectorAssignment diff --git a/migrations/versions/phase51_external_inspector.py b/migrations/versions/phase51_external_inspector.py new file mode 100644 index 0000000..114db3c --- /dev/null +++ b/migrations/versions/phase51_external_inspector.py @@ -0,0 +1,64 @@ +"""phase51 — add 'external_inspector' role to users.role ENUM + +Introduces the External Inspector role: an inspector employed by the customer +or a third party rather than by the tenant. It has exactly the same +capabilities as the internal 'inspector' role and is scoped the same way — +through InspectorAssignment rows, resolved by get_inspector_scope(). + +Every capability/scoping check that used to test `role == 'inspector'` now +tests membership of User.INSPECTOR_ROLES (exposed as the `is_inspector` +property), so the new role picks up inspector behaviour everywhere without a +per-route allowlist. + +This is a pure ENUM expansion (adds a value, removes and migrates nothing), so +the 3-step ENUM protocol does not apply and re-running the same MODIFY is a +no-op — safe to re-run. + +MULTI-TENANT NOTE +----------------- +This runs once per tenant database via control/tenant_migrate.py, like every +other script in migrations/versions. It is a metadata-only ALTER on a small +table, but MySQL still takes an exclusive metadata lock for the duration, so +run it during the normal migration window rather than under load. + +Notification matrix rows for the new 'external_inspector' column are NOT seeded +here: MATRIX_DEFAULTS mirrors the Inspector column at runtime and is_enabled() +falls back to that default when a row is absent, so an unseeded tenant behaves +exactly like the Inspector column until an admin saves the matrix page. +""" + +revision = 'phase51_external_inspector' +down_revision = 'phase50_sched_acknowledged' +branch_labels = None +depends_on = None + +from alembic import op +import sqlalchemy as sa + + +_ENUM_WITH_EXTERNAL = ( + "ENUM('admin','director','inspector','project_manager','customer'," + "'auditor','external_inspector')" +) +_ENUM_WITHOUT_EXTERNAL = ( + "ENUM('admin','director','inspector','project_manager','customer','auditor')" +) + + +def upgrade(): + # Idempotent: MODIFY to the expanded set is harmless if already applied. + op.execute(sa.text( + f"ALTER TABLE users MODIFY COLUMN role {_ENUM_WITH_EXTERNAL} NOT NULL" + )) + + +def downgrade(): + # Reassign any external_inspector rows before contracting the ENUM so no + # account is orphaned. They become internal inspectors, which keeps their + # InspectorAssignment scoping intact — the same contracts still apply. + op.execute(sa.text( + "UPDATE users SET role = 'inspector' WHERE role = 'external_inspector'" + )) + op.execute(sa.text( + f"ALTER TABLE users MODIFY COLUMN role {_ENUM_WITHOUT_EXTERNAL} NOT NULL" + )) diff --git a/tests/test_external_inspector.py b/tests/test_external_inspector.py new file mode 100644 index 0000000..bff3c11 --- /dev/null +++ b/tests/test_external_inspector.py @@ -0,0 +1,303 @@ +""" +tests/test_external_inspector.py +-------------------------------- +Behaviour tests for MT-15 — the 'external_inspector' role. + +Runs on the in-memory SQLite app fixture (multi-tenancy inert). Covers: + + * the role predicates: is_inspector covers both roles, is_external_inspector + does not, and role_label renders the display name + * get_inspector_scope() applies the SAME InspectorAssignment scoping to an + external inspector as to an internal one — the regression this phase exists + to prevent is an external inspector falling into the privileged branch + (scope None = unrestricted) and seeing every contract + * an unassigned external inspector sees nothing (strict mode) + * facility / inspection / issue list routes stay scoped for the new role + * the notification matrix exposes an External Inspector column whose defaults + mirror the Inspector column + * creating an external inspector sends an invitation instead of setting a + password: password_set is False and a set-password token is minted + * creating any other role still requires a password + * the assign-contracts page accepts an external inspector (it 404'd before) + +The regression guard that matters most is +test_external_inspector_scope_is_not_unrestricted: if a future edit reverts a +membership test back to `role == 'inspector'`, an external inspector silently +gains org-wide visibility, which is a cross-customer data leak rather than a +cosmetic bug. +""" + +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 + # get_inspector_scope() imports this model lazily, so the mapper is not + # registered at create_all() time and the table is missing when a + # logged-in inspector hits the dashboard. Import it up front. + 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, **kw): + 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, **kw) + u.set_password('pw-correct1') + db.session.add(u) + db.session.commit() + return u + + +def _seed(): + """Two contracts, one facility each, and the three inspector-ish accounts.""" + from app import db + from app.models.facility import Facility + from app.models.inspection import InspectionTemplate + from app.models.project import Project + + proj_a = Project(name='Contract A', active=True) + proj_b = Project(name='Contract B', active=True) + tmpl = 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_all([proj_a, proj_b, tmpl]) + db.session.commit() + + fac_a = Facility(name='Client A Site', active=True, project_id=proj_a.id) + fac_b = Facility(name='Client B Site', active=True, project_id=proj_b.id) + db.session.add_all([fac_a, fac_b]) + db.session.commit() + + internal = _user('ivy', 'inspector') + external = _user('xan', 'external_inspector') + admin = _user('ada', 'admin') + + return dict(tmpl=tmpl, proj_a=proj_a, proj_b=proj_b, + fac_a=fac_a, fac_b=fac_b, + internal=internal, external=external, admin=admin) + + +def _assign(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 _login(client, user): + return client.post('/auth/login', + data={'username': user.username, 'password': 'pw-correct1'}, + follow_redirects=True) + + +# ── Role predicates ────────────────────────────────────────────────────────── + +def test_is_inspector_covers_both_inspector_roles(client): + env = _seed() + assert env['internal'].is_inspector is True + assert env['external'].is_inspector is True + assert env['admin'].is_inspector is False + + +def test_is_external_inspector_distinguishes_the_two(client): + env = _seed() + assert env['external'].is_external_inspector is True + assert env['internal'].is_external_inspector is False + + +def test_role_label_renders_display_name(client): + env = _seed() + assert env['external'].role_label == 'External Inspector' + assert env['internal'].role_label == 'Inspector' + + +def test_inspector_roles_tuple_contains_both(client): + from app.models.user import User + assert set(User.INSPECTOR_ROLES) == {'inspector', 'external_inspector'} + + +# ── Scoping — the security-critical behaviour ──────────────────────────────── + +def test_external_inspector_scope_is_not_unrestricted(client): + """An external inspector must NEVER get scope None (= see everything). + + This is the regression this phase exists to prevent. If a membership test + is ever reverted to `role == 'inspector'`, get_inspector_scope() returns + None for the external role and every downstream query drops its facility + filter — a cross-customer leak, not a cosmetic bug. + """ + from app.utils.scope import get_inspector_scope + + env = _seed() + _assign(env['external'], env['proj_a']) + + scope = get_inspector_scope(env['external']) + assert scope is not None, 'external inspector fell into the unrestricted branch' + assert scope == [env['fac_a'].id] + assert env['fac_b'].id not in scope + + +def test_external_and_internal_inspectors_scope_identically(client): + from app.utils.scope import get_inspector_scope + + env = _seed() + _assign(env['internal'], env['proj_a']) + _assign(env['external'], env['proj_a']) + + assert (get_inspector_scope(env['external']) + == get_inspector_scope(env['internal']) + == [env['fac_a'].id]) + + +def test_unassigned_external_inspector_sees_nothing(client): + """Strict mode: no assignments means an empty list, not unrestricted.""" + from app.utils.scope import get_inspector_scope + + env = _seed() + assert get_inspector_scope(env['external']) == [] + + +def test_non_inspector_roles_still_unrestricted(client): + from app.utils.scope import get_inspector_scope + + env = _seed() + assert get_inspector_scope(env['admin']) is None + + +# ── Route-level scoping ────────────────────────────────────────────────────── + +def test_facility_list_is_scoped_for_external_inspector(client): + env = _seed() + _assign(env['external'], env['proj_a']) + _login(client, env['external']) + + resp = client.get('/facilities/') + assert resp.status_code == 200 + body = resp.get_data(as_text=True) + assert 'Client A Site' in body + assert 'Client B Site' not in body + + +def test_inspection_and_issue_lists_load_for_external_inspector(client): + """The new role must not 403 or 500 on the core scoped list routes.""" + env = _seed() + _assign(env['external'], env['proj_a']) + _login(client, env['external']) + + for path in ('/inspections/', '/issues/'): + resp = client.get(path) + assert resp.status_code == 200, f'{path} returned {resp.status_code}' + + +# ── Notification matrix ────────────────────────────────────────────────────── + +def test_matrix_exposes_external_inspector_column(client): + from app.models.notification_matrix import MATRIX_ROLES + keys = [k for k, _ in MATRIX_ROLES] + assert 'external_inspector' in keys + + +def test_matrix_defaults_mirror_the_inspector_column(client): + from app.models.notification_matrix import MATRIX_DEFAULTS, MATRIX_EVENTS + + mirrored = 0 + for event in MATRIX_EVENTS: + if (event, 'inspector') in MATRIX_DEFAULTS: + assert (event, 'external_inspector') in MATRIX_DEFAULTS, event + assert (MATRIX_DEFAULTS[(event, 'external_inspector')] + == MATRIX_DEFAULTS[(event, 'inspector')]), event + mirrored += 1 + assert mirrored > 0, 'no inspector defaults found to mirror' + + +# ── Invitation flow ────────────────────────────────────────────────────────── + +def test_creating_external_inspector_invites_instead_of_setting_password(client): + from app.models.user import User + + env = _seed() + _login(client, env['admin']) + + resp = client.post('/auth/users/new', data={ + 'username': 'newxan', + 'full_name': 'New Xan', + 'email': 'newxan@example.com', + 'role': 'external_inspector', + 'password': '', + 'confirm_password': '', + }, follow_redirects=True) + assert resp.status_code == 200 + + created = User.query.filter_by(username='newxan').first() + assert created is not None, 'external inspector was not created' + assert created.role == 'external_inspector' + # Invited, not password-set: login is blocked until they use the link. + assert created.password_set is False + assert created.set_password_token is not None + assert created.set_password_token_expires is not None + + +def test_invited_external_inspector_cannot_log_in_until_setup(client): + from app.models.user import User + + env = _seed() + _login(client, env['admin']) + client.post('/auth/users/new', data={ + 'username': 'newxan', + 'full_name': 'New Xan', + 'email': 'newxan@example.com', + 'role': 'external_inspector', + 'password': '', + 'confirm_password': '', + }, follow_redirects=True) + client.get('/auth/logout', follow_redirects=True) + + created = User.query.filter_by(username='newxan').first() + # The placeholder hash is random, so no password can work; assert the + # account is in the blocked state rather than guessing a credential. + assert created.password_set is False + + +def test_creating_a_normal_role_still_requires_a_password(client): + from app.models.user import User + + env = _seed() + _login(client, env['admin']) + + resp = client.post('/auth/users/new', data={ + 'username': 'nopw', + 'full_name': 'No Password', + 'email': 'nopw@example.com', + 'role': 'inspector', + 'password': '', + 'confirm_password': '', + }, follow_redirects=True) + assert resp.status_code == 200 + assert User.query.filter_by(username='nopw').first() is None + + +# ── Assign-contracts page ──────────────────────────────────────────────────── + +def test_assign_contracts_page_accepts_external_inspector(client): + env = _seed() + _login(client, env['admin']) + + resp = client.get(f"/auth/users/{env['external'].id}/assign-contracts") + assert resp.status_code == 200 + + +def test_assign_contracts_page_still_rejects_non_inspectors(client): + env = _seed() + _login(client, env['admin']) + + resp = client.get(f"/auth/users/{env['admin'].id}/assign-contracts") + assert resp.status_code == 404