diff --git a/app/__init__.py b/app/__init__.py index 29eb66b..5d9109d 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -338,6 +338,12 @@ def create_app(config_name='default'): app.register_blueprint(signup.bp) app.register_blueprint(landing.bp) app.register_blueprint(ui.bp) + + # ── Enrollment form (self-contained — see app/enrollment/__init__.py) ──── + # Registered last and via its own helper so the package stays deletable: + # removing app/enrollment/ and these two lines removes the feature entirely. + from app.enrollment import register_enrollment + register_enrollment(app) # Billing blueprint is CSRF-exempt: /billing/webhook receives raw POST from # Stripe and cannot carry a CSRF token. Subscribe/portal are GET redirects # which Flask-WTF does not protect anyway (CSRF only applies to unsafe methods). diff --git a/app/enrollment/__init__.py b/app/enrollment/__init__.py new file mode 100644 index 0000000..2c8616c --- /dev/null +++ b/app/enrollment/__init__.py @@ -0,0 +1,48 @@ +""" +app/enrollment +-------------- +The JQC Enrollment Form — a self-contained onboarding intake, deliberately +held apart from the rest of the application. + + /enrollment public form emailed to a prospective customer + /enrollment/admin admin-only inbox of submissions + +Separation contract (please keep this true) +------------------------------------------- +1. NO app.models imports, and nothing here writes to the database. Enrollment + happens before any contract, facility or user exists, so there is nothing to + key a row against. Submissions are flat JSON files (see storage.py). + MT NOTE: "no database" does NOT mean "no tenant". Submissions are filed per + tenant on disk, and the admin views only ever list the calling tenant's own + directory — see storage.enrollment_dir(). +2. NO migration, NO model, NO notification-matrix event, NO iPad/API surface. + Deleting this package would remove the two routes and nothing else. +3. Its own template folder (app/enrollment/templates/enrollment/) — enrollment + markup never mixes into app/templates. +4. The only shared code it uses is what it should not reinvent: the app factory, + Flask-WTF CSRF, the rate limiter, and @admin_required. + +If this ever needs to CREATE the accounts it describes, do that as a separate, +explicit admin action that reads a stored submission — do not let the public +form reach into the app's models. +""" + +from .routes import bp # noqa: F401 (re-exported for register_enrollment) + + +def register_enrollment(app): + """Register the blueprint and make sure the storage directory exists.""" + import os + + app.config.setdefault( + 'ENROLLMENT_DIR', + os.path.join(app.instance_path, 'enrollments'), + ) + # Only the ROOT is created at boot. Per-tenant subdirectories are created + # lazily on first use by storage.enrollment_dir(), because the tenant is + # not known until a request is bound. + os.makedirs(app.config['ENROLLMENT_DIR'], exist_ok=True) + app.register_blueprint(bp) + app.logger.info('Enrollment | storage root: %s | per-tenant=%s', + app.config['ENROLLMENT_DIR'], + bool(app.config.get('MULTI_TENANT_ENABLED'))) diff --git a/app/enrollment/mailer.py b/app/enrollment/mailer.py new file mode 100644 index 0000000..92125e6 --- /dev/null +++ b/app/enrollment/mailer.py @@ -0,0 +1,229 @@ +""" +app/enrollment/mailer.py +------------------------ +The enrollment confirmation email. + +Sent to the requester after a submission is stored. One job, and it must never +be able to break that: the record is already safely on disk before this runs, +so every failure path here is logged and swallowed. A bounced confirmation must +not cost the customer their enrollment. + +Sending happens on a background thread (rule 14 — never block the HTTP +response), and the From identity comes from branded_sender() so it stays an +SMTP-authorized address that actually delivers (rules 64 / 76). + +This is the only part of app/enrollment that touches shared mail +infrastructure. It performs exactly ONE database read — resolving the active +admin accounts to notify — and no write. That read is a deliberate, narrowed +exception to the package's no-models rule (rule 88): the alternative, a +hand-maintained recipient list in config, drifts out of step with reality the +first time someone joins or leaves. Everything else here stays model-free. +""" + +import logging +import threading + +from flask import current_app, render_template + +logger = logging.getLogger(__name__) + + +def _tenant_branding(): + """(display_name, support_email) for the calling tenant. + + MT: ST hardcodes its own company name and a personal corrections address in + this module and in schema.py. Sending either to another tenant's customers + would be wrong and confusing, so both are resolved from TenantSettings at + send time, falling back to neutral defaults. + + Best-effort: any failure returns the defaults rather than blocking the + email, matching the rest of this module's never-raise contract. + """ + name, support = 'Janitorial QC', None + try: + from app.models.tenant_settings import TenantSettings + row = TenantSettings.query.first() + if row is not None: + name = row.display_name or name + support = row.support_email or None + except Exception: + logger.debug('ENROLLMENT | tenant branding unavailable, using defaults') + if not support: + support = (current_app.config.get('ENROLLMENT_CORRECTIONS_EMAIL') + or current_app.config.get('MAIL_DEFAULT_SENDER') or '') + return name, support + + +def _text_body(record, people, corrections_email): + """Plain-text alternative — some recipients see only this.""" + lines = [ + f'Hi {record.get("request_by") or "there"},', + '', + 'Thank you — we have received your JQC enrollment form.', + '', + f'Reference: {record.get("id")}', + f'Project: {record.get("project_name")}', + '', + f'People to be set up ({len(people)}):', + ] + for i, person in enumerate(people, start=1): + lines.append( + f' {i}. {person["name"]} — {person["role_label"]} — {person["email"]}' + ) + lines += [ + '', + 'Our team will create these accounts. Each person will receive their own ' + 'email invitation with sign-in instructions.', + '', + f'If anything above is wrong, simply send an email to ' + f'{corrections_email}, and we will correct it.', + '', + _tenant_branding()[0], + ] + return '\n'.join(lines) + + +def _dispatch(msg, label, record): + """Send one message on a background thread. Never raises. + + Rule 14 — the HTTP response must not wait on SMTP. The submission is + already on disk by the time anything here runs, so a mail failure is + logged and dropped rather than surfaced to the customer. + """ + app = current_app._get_current_object() + + def _send(): + with app.app_context(): + try: + from app import mail + mail.send(msg) + logger.info('ENROLLMENT %s SENT | to=%s | id=%s', + label, msg.recipients, record.get('id')) + except Exception as exc: + logger.error('ENROLLMENT %s FAILED | to=%s | id=%s | error=%s', + label, msg.recipients, record.get('id'), exc) + + threading.Thread(target=_send, daemon=True).start() + + +def _admin_recipients(): + """Addresses to alert when a new enrollment arrives. + + Active `admin` accounts, plus any extra addresses in the optional + ENROLLMENT_NOTIFY_EMAILS config (comma-separated) for people who should be + told but do not hold a JQC login. Deduplicated case-insensitively. + + The User import is function-local and read-only — see the module docstring. + """ + emails = [] + try: + from app.models.user import User + rows = User.query.filter(User.role == 'admin', + User.active == True).all() # noqa: E712 + emails += [u.email for u in rows if u.email] + except Exception: + # A DB problem must not stop the confirmation going out, nor the + # submission from succeeding. + logger.exception('ENROLLMENT | could not resolve admin recipients') + + extra = current_app.config.get('ENROLLMENT_NOTIFY_EMAILS') or '' + emails += [e.strip() for e in extra.split(',') if e.strip()] + + seen, out = set(), [] + for e in emails: + low = e.lower() + if low not in seen: + seen.add(low) + out.append(e) + return out + + +def send_admin_notification(record, base_url=None): + """Alert JQC admins that a new enrollment form has arrived. Never raises.""" + if not current_app.config.get('MAIL_SERVER'): + logger.warning('ENROLLMENT ADMIN EMAIL SKIPPED | no MAIL_SERVER | id=%s', + record.get('id')) + return + + try: + from flask_mail import Message + from app.utils.mail_utils import branded_sender + from . import schema + + recipients = _admin_recipients() + if not recipients: + logger.warning('ENROLLMENT | no admin recipients for id=%s', + record.get('id')) + return + + effective_base = (base_url + or current_app.config.get('APP_BASE_URL', '')).rstrip('/') + people = schema.people_of(record) + link = f'{effective_base}/enrollment/admin/{record.get("id")}' + + lines = [ + 'A new JQC enrollment form has been submitted.', + '', + f'Project: {record.get("project_name")}', + f'Requester: {record.get("request_by")} <{record.get("requester_email")}>', + f'Reference: {record.get("id")}', + f'People: {len(people)}', + '', + f'Open it here: {link}', + ] + if record.get('notes'): + lines += ['', f'Customer notes: {record["notes"]}'] + + msg = Message( + subject = f'[JQC] New enrollment — {record.get("project_name")}', + sender = branded_sender(effective_base), + recipients = recipients, + body = '\n'.join(lines), + html = render_template('enrollment/email_admin_notice.html', + record=record, people=people, + schema=schema, link=link), + ) + _dispatch(msg, 'ADMIN EMAIL', record) + + except Exception: + logger.exception('ENROLLMENT ADMIN EMAIL BUILD FAILED | id=%s', + record.get('id')) + + +def send_confirmation(record, base_url=None): + """Email the requester a copy of what they submitted. Never raises.""" + email = (record.get('requester_email') or '').strip() + if not email: + return + + if not current_app.config.get('MAIL_SERVER'): + logger.warning('ENROLLMENT EMAIL SKIPPED | no MAIL_SERVER | id=%s', + record.get('id')) + return + + try: + from flask_mail import Message + from app.utils.mail_utils import branded_sender + from . import schema + + effective_base = (base_url + or current_app.config.get('APP_BASE_URL', '')).rstrip('/') + people = schema.people_of(record) + + msg = Message( + subject = f'[JQC] Enrollment received — {record.get("project_name")}', + sender = branded_sender(effective_base), + recipients = [email], + body = _text_body(record, people, _tenant_branding()[1]), + html = render_template('enrollment/email_confirmation.html', + record=record, people=people, + schema=schema, + corrections_email=_tenant_branding()[1]), + ) + + _dispatch(msg, 'EMAIL', record) + + except Exception: + # Building the message failed (bad template, mail misconfigured, …). + # The submission is already saved — log it and move on. + logger.exception('ENROLLMENT EMAIL BUILD FAILED | id=%s', record.get('id')) diff --git a/app/enrollment/routes.py b/app/enrollment/routes.py new file mode 100644 index 0000000..0f478dc --- /dev/null +++ b/app/enrollment/routes.py @@ -0,0 +1,352 @@ +""" +app/enrollment/routes.py +------------------------ +The JQC Enrollment Form. + + GET /enrollment public form (NO login) + POST /enrollment submit → thank-you page + GET /enrollment/admin admin: all submissions + GET /enrollment/admin/ admin: one submission + POST /enrollment/admin/ admin: office-use fields + status + GET /enrollment/admin/.json admin: raw JSON download + GET /enrollment/admin/export.csv admin: all submissions as CSV + +Separation +---------- +This module imports NOTHING from app.models and writes NOTHING to the database +(see app/enrollment/__init__.py). Its only couplings to the rest of the app are +the ones it cannot sensibly reinvent: the app factory, CSRF, the rate limiter, +and @admin_required for the admin views. + +The public page is login-free, so it follows the same hardening as the `public` +blueprint (rule 74): CSRF-protected form, rate limited, honeypot-guarded, and +its own standalone template with no authenticated nav. +""" + +import csv +import io +import logging +import re +from datetime import datetime + +from flask import (Blueprint, render_template, request, redirect, url_for, + flash, abort, Response, current_app) +from flask_login import login_required + +from app import limiter +from app.utils.decorators import admin_required + +from . import mailer, schema, storage + +logger = logging.getLogger(__name__) + +bp = Blueprint( + 'enrollment', __name__, + url_prefix='/enrollment', + # Own template folder — enrollment templates never mix into app/templates. + template_folder='templates', +) + +#: Bots find public forms fast. A human filling in a 6-person enrollment form +#: does not need more than a few attempts an hour from one address. +_SUBMIT_RATE_LIMIT = '5 per hour' + +_MAX_TEXT = 200 # per free-text field; anything longer is truncated +_MAX_NOTES = 2000 + + +def _clean(value, limit=_MAX_TEXT): + """Trim and length-cap one submitted text field.""" + return (value or '').strip()[:limit] + + +#: Person rows are named person__. The client controls (rows can +#: be added and removed in any order), so the server discovers the indexes that +#: were actually posted rather than trusting a count field. +_PERSON_FIELD_RE = re.compile(r'^person_(\d+)_role$') + + +def _parse_people(form): + """Return the submitted people as an ordered list of dicts. + + Each entry gets a stable `key` (p1, p2, …) assigned by POSITION, not by the + client's index — so the matrix keys in a stored submission are always dense + and predictable no matter which rows the customer deleted before sending. + """ + indexes = sorted( + int(m.group(1)) + for m in (_PERSON_FIELD_RE.match(k) for k in form.keys()) if m + ) + + people = [] + for idx in indexes: + role = form.get(f'person_{idx}_role', '') + if role not in schema.ROLE_KEYS: + role = schema.DEFAULT_FIRST_ROLE + name = _clean(form.get(f'person_{idx}_name')) + job_title = _clean(form.get(f'person_{idx}_job_title')) + email = _clean(form.get(f'person_{idx}_email')) + # Drop rows the customer added but left completely blank. + if not (name or job_title or email): + continue + people.append({ + 'key': f'p{len(people) + 1}', + 'form_index': idx, # so the matrix cells can be read back + 'role': role, + 'name': name, + 'job_title': job_title, + 'email': email, + }) + if len(people) >= schema.MAX_PEOPLE: + logger.warning('ENROLLMENT | people capped at %d', schema.MAX_PEOPLE) + break + return people + + +def _seed_people(people, matrix, mobile_app): + """Shape the submitted people for the page to re-render after an error. + + Folds each person's ticked tasks into their own row, so the browser can + rebuild the table from scratch with fresh row indexes and still restore + every answer. + """ + seed = [] + for person in people: + seed.append({ + 'role': person['role'], + 'name': person['name'], + 'job_title': person['job_title'], + 'email': person['email'], + 'tasks': [ref for ref, _l, _s in schema.TASKS + if matrix.get(str(ref), {}).get(person['key'])], + 'mobile': bool(mobile_app.get(person['key'])), + }) + return seed + + +# ── Public form ────────────────────────────────────────────────────────────── + +@bp.route('', methods=['GET']) +@bp.route('/', methods=['GET']) +def form(): + """Render the blank enrollment form. No login — the link is emailed out.""" + return render_template('enrollment/form.html', schema=schema, + seed_people=[]) + + +@bp.route('', methods=['POST']) +@bp.route('/', methods=['POST']) +@limiter.limit(_SUBMIT_RATE_LIMIT) +def submit(): + """Parse, validate and store one enrollment submission.""" + # Honeypot: a field hidden from humans via CSS. Anything that fills it in + # is a bot. Answer 200 as though accepted so it learns nothing. + if (request.form.get('website') or '').strip(): + logger.info('ENROLLMENT | honeypot tripped | ip=%s', request.remote_addr) + return render_template('enrollment/submitted.html', reference=None) + + project_name = _clean(request.form.get('project_name')) + request_by = _clean(request.form.get('request_by')) + requester_email = _clean(request.form.get('requester_email')) + + people = _parse_people(request.form) + + # ── Step 2 matrix + Step 3 mobile, keyed by person ─────────────────── + # Only cells the person's role actually offers are read, so a crafted POST + # cannot record an admin-only task against an inspector. + matrix = {} + for ref, _label, scope in schema.TASKS: + row = {} + for person in people: + if schema.task_applies(scope, person['role']): + row[person['key']] = bool( + request.form.get(f'task_{ref}_person_{person["form_index"]}')) + matrix[str(ref)] = row + + mobile_app = { + p['key']: bool(request.form.get(f'mobile_person_{p["form_index"]}')) + for p in people + } + + # ── Validation ─────────────────────────────────────────────────────── + # A person counts only with BOTH a name and an email — a half-filled row + # cannot be set up, so it must not pass as one. + named = [p for p in people if p['name'] and p['email']] + errors = [] + if not project_name: + errors.append('Project Name is required.') + if not request_by: + errors.append('Request by is required.') + if not requester_email: + errors.append('Requester email is required — we send your confirmation ' + 'there.') + elif '@' not in requester_email: + errors.append('The requester email address does not look valid.') + if not named: + errors.append('Please add at least one person with both a name and an ' + 'email address.') + for p in people: + if p['email'] and '@' not in p['email']: + errors.append(f'"{p["name"] or p["key"]}" has an email address that ' + f'does not look valid.') + seen = set() + for p in named: + low = p['email'].lower() + if low in seen: + errors.append(f'{p["email"]} is listed more than once — each person ' + f'needs their own email address.') + seen.add(low) + + prior = { + 'project_name': project_name, + 'request_by': request_by, + 'requester_email': requester_email, + 'date_requested': _clean(request.form.get('date_requested')), + 'notes': _clean(request.form.get('notes'), _MAX_NOTES), + 'people': people, + 'matrix': matrix, + 'mobile_app': mobile_app, + } + + if errors: + for e in errors: + flash(e, 'danger') + # Re-render with what they typed so nothing is retyped. + return render_template( + 'enrollment/form.html', schema=schema, submitted=prior, + seed_people=_seed_people(people, matrix, mobile_app)), 400 + + now = datetime.now() + record = dict(prior) + record.update({ + 'id': storage.new_id(now), + 'submitted_at': now.isoformat(timespec='seconds'), + # Filled in later by staff on the admin page. + 'office': {k: '' for k, _ in schema.OFFICE_FIELDS}, + 'status': 'new', + 'meta': { + 'ip': request.remote_addr, + 'user_agent': (request.headers.get('User-Agent') or '')[:300], + }, + }) + + try: + storage.save(record) + except Exception: + logger.exception('ENROLLMENT | save failed | project=%r', project_name) + flash('Sorry — we could not save your form. Please try again, or ' + 'email us directly.', 'danger') + return render_template( + 'enrollment/form.html', schema=schema, submitted=prior, + seed_people=_seed_people(people, matrix, mobile_app)), 500 + + logger.info('ENROLLMENT | submitted | id=%s project=%r people=%d ip=%s', + record['id'], project_name, len(named), request.remote_addr) + + # Both emails fire AFTER the save and are fully guarded — a mail problem + # must never cost the customer their submission. + mailer.send_confirmation(record, base_url=request.host_url) # requester + mailer.send_admin_notification(record, base_url=request.host_url) # JQC admins + + return render_template('enrollment/submitted.html', reference=record['id'], + email=requester_email) + + +# ── Admin ──────────────────────────────────────────────────────────────────── + +@bp.route('/admin') +@login_required +@admin_required +def admin_list(): + records = storage.load_all() + logger.info('ENROLLMENT | admin_list | count=%d', len(records)) + return render_template('enrollment/admin_list.html', + records=records, schema=schema) + + +@bp.route('/admin/', methods=['GET', 'POST']) +@login_required +@admin_required +def admin_detail(submission_id): + record = storage.load(submission_id) + if record is None: + abort(404) + + if request.method == 'POST': + office = {k: _clean(request.form.get(k)) for k, _ in schema.OFFICE_FIELDS} + status = request.form.get('status', 'new') + if status not in schema.STATUSES: + status = record.get('status', 'new') + record = storage.update_office(submission_id, office, status) + if record is None: + abort(404) + flash('Enrollment record updated.', 'success') + return redirect(url_for('enrollment.admin_detail', + submission_id=submission_id)) + + return render_template('enrollment/admin_detail.html', + record=record, schema=schema) + + +@bp.route('/admin/.json') +@login_required +@admin_required +def admin_download(submission_id): + import json + record = storage.load(submission_id) + if record is None: + abort(404) + return Response( + json.dumps(record, indent=2, ensure_ascii=False), + mimetype='application/json', + headers={'Content-Disposition': + f'attachment; filename=enrollment-{submission_id}.json'}, + ) + + +@bp.route('/admin/export.csv') +@login_required +@admin_required +def admin_export_csv(): + """One row per PERSON (not per submission) — that is the unit of work when + actually setting the accounts up. Reads through schema.people_of(), so + submissions stored in the older fixed-seat format export identically.""" + records = storage.load_all() + + buf = io.StringIO() + w = csv.writer(buf) + task_headers = [f'{ref}. {label}' for ref, label, _ in schema.TASKS] + w.writerow(['Submission ID', 'Submitted At', 'Status', 'Project Name', + 'Requested By', 'Date Requested', 'Role', 'Name', 'Job Title', + 'Email', 'Mobile App'] + task_headers) + + for rec in records: + for person in schema.people_of(rec): + row = [ + rec.get('id', ''), + rec.get('submitted_at', ''), + schema.STATUS_LABELS.get(rec.get('status'), rec.get('status', '')), + rec.get('project_name', ''), + rec.get('request_by', ''), + rec.get('date_requested', ''), + person['role_label'], + person['name'], + person['job_title'], + person['email'], + 'Yes' if schema.wants_mobile(rec, person['key']) else '', + ] + for ref, _label, scope in schema.TASKS: + if not schema.task_applies(scope, person['role']): + row.append('n/a') + else: + row.append('Yes' if schema.cell(rec, ref, person['key']) else '') + w.writerow(row) + + logger.info('ENROLLMENT | csv export | submissions=%d', len(records)) + stamp = datetime.now().strftime('%Y%m%d') + return Response( + buf.getvalue(), + mimetype='text/csv', + headers={'Content-Disposition': + f'attachment; filename=jqc-enrollments-{stamp}.csv'}, + ) diff --git a/app/enrollment/schema.py b/app/enrollment/schema.py new file mode 100644 index 0000000..0f949e0 --- /dev/null +++ b/app/enrollment/schema.py @@ -0,0 +1,218 @@ +""" +app/enrollment/schema.py +------------------------ +The JQC Enrollment Form, expressed as data. + +This is the SINGLE source of truth for the form's shape. The public template +renders from it, the POST handler parses against it, and the admin detail view +re-renders a stored submission through it. Changing a task label or adding a +role is a one-line edit here — no template or parser change. + +Deliberately free of any app model / DB import: the enrollment form describes +what a prospective customer *wants set up*, not anything that exists in the +system yet. Keep it that way (see app/enrollment/__init__.py). The ROLES below +happen to mirror the app's user roles, but they are a COPY on purpose — the +public form must not import the User model. +""" + +# ── Roles a person can be enrolled as ──────────────────────────────────────── +# key -> label, shown in the Step 1 role dropdown. +ROLES = [ + ('admin', 'Admin'), + ('director', 'Director'), + ('auditor', 'Auditor'), + ('inspector', 'Inspector'), + ('external_inspector', 'External Inspector'), +] + +ROLE_LABELS = dict(ROLES) +ROLE_KEYS = [k for k, _ in ROLES] + +#: Roles that act on the administrative side of the printed form (the +#: "Admin / Director" column). Everything else is an inspector seat. Drives +#: both the recommendation preset and eligibility for admin-only tasks. +ADMIN_ROLES = {'admin', 'director', 'auditor'} + +#: Role pre-selected for the first row — the form starts with one +#: administrative contact, as on the printed sheet. +DEFAULT_FIRST_ROLE = 'admin' + +#: Upper bound on people per submission. Generous for a real enrollment, but +#: bounded so a scripted POST cannot make us build an unbounded matrix. +MAX_PEOPLE = 25 + + +def is_admin_role(role): + return role in ADMIN_ROLES + + +# ── Task rows ──────────────────────────────────────────────────────────────── +# ref, label, scope. scope 'admin_only' means the cell is offered only to +# people in an ADMIN_ROLES role (ref 10 on the printed form). +TASKS = [ + (1, 'Receive new inspection submitted notification', 'all'), + (2, 'Receive issue-related notification', 'all'), + (3, 'New issue created', 'all'), + (4, 'Issue status updated', 'all'), + (5, 'Issue comment added', 'all'), + (6, 'Request follow up / re-inspection', 'all'), + (7, 'Add Comments (issue detail page)', 'all'), + (8, 'Receive issue SLA (at-risk, breached)', 'all'), + (9, 'Log new issue', 'all'), + (10, 'Search/Export Reports (inspection/issue)', 'admin_only'), +] + +TASK_LABELS = {ref: label for ref, label, _ in TASKS} + + +def task_applies(scope, role): + """True when a task row offers a checkbox to someone in `role`.""" + return scope == 'all' or is_admin_role(role) + + +# ── Step 3 ─────────────────────────────────────────────────────────────────── +MOBILE_APP_LABEL = 'JQC Mobile App For Smart Device' + + +# ── Recommended defaults ───────────────────────────────────────────────────── +# ref -> (recommended for admin-side roles, recommended for inspector roles). +# None = the row offers that side no cell. +# +# The printed form showed this as a separate RECOMMENDATION table for the +# customer to copy by hand. It is now applied by the "Recommendation selection" +# button instead, so the table is no longer rendered — but this mapping is +# still the authority, and is handed to the page as JSON. +RECOMMENDATION = { + 1: (False, True), + 2: (False, True), + 3: (True, True), + 4: (False, True), + 5: (True, True), + 6: (True, True), + 7: (True, True), + 8: (False, True), + 9: (True, True), + 10: (True, None), +} + + +def recommendation_for(role): + """Return {task_ref: bool} — the recommended preset for one role. + + Rows that offer this role no cell are omitted rather than set False, so + the caller never ticks a checkbox that does not exist. + """ + admin_side = is_admin_role(role) + preset = {} + for ref, _label, scope in TASKS: + if not task_applies(scope, role): + continue + rec = RECOMMENDATION.get(ref, (False, False)) + value = rec[0] if admin_side else rec[1] + if value is None: + continue + preset[ref] = bool(value) + return preset + + +def recommendation_map(): + """{role_key: {task_ref: bool}} for every role — serialised to the page.""" + return {role: recommendation_for(role) for role in ROLE_KEYS} + + +#: Where a customer should write if their submission needs correcting. The +#: confirmation email is sent FROM the unmonitored no-reply identity +#: (branded_sender), so "reply to this email" would go nowhere — point them +#: here instead. Used by both the text and HTML bodies of the confirmation. +# MT: retained only as a last-resort default. The address actually shown to a +# customer is resolved per tenant at send time from TenantSettings.support_email +# — see mailer._tenant_branding(). Never send one tenant's customers another +# tenant's (or a developer's personal) address. +CORRECTIONS_EMAIL = '' + + +NOTES = [ + 'Each user will receive instructions on how to sign up and install the app ' + 'on their smart device.', + 'Along with the installation instructions, users will receive a quick guide ' + 'to navigate the web portal and app based on their credentials.', +] + + +# ── Office-use fields ──────────────────────────────────────────────────────── +# Filled in by the tenant AFTER receipt, on the admin detail page only. The +# printed sheet showed these to the customer as a blank "for office use" block; +# the web form does not render them at all — a customer cannot fill them in, so +# showing them was only noise. +OFFICE_FIELDS = [ + ('receive_date', 'Receive Date'), + ('program_by', 'Program By'), + ('date_email_invitation', 'Date email invitation'), +] + +STATUSES = ['new', 'in_progress', 'completed'] + +STATUS_LABELS = { + 'new': 'New', + 'in_progress': 'In Progress', + 'completed': 'Completed', +} + + +# ── Legacy record support ──────────────────────────────────────────────────── +# Submissions taken before the form moved to free-form people used six fixed +# seats. Stored files are never rewritten, so the admin views normalise on +# read instead — one shape to render, whichever format is on disk. +_LEGACY_SEAT_ROLES = { + 'admin': 'admin', + 'inspector_1': 'inspector', + 'inspector_2': 'inspector', + 'inspector_3': 'inspector', + 'inspector_4': 'inspector', + 'inspector_5': 'inspector', +} + + +def people_of(record): + """Return a submission's people as a uniform list, old format or new. + + Each entry: {key, role, role_label, name, job_title, email}. + """ + if record.get('people'): + out = [] + for p in record['people']: + role = p.get('role', 'inspector') + out.append({ + 'key': p.get('key', ''), + 'role': role, + 'role_label': ROLE_LABELS.get(role, role.replace('_', ' ').title()), + 'name': p.get('name', ''), + 'job_title': p.get('job_title', ''), + 'email': p.get('email', ''), + }) + return out + + # Legacy: fixed seats under 'registrants'. + out = [] + for reg in record.get('registrants', []): + if not (reg.get('name') or reg.get('email')): + continue + role = _LEGACY_SEAT_ROLES.get(reg.get('key'), 'inspector') + out.append({ + 'key': reg.get('key', ''), + 'role': role, + 'role_label': ROLE_LABELS.get(role, role.title()), + 'name': reg.get('name', ''), + 'job_title': reg.get('job_title', ''), + 'email': reg.get('email', ''), + }) + return out + + +def cell(record, ref, person_key): + """True when `person_key` was ticked for task `ref` in this submission.""" + return bool(record.get('matrix', {}).get(str(ref), {}).get(person_key)) + + +def wants_mobile(record, person_key): + return bool(record.get('mobile_app', {}).get(person_key)) diff --git a/app/enrollment/storage.py b/app/enrollment/storage.py new file mode 100644 index 0000000..4b9b81c --- /dev/null +++ b/app/enrollment/storage.py @@ -0,0 +1,194 @@ +""" +app/enrollment/storage.py +------------------------- +Flat-file persistence for enrollment submissions — one JSON document per +submission, under the directory named by config ENROLLMENT_DIR, in a +per-tenant subdirectory (see enrollment_dir()). + +Why files and not a table +------------------------- +Enrollment happens BEFORE anything exists in the system: there is no contract, +no facility and no user account to key a row against, and the volume is a +handful of documents a year. A directory of readable JSON keeps this feature +completely outside the schema — no model, no migration, nothing to keep in sync +with the rest of the app. It can be backed up with `cp` and read with `cat`. + +File naming +----------- + -<8 hex>.json + +Time-ordered so a plain directory listing sorts chronologically, with random +suffix so two submissions in the same second cannot collide. The stem is the +submission's id and is the ONLY thing the admin URLs accept — see _safe_id(). +""" + +import json +import logging +import os +import re +import secrets +import tempfile +from datetime import datetime + +from flask import current_app + +logger = logging.getLogger(__name__) + +#: Submission ids are generated by us and must round-trip through a URL and a +#: file path. Anything not matching is rejected before touching the filesystem, +#: so a crafted id can never escape the enrollment directory (path traversal). +_ID_RE = re.compile(r'^\d{8}-\d{6}-[0-9a-f]{8}$') + + +class TenantUnresolved(RuntimeError): + """Raised when multi-tenancy is on but no tenant is bound to the request. + + Deliberately fatal rather than falling back to the shared root directory: + a fallback would put one tenant's submissions where every other tenant's + admin can read them. + """ + + +def enrollment_dir(): + """Absolute path of THIS TENANT's submission directory, created on first use. + + Multi-tenant isolation (MT-17) + ------------------------------ + ST keeps every submission in one flat directory. In MT that directory is + shared by every tenant on the host, so /enrollment/admin would list other + organisations' submissions — names, emails and phone numbers of people at + another company. Submissions are therefore filed under a per-tenant + subdirectory: + + /t/.json + + ``t`` mirrors ``storage.tenant_key_prefix()`` so the on-disk layout is + the same shape as the media object keys. + + When MULTI_TENANT_ENABLED is false the root directory is used unchanged, + so a single-tenant deploy behaves exactly like ST. + + When multi-tenancy IS enabled but no tenant is bound, this raises rather + than falling back to the root — see TenantUnresolved. + """ + root = current_app.config['ENROLLMENT_DIR'] + + if not current_app.config.get('MULTI_TENANT_ENABLED'): + os.makedirs(root, exist_ok=True) + return root + + from flask import g + tenant = getattr(g, 'tenant', None) + if tenant is None: + logger.error('ENROLLMENT | no tenant bound — refusing to touch storage') + raise TenantUnresolved( + 'enrollment storage requires a resolved tenant when ' + 'MULTI_TENANT_ENABLED is set' + ) + + path = os.path.join(root, f't{tenant.id}') + os.makedirs(path, exist_ok=True) + return path + + +def new_id(when=None): + """Mint a time-ordered, collision-safe submission id.""" + when = when or datetime.now() + return f'{when:%Y%m%d-%H%M%S}-{secrets.token_hex(4)}' + + +def _safe_id(submission_id): + """Return the id if it is one of ours, else None. + + Never interpolate an unvalidated id into a path — `../../etc/passwd` and + friends. Callers should 404 on None. + """ + if not submission_id or not _ID_RE.match(submission_id): + logger.warning('ENROLLMENT | rejected malformed id=%r', submission_id) + return None + return submission_id + + +def _path_for(submission_id): + sid = _safe_id(submission_id) + if sid is None: + return None + return os.path.join(enrollment_dir(), f'{sid}.json') + + +def save(record): + """Write a submission atomically. Returns the id. + + Written to a temp file in the same directory then os.replace()d, so a + crash mid-write can never leave a truncated JSON document that would break + the admin list for every other submission. + """ + sid = record['id'] + path = _path_for(sid) + if path is None: + raise ValueError(f'refusing to save malformed id {sid!r}') + + directory = os.path.dirname(path) + fd, tmp = tempfile.mkstemp(dir=directory, suffix='.tmp') + try: + with os.fdopen(fd, 'w', encoding='utf-8') as fh: + json.dump(record, fh, indent=2, ensure_ascii=False) + os.replace(tmp, path) + except Exception: + # Never leave the temp file behind on a failed write. + try: + os.unlink(tmp) + except OSError: + pass + raise + + logger.info('ENROLLMENT | saved | id=%s project=%r', + sid, record.get('project_name')) + return sid + + +def load(submission_id): + """Return one submission dict, or None if unknown/unreadable.""" + path = _path_for(submission_id) + if path is None or not os.path.isfile(path): + return None + try: + with open(path, encoding='utf-8') as fh: + return json.load(fh) + except (OSError, ValueError): + logger.exception('ENROLLMENT | unreadable submission id=%s', submission_id) + return None + + +def load_all(): + """Return every submission, newest first. + + A single corrupt file is skipped with a log line rather than breaking the + whole admin list. + """ + directory = enrollment_dir() + records = [] + for name in sorted(os.listdir(directory), reverse=True): + if not name.endswith('.json'): + continue + rec = load(name[:-len('.json')]) + if rec is not None: + records.append(rec) + return records + + +def update_office(submission_id, office, status): + """Merge the office-use block + status into a stored submission. + + Returns the updated record, or None if the id is unknown. Only these + fields are writable after submission — the customer's own answers are + immutable, so the file stays an accurate record of what they asked for. + """ + rec = load(submission_id) + if rec is None: + return None + rec.setdefault('office', {}).update(office) + rec['status'] = status + rec['updated_at'] = datetime.now().isoformat(timespec='seconds') + save(rec) + return rec diff --git a/app/enrollment/templates/enrollment/admin_detail.html b/app/enrollment/templates/enrollment/admin_detail.html new file mode 100644 index 0000000..2de575c --- /dev/null +++ b/app/enrollment/templates/enrollment/admin_detail.html @@ -0,0 +1,181 @@ +{% extends "base.html" %} +{% block title %}Enrollment — {{ record.project_name }}{% endblock %} + +{# One submitted enrollment form, rendered through the same schema the public + page uses. The customer's answers are READ-ONLY here — only the office-use + block and the status are editable, so the file stays a faithful record of + what was actually requested. #} + +{% block content %} +
+
+

{{ record.project_name }}

+
+ Reference {{ record.id }} · submitted {{ record.submitted_at | replace('T', ' ') }} + {% if record.updated_at %} + · updated {{ record.updated_at | replace('T', ' ') }} + {% endif %} +
+
+
+ + Back + + + Download JSON + +
+
+ +
+ {# ── Request details ─────────────────────────────────────────────── #} +
+
+
Request
+
+
+
Project Name
{{ record.project_name or '—' }}
+
Request by
{{ record.request_by or '—' }}
+
Requester email
+
+ {% if record.requester_email %} + {{ record.requester_email }} + {% else %}—{% endif %} +
+
Date Requested
{{ record.date_requested or '—' }}
+
+ {% if record.notes %} +
+
Customer notes
+
{{ record.notes }}
+ {% endif %} +
+
+
+ + {# ── Office use — the only editable part ─────────────────────────── #} +
+
+
For Office Use
+
+
+ + {% for key, label in schema.OFFICE_FIELDS %} +
+ + +
+ {% endfor %} +
+ + +
+ +
+
+
+
+
+ +{# ── The people to set up ───────────────────────────────────────────── #} +{% set people = schema.people_of(record) %} +
+
+ Users to Register + {{ people | length }} +
+
+
+ + + + + + + + + + {% for person in people %} + + + + + + + + + {% endfor %} + +
No.RoleNameJob TitleEmailMobile App
{{ loop.index }}{{ person.role_label }}{{ person.name or '—' }}{{ person.job_title or '—' }} + {% if person.email %} + {{ person.email }} + {% else %}—{% endif %} + + {% if schema.wants_mobile(record, person.key) %} + + {% else %}—{% endif %} +
+
+
+
+ +{# ── The requested task matrix — one column per person ──────────────── #} +
+
Requested Tasks & Functions
+
+
+ + + + + + {% for person in people %} + + {% endfor %} + + + + {% for ref, label, scope in schema.TASKS %} + + + + {% for person in people %} + + {% endfor %} + + {% endfor %} + +
RefTask / Function + {{ person.name or 'Person ' ~ loop.index }} +
+ {{ person.role_label }} +
+
{{ ref }}{{ label }} + {% if not schema.task_applies(scope, person.role) %} + · + {% elif schema.cell(record, ref, person.key) %} + + {% else %} + — + {% endif %} +
+
+
+
+ +{% if record.meta %} +
+ Submitted from {{ record.meta.ip or 'unknown address' }} +
+{% endif %} +{% endblock %} diff --git a/app/enrollment/templates/enrollment/admin_list.html b/app/enrollment/templates/enrollment/admin_list.html new file mode 100644 index 0000000..c3641e3 --- /dev/null +++ b/app/enrollment/templates/enrollment/admin_list.html @@ -0,0 +1,98 @@ +{% extends "base.html" %} +{% block title %}Enrollment Forms{% endblock %} + +{# Admin inbox of enrollment submissions. Extends base.html so it picks up + whichever design (classic / modern) the admin has selected. #} + +{% block content %} +
+

Enrollment Forms

+
+ + View public form + + {% if records %} + + Export CSV + + {% endif %} +
+
+ +
+ +
+ Send customers this link to enroll: + {{ url_for('enrollment.form', _external=True) }}
+ + Submissions are stored as JSON files on the server, outside the database — + one file per form. + +
+
+ +{% if records %} +
+
+
+ + + + + + + + + + + + + + {% for r in records %} + {# people_of() normalises both the current and the legacy stored shape #} + {% set named = schema.people_of(r) %} + {% set mobile_count = r.mobile_app.values() | select | list | length %} + + + + + + + + + + {% endfor %} + +
SubmittedProjectRequested ByUsersMobile AppStatus
+ {{ r.submitted_at | replace('T', ' ') }} + {{ r.project_name or '—' }}{{ r.request_by or '—' }} + {{ named | length }} + + {% if mobile_count %} + {{ mobile_count }} + {% else %}—{% endif %} + + + {{ schema.STATUS_LABELS.get(r.status, r.status) }} + + + + Open + +
+
+
+
+{% else %} +
+
+ + No enrollment forms have been submitted yet. +
+
+{% endif %} +{% endblock %} diff --git a/app/enrollment/templates/enrollment/email_admin_notice.html b/app/enrollment/templates/enrollment/email_admin_notice.html new file mode 100644 index 0000000..8d91da8 --- /dev/null +++ b/app/enrollment/templates/enrollment/email_admin_notice.html @@ -0,0 +1,89 @@ +{# Internal alert to JQC admins when a new enrollment form arrives. Inline + styles only and no external assets — mail clients strip + + +
+ +

JQC Enrollment Form

+
{{ tenant_branding.display_name if tenant_branding else 'Janitorial QC' }}
+ + {% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %} + {% for category, message in messages %} +
{{ message }}
+ {% endfor %} + {% endif %} + {% endwith %} + + + +
+ + + {# Honeypot — must stay empty. #} + + + {# ── Header ─────────────────────────────────────────────────────── #} + {# The printed sheet carried a blank "for office use" block here. It is not + rendered on the web form — a customer cannot fill it in. Those fields + still exist and are filled by staff on the admin detail page. #} +
+
+ + + + + + + + + + + + + + + + + +
Project Name
Request by:
Requester email:
Date Requested:
+
+ We send your confirmation, with a copy of everything below, to the + requester email. +
+
+
+ + {# ── Step 1 — the people ────────────────────────────────────────── #} +
+ Step 1: Please list everyone who needs access. Each person will + receive an email invitation at the address you provide. +
+ +
+ + + + + + + + + + + + +
No.RoleFirst and last nameJob TitleEmail Address
+
+ +
+ + +
+ + {# ── Step 2 — the task matrix, built from Step 1 ────────────────── #} +
+ Step 2: Please check the task/function for each user, or apply our + recommended selection and adjust it. +
+ +
+ + +
+ Our recommendation keeps administrators and directors from receiving an + overwhelming number of email notifications. You can change any box afterwards. +
+
+ +
+ + {# ── Step 3 — mobile app ────────────────────────────────────────── #} +
+ Step 3: Please check the box next to the user who will receive the + app for smart devices. +
+ +
+ + {# ── Notes ──────────────────────────────────────────────────────── #} +
Anything else we should know? (optional)
+ + +
Note:
+
    + {% for note in schema.NOTES %}
  1. {{ note }}
    2. {% endfor %} +
+ +
+ + +
+
+
+ + + + + diff --git a/app/enrollment/templates/enrollment/submitted.html b/app/enrollment/templates/enrollment/submitted.html new file mode 100644 index 0000000..74d9a3f --- /dev/null +++ b/app/enrollment/templates/enrollment/submitted.html @@ -0,0 +1,46 @@ + + + + + + +Enrollment received — JQC + + + + + +
+
+
+ +

Thank you — your enrollment form has been received.

+

+ Our team will set up the accounts you listed. Each user will receive an + email invitation with sign-in instructions, and a quick guide for the + web portal and the mobile app. +

+ {% if email %} +

+ + A confirmation has been sent to {{ email }}. +

+ {% endif %} + {% if reference %} +
+
Your reference number
+
{{ reference }}
+
+

+ Please quote this reference if you contact us about your enrollment. +

+ {% endif %} +
+
+
{{ tenant_branding.display_name if tenant_branding else 'Janitorial QC' }}
+
+ + diff --git a/tests/test_enrollment.py b/tests/test_enrollment.py new file mode 100644 index 0000000..a1d34d9 --- /dev/null +++ b/tests/test_enrollment.py @@ -0,0 +1,244 @@ +""" +tests/test_enrollment.py +------------------------ +Behaviour tests for MT-17 — the enrollment intake form. + +Runs on the in-memory SQLite app fixture (multi-tenancy inert). Covers: + + * the public form renders without a login and carries no authenticated nav + * a submission is stored, is readable back, and the customer's own answers + are immutable afterwards (only the office block and status are writable) + * malformed submission ids are rejected before touching the filesystem + (path traversal) + * the admin views require a login + * TENANT ISOLATION: with multi-tenancy enabled, submissions are filed under + a per-tenant directory, one tenant's admin list never contains another's, + and an unresolved tenant raises rather than falling back to the shared root + +The isolation tests are the reason this module exists. ST keeps every +submission in one flat directory, which in MT would put the names, emails and +phone numbers of one organisation's staff in front of another organisation's +admin. `test_unresolved_tenant_refuses_rather_than_sharing` pins the fail-closed +behaviour specifically, because a fallback to the root directory would be a +silent cross-tenant leak rather than a visible error. +""" + +import os + +import pytest + + +@pytest.fixture +def client(app, tmp_path): + """Fresh schema + test client, with enrollment storage in a temp dir.""" + with app.app_context(): + from app import db + from app.models import inspector_assignment # noqa: F401 + db.drop_all() + db.create_all() + app.config['ENROLLMENT_DIR'] = str(tmp_path / 'enrollments') + os.makedirs(app.config['ENROLLMENT_DIR'], exist_ok=True) + yield app.test_client() + db.session.remove() + + +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 _login(client, user): + resp = client.post('/auth/login', + data={'username': user.username, 'password': 'pw-correct1'}, + follow_redirects=True) + assert 'Login - ' not in resp.get_data(as_text=True), 'login failed' + return resp + + +def _record(app, project_name='Acme Tower'): + """Build and save one submission through the real storage layer. + + Mirrors the payload routes.submit() actually writes — the admin templates + read every one of these keys, so a minimal stub renders as UndefinedError + rather than exercising the page. + """ + from datetime import datetime + from app.enrollment import storage, schema + + now = datetime.now() + rec = { + 'project_name': project_name, + 'request_by': 'Reception', + 'requester_email': 'reception@example.com', + 'date_requested': now.strftime('%Y-%m-%d'), + 'notes': '', + 'people': [], + 'matrix': {}, + 'mobile_app': {}, + 'id': storage.new_id(now), + 'submitted_at': now.isoformat(timespec='seconds'), + 'office': {k: '' for k, _ in schema.OFFICE_FIELDS}, + 'status': 'new', + 'meta': {'ip': '127.0.0.1', 'user_agent': 'pytest'}, + } + storage.save(rec) + return rec + + +# ── Public form ────────────────────────────────────────────────────────────── + +def test_public_form_renders_without_login(client): + resp = client.get('/enrollment') + assert resp.status_code == 200 + body = resp.get_data(as_text=True) + assert 'Login - ' not in body + # Login-free page: it must not carry the authenticated portal nav. + assert 'jqc-sidebar' not in body + + +def test_admin_list_requires_login(client): + resp = client.get('/enrollment/admin') + assert resp.status_code == 302 + assert '/auth/login' in resp.headers.get('Location', '') + + +def test_admin_list_renders_for_admin(client, app): + admin = _user('ada', 'admin') + _login(client, admin) + with app.test_request_context(): + _record(app, 'Listed Site') + resp = client.get('/enrollment/admin') + assert resp.status_code == 200 + assert 'Listed Site' in resp.get_data(as_text=True) + + +# ── Storage round-trip ─────────────────────────────────────────────────────── + +def test_submission_round_trips(client, app): + from app.enrollment import storage + with app.test_request_context(): + rec = _record(app) + back = storage.load(rec['id']) + assert back is not None + assert back['project_name'] == 'Acme Tower' + + +def test_customer_answers_are_immutable_after_submission(client, app): + """Only the office block and status are writable — the file must stay an + accurate record of what the customer actually asked for.""" + from app.enrollment import storage + with app.test_request_context(): + rec = _record(app, project_name='Original Name') + storage.update_office(rec['id'], {'assigned_to': 'ops'}, 'in_review') + back = storage.load(rec['id']) + + assert back['project_name'] == 'Original Name' # untouched + assert back['office']['assigned_to'] == 'ops' + assert back['status'] == 'in_review' + + +@pytest.mark.parametrize('bad_id', [ + '../../etc/passwd', + '..%2f..%2fetc', + 'not-an-id', + '', + '20260101-120000-ZZZZZZZZ', # non-hex suffix +]) +def test_malformed_ids_are_rejected(client, app, bad_id): + """A crafted id must never be interpolated into a filesystem path.""" + from app.enrollment import storage + with app.test_request_context(): + assert storage.load(bad_id) is None + + +# ── Tenant isolation ───────────────────────────────────────────────────────── + +class _FakeTenant: + def __init__(self, tid): + self.id = tid + + +def test_single_tenant_mode_uses_the_root_directory(client, app): + """With MT off, behaviour matches ST exactly — no per-tenant subdirectory.""" + from app.enrollment import storage + with app.test_request_context(): + assert app.config.get('MULTI_TENANT_ENABLED') is False + assert storage.enrollment_dir() == app.config['ENROLLMENT_DIR'] + + +def test_submissions_are_filed_per_tenant(client, app): + from flask import g + from app.enrollment import storage + + root = app.config['ENROLLMENT_DIR'] + app.config['MULTI_TENANT_ENABLED'] = True + try: + with app.test_request_context(): + g.tenant = _FakeTenant(7) + assert storage.enrollment_dir() == os.path.join(root, 't7') + _record(app, 'Tenant Seven Site') + with app.test_request_context(): + g.tenant = _FakeTenant(9) + assert storage.enrollment_dir() == os.path.join(root, 't9') + finally: + app.config['MULTI_TENANT_ENABLED'] = False + + assert os.path.isdir(os.path.join(root, 't7')) + + +def test_one_tenant_never_sees_anothers_submissions(client, app): + """The leak this phase exists to prevent.""" + from flask import g + from app.enrollment import storage + + app.config['MULTI_TENANT_ENABLED'] = True + try: + with app.test_request_context(): + g.tenant = _FakeTenant(7) + _record(app, 'Seven Confidential') + + with app.test_request_context(): + g.tenant = _FakeTenant(9) + others = storage.load_all() + + names = [r.get('project_name') for r in others] + assert 'Seven Confidential' not in names + assert others == [] + finally: + app.config['MULTI_TENANT_ENABLED'] = False + + +def test_unresolved_tenant_refuses_rather_than_sharing(client, app): + """Fail closed. Falling back to the shared root would be a silent + cross-tenant leak; an exception is loud and safe.""" + from flask import g + from app.enrollment import storage + + app.config['MULTI_TENANT_ENABLED'] = True + try: + with app.test_request_context(): + g.tenant = None + with pytest.raises(storage.TenantUnresolved): + storage.enrollment_dir() + finally: + app.config['MULTI_TENANT_ENABLED'] = False + + +# ── Branding ───────────────────────────────────────────────────────────────── + +def test_no_hardcoded_upstream_branding_reaches_a_tenant(client): + """ST hardcodes its own company name and a personal address in this module. + Neither may ever be shown to another tenant's customers.""" + import pathlib + root = pathlib.Path('app/enrollment') + for path in root.rglob('*'): + if path.is_file() and path.suffix in ('.py', '.html'): + text = path.read_text(encoding='utf-8') + assert 'L.T' not in text, f'{path} still carries upstream branding' + assert 'da.nguyen8744' not in text, f'{path} still carries a personal address'