afce24379c
A sweep of the URL map found 165 GET routes that answered anybody, with no token at all: the whole asset inventory, every machine, PC, printer and network device, the USB checkout history and who is holding which drive, the staff directory with its photographs, every core report. Guest access had been argued about one feature-flag at a time while the front door stood open. Each of those now carries the read permission it belongs to. That also makes the guest role mean something: a site can open exactly the reads it chooses, and nothing follows by accident. Thirty-three endpoints stay public, and each is named with its reason rather than left to be inferred: the app shell and its bootstrap, the settings allowlist (the handler already filters an anonymous caller down to branding and a few keys), images fetched by <img src> which carries no Authorization header, what a display kiosk renders before anybody signs in, the parts and USB kiosks, the client scripts that run as SYSTEM before a user session exists, and a secure-send link whose unguessable token IS the authorisation. tests/test_core/test_public_read_surface.py is the standing gate, the companion to the mutating-route sweep that has guarded writes for months. A new public read fails CI until somebody adds it to that list in front of a reviewer. Four further reads need a login but no permission, because the handler decides from who you are - your own user record, your own API tokens, the role catalogue the grid renders - and the same test proves each still refuses an anonymous caller. Verified not vacuous by ungating one route and watching the sweep name it. The employees plugin gained employees.view; the directory is names, SSOs, photographs and reporting lines, and it had no permission of any kind. Around ninety tests asserted payload shapes against endpoints that answered anybody. They are about what the answer says, not about who may ask, so they are signed in now rather than the gate being loosened. The ones that assert an anonymous caller is refused were left exactly as they were.
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', headers=auth_headers)
|
|
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}', headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['data']['photourl'] == f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png'
|