1d21bf0206
Model photos: upload/replace/delete on /api/models/<id>/image (admin), stored under instance/modelimages/ with a public serve route; thumbnail plus Upload/Replace/Remove controls in the Models settings modal; the URL field remains as a manual alternative. Employee photos, mode-aware: self-hosted directory employees get upload/replace/delete (photo-<sso> under instance/employeephotos/, employees plugin migration 0002); external directory mode passes the HR-supplied picture URL through read-only (writes 409). One resolver feeds both consumers - the shopfloor recognition/recert kiosk cards and the employee detail hero - in either mode. Navigation fix: router-view is keyed on route path, so following a relationship link between two assets of the same type (machine -> dualpath machine) reloads the page instead of showing stale content; query-only URL changes still avoid a remount. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
213 lines
8.9 KiB
Python
213 lines
8.9 KiB
Python
"""Tests for self-hosted employee photo upload, replace, delete, serve, and the
|
|
shared photo-URL resolver used by both EmployeeDetail and the kiosk cards.
|
|
|
|
Covers: admin-gated upload/delete, the public serve route, one-photo-per-person
|
|
replace semantics, external-mode 409s, and resolve_employee_photo_url in both
|
|
directory modes. The wjf_employees HR DB is not available under test, so external
|
|
resolution is exercised by passing the Picture value directly (no query)."""
|
|
|
|
import io
|
|
import os
|
|
|
|
from shopdb.core.models import Setting
|
|
from plugins.employees.models import DirectoryEmployee
|
|
from plugins.employees.api.routes import (
|
|
resolve_employee_photo_url,
|
|
EMPLOYEE_PHOTO_URL_PREFIX,
|
|
EMPLOYEE_PHOTO_STATIC_PREFIX,
|
|
)
|
|
|
|
|
|
def _set_mode(db, mode):
|
|
row = Setting.query.filter_by(key='employee_directory_mode').first()
|
|
if row:
|
|
row.value = mode
|
|
else:
|
|
db.session.add(Setting(key='employee_directory_mode', value=mode,
|
|
valuetype='string', category='site'))
|
|
db.session.commit()
|
|
|
|
|
|
def _make_employee(db, sso=210000001):
|
|
emp = DirectoryEmployee(sso=sso, firstname='Test', lastname='Person')
|
|
db.session.add(emp)
|
|
db.session.commit()
|
|
return emp
|
|
|
|
|
|
# --- upload / replace / delete (self-hosted) --------------------------------
|
|
|
|
def test_upload_forbidden_for_non_admin(client, db, member_headers):
|
|
"""A role-less authenticated user cannot upload an employee photo."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
data = {'file': (io.BytesIO(b'bytes'), 'p.png')}
|
|
resp = client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=member_headers)
|
|
assert resp.status_code == 403
|
|
assert resp.get_json()['data']['error']['code'] == 'FORBIDDEN'
|
|
|
|
|
|
def test_upload_sets_photofilename_and_file_exists(client, db, auth_headers, app):
|
|
"""Admin upload writes photofilename + photourl and lands the file on disk."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
payload = b'\x89PNG\r\n\x1a\nfake-png'
|
|
data = {'file': (io.BytesIO(payload), 'p.png')}
|
|
resp = client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
assert resp.status_code == 200, resp.get_json()
|
|
|
|
filename = f'photo-{emp.sso}.png'
|
|
body = resp.get_json()['data']
|
|
assert body['photofilename'] == filename
|
|
assert body['photourl'] == f'{EMPLOYEE_PHOTO_URL_PREFIX}{filename}'
|
|
|
|
refreshed = db.session.get(DirectoryEmployee, emp.sso)
|
|
assert refreshed.photofilename == filename
|
|
|
|
path = os.path.join(app.instance_path, 'employeephotos', filename)
|
|
assert os.path.exists(path)
|
|
with open(path, 'rb') as handle:
|
|
assert handle.read() == payload
|
|
|
|
|
|
def test_upload_rejects_invalid_extension(client, db, auth_headers):
|
|
"""A disallowed file extension is rejected."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
data = {'file': (io.BytesIO(b'MZ'), 'p.exe')}
|
|
resp = client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()['data']['error']['code'] == 'VALIDATION_ERROR'
|
|
|
|
|
|
def test_upload_missing_employee_is_404(client, db, auth_headers):
|
|
"""Uploading to a nonexistent SSO is a 404."""
|
|
_set_mode(db, 'selfhosted')
|
|
data = {'file': (io.BytesIO(b'x'), 'p.png')}
|
|
resp = client.post('/api/employees/999999/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_reupload_replaces_and_removes_old_extension(client, db, auth_headers, app):
|
|
"""Re-upload with a different extension deletes the prior file."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
first = {'file': (io.BytesIO(b'first'), 'p.png')}
|
|
client.post(f'/api/employees/{emp.sso}/photo', data=first,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
oldpath = os.path.join(app.instance_path, 'employeephotos', f'photo-{emp.sso}.png')
|
|
assert os.path.exists(oldpath)
|
|
|
|
second = {'file': (io.BytesIO(b'second'), 'p.jpg')}
|
|
resp = client.post(f'/api/employees/{emp.sso}/photo', data=second,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
|
|
assert not os.path.exists(oldpath)
|
|
newpath = os.path.join(app.instance_path, 'employeephotos', f'photo-{emp.sso}.jpg')
|
|
assert os.path.exists(newpath)
|
|
refreshed = db.session.get(DirectoryEmployee, emp.sso)
|
|
assert refreshed.photofilename == f'photo-{emp.sso}.jpg'
|
|
|
|
|
|
def test_serve_returns_bytes(client, db, auth_headers):
|
|
"""The public serve route returns the uploaded bytes without auth."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
payload = b'\x89PNG\r\n\x1a\nserved'
|
|
data = {'file': (io.BytesIO(payload), 'p.png')}
|
|
client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
|
|
served = client.get(f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png')
|
|
assert served.status_code == 200
|
|
assert served.get_data() == payload
|
|
|
|
|
|
def test_delete_clears_field_and_removes_file(client, db, auth_headers, app):
|
|
"""Delete clears photofilename and removes the uploaded file."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
data = {'file': (io.BytesIO(b'bytes'), 'p.png')}
|
|
client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
path = os.path.join(app.instance_path, 'employeephotos', f'photo-{emp.sso}.png')
|
|
assert os.path.exists(path)
|
|
|
|
resp = client.delete(f'/api/employees/{emp.sso}/photo', headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['data']['photourl'] is None
|
|
assert not os.path.exists(path)
|
|
|
|
refreshed = db.session.get(DirectoryEmployee, emp.sso)
|
|
assert refreshed.photofilename is None
|
|
|
|
|
|
def test_delete_forbidden_for_non_admin(client, db, member_headers):
|
|
"""A role-less authenticated user cannot delete an employee photo."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
resp = client.delete(f'/api/employees/{emp.sso}/photo', headers=member_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# --- external-mode 409 ------------------------------------------------------
|
|
|
|
def test_upload_conflict_in_external_mode(client, db, auth_headers):
|
|
"""Upload is a 409 when the directory is external (HR owns the photo)."""
|
|
_set_mode(db, 'external')
|
|
data = {'file': (io.BytesIO(b'x'), 'p.png')}
|
|
resp = client.post('/api/employees/210000001/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
assert resp.status_code == 409
|
|
assert resp.get_json()['data']['error']['code'] == 'CONFLICT'
|
|
|
|
|
|
def test_delete_conflict_in_external_mode(client, db, auth_headers):
|
|
"""Delete is a 409 when the directory is external."""
|
|
_set_mode(db, 'external')
|
|
resp = client.delete('/api/employees/210000001/photo', headers=auth_headers)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
# --- resolver ---------------------------------------------------------------
|
|
|
|
def test_resolver_selfhosted_returns_upload_url_or_none(client, db, auth_headers):
|
|
"""Self-hosted: served upload URL when a photo exists, else None."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
assert resolve_employee_photo_url(emp.sso) is None
|
|
|
|
data = {'file': (io.BytesIO(b'x'), 'p.png')}
|
|
client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
assert resolve_employee_photo_url(emp.sso) == f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png'
|
|
|
|
|
|
def test_resolver_external_prefixes_relative_and_passes_urls(db):
|
|
"""External: relative HR paths get the static prefix; absolute URLs pass through."""
|
|
_set_mode(db, 'external')
|
|
assert resolve_employee_photo_url(210000001, 'Support/210000001.png') == \
|
|
f'{EMPLOYEE_PHOTO_STATIC_PREFIX}Support/210000001.png'
|
|
assert resolve_employee_photo_url(210000001, 'https://hr.example.net/p.png') == \
|
|
'https://hr.example.net/p.png'
|
|
assert resolve_employee_photo_url(210000001, None) is None
|
|
assert resolve_employee_photo_url(None) is None
|
|
|
|
|
|
def test_lookup_includes_photourl(client, db, auth_headers):
|
|
"""The lookup serializer carries the resolved photourl (self-hosted)."""
|
|
_set_mode(db, 'selfhosted')
|
|
emp = _make_employee(db)
|
|
data = {'file': (io.BytesIO(b'x'), 'p.png')}
|
|
client.post(f'/api/employees/{emp.sso}/photo', data=data,
|
|
content_type='multipart/form-data', headers=auth_headers)
|
|
|
|
resp = client.get(f'/api/employees/lookup/{emp.sso}')
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['data']['photourl'] == f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png'
|