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.
425 lines
19 KiB
Python
425 lines
19 KiB
Python
"""Printedparts ledger + badge tests.
|
|
|
|
The invariants that make the plugin trustworthy: itemcode minting, the
|
|
single-commit cache==ledger rule, the below-zero guard, quantity edits
|
|
forced through the ledger, and the badge contract (SSO digits, PayNo
|
|
wrap, unknown-badge policy).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from shopdb.api import db
|
|
from shopdb.core.models import Setting
|
|
from plugins.printedparts.models import PrintedItem, PrintedItemTransaction
|
|
|
|
|
|
@pytest.fixture
|
|
def item(app, db):
|
|
with app.app_context():
|
|
row = PrintedItem(itemcode='3DP-9001', itemname='Test clip',
|
|
quantityonhand=0, lowstockthreshold=5)
|
|
db.session.add(row)
|
|
db.session.commit()
|
|
yield row.printeditemid
|
|
|
|
|
|
@pytest.fixture
|
|
def directory_employee(app):
|
|
with app.app_context():
|
|
from plugins.employees.models import DirectoryEmployee
|
|
if not db.session.get(DirectoryEmployee, 502000001):
|
|
db.session.add(DirectoryEmployee(
|
|
sso=502000001, firstname='Pat', lastname='Printer'))
|
|
db.session.commit()
|
|
return '502000001'
|
|
|
|
|
|
def test_create_mints_itemcode(client, auth_headers):
|
|
response = client.post('/api/printedparts/items', json={'itemname': 'Bracket'},
|
|
headers=auth_headers)
|
|
assert response.status_code == 201
|
|
data = response.get_json()['data']
|
|
assert data['itemcode'] == f"3DP{data['printeditemid']:04d}"
|
|
assert data['quantityonhand'] == 0
|
|
|
|
|
|
def test_update_refuses_quantity(client, auth_headers, item):
|
|
response = client.put(f'/api/printedparts/items/{item}',
|
|
json={'quantityonhand': 50}, headers=auth_headers)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_restock_writes_ledger_and_cache(client, auth_headers, app, item,
|
|
directory_employee):
|
|
response = client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 10, 'badge': directory_employee},
|
|
headers=auth_headers)
|
|
assert response.status_code == 200
|
|
assert response.get_json()['data']['quantityonhand'] == 10
|
|
with app.app_context():
|
|
rows = PrintedItemTransaction.query.filter_by(printeditemid=item).all()
|
|
assert len(rows) == 1
|
|
assert rows[0].transactiontype == 'restock'
|
|
assert rows[0].quantitychange == 10
|
|
assert rows[0].employeename == 'Pat Printer'
|
|
cached = db.session.get(PrintedItem, item).quantityonhand
|
|
assert cached == sum(r.quantitychange for r in rows)
|
|
|
|
|
|
def test_payno_badge_shape_resolves(client, auth_headers, item,
|
|
directory_employee):
|
|
response = client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 1,
|
|
'badge': f'0{directory_employee}BZ'},
|
|
headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_adjust_requires_reason_and_floors_at_zero(client, auth_headers, item,
|
|
directory_employee):
|
|
no_reason = client.post(f'/api/printedparts/items/{item}/adjust',
|
|
json={'quantitychange': -1,
|
|
'badge': directory_employee},
|
|
headers=auth_headers)
|
|
assert no_reason.status_code == 400
|
|
|
|
below_zero = client.post(f'/api/printedparts/items/{item}/adjust',
|
|
json={'quantitychange': -1, 'reason': 'test',
|
|
'badge': directory_employee},
|
|
headers=auth_headers)
|
|
assert below_zero.status_code == 400
|
|
|
|
|
|
def test_unknown_badge_denied_then_allowed_by_policy(client, auth_headers, app,
|
|
item):
|
|
denied = client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 1, 'badge': '999999999'},
|
|
headers=auth_headers)
|
|
assert denied.status_code == 422
|
|
|
|
with app.app_context():
|
|
Setting.set('printedparts_unknown_badge', 'allow',
|
|
valuetype='string', category='printedparts')
|
|
db.session.commit()
|
|
try:
|
|
allowed = client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 1, 'badge': '999999999'},
|
|
headers=auth_headers)
|
|
assert allowed.status_code == 200
|
|
assert allowed.get_json()['data']['quantityonhand'] == 1
|
|
finally:
|
|
with app.app_context():
|
|
Setting.set('printedparts_unknown_badge', 'deny',
|
|
valuetype='string', category='printedparts')
|
|
db.session.commit()
|
|
|
|
|
|
def test_anonymous_cannot_mutate(client, item):
|
|
assert client.post('/api/printedparts/items',
|
|
json={'itemname': 'X'}).status_code == 401
|
|
assert client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 1, 'badge': '1'}).status_code == 401
|
|
|
|
|
|
def test_catalog_reads_require_view_permission(client, member_headers, item):
|
|
"""Browsing the catalog is printedparts.view-gated; the kiosk stays open."""
|
|
assert client.get('/api/printedparts/items').status_code == 401
|
|
assert client.get(f'/api/printedparts/items/{item}').status_code == 401
|
|
assert client.get('/api/printedparts/items',
|
|
headers=member_headers).status_code == 403
|
|
assert client.get('/api/printedparts/kiosk/item/3DP-9001').status_code == 200
|
|
|
|
|
|
def test_member_without_permission_gets_403(client, member_headers, item):
|
|
"""Authentication alone is not authorization: a role-less user is denied."""
|
|
assert client.post('/api/printedparts/items', json={'itemname': 'X'},
|
|
headers=member_headers).status_code == 403
|
|
assert client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 1, 'badge': '1'},
|
|
headers=member_headers).status_code == 403
|
|
|
|
|
|
def test_kiosk_take_is_open_decrement_only(client, auth_headers, app, item,
|
|
directory_employee):
|
|
"""The kiosk endpoint needs no auth but only ever decrements stock."""
|
|
itemcode = '3DP-9001'
|
|
stocked = client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 5, 'badge': directory_employee},
|
|
headers=auth_headers)
|
|
assert stocked.status_code == 200
|
|
|
|
lookup = client.get(f'/api/printedparts/kiosk/item/{itemcode}', headers=auth_headers)
|
|
assert lookup.status_code == 200
|
|
|
|
take = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 2}, headers=auth_headers)
|
|
assert take.status_code == 200, take.get_json()
|
|
assert take.get_json()['data']['quantityonhand'] == 3
|
|
|
|
too_many = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 99}, headers=auth_headers)
|
|
assert too_many.status_code == 400
|
|
|
|
unknown = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1}, headers=auth_headers)
|
|
assert unknown.status_code == 422
|
|
|
|
with app.app_context():
|
|
rows = PrintedItemTransaction.query.filter_by(
|
|
printeditemid=item, transactiontype='take').all()
|
|
assert len(rows) == 1
|
|
assert rows[0].quantitychange == -2
|
|
assert rows[0].employeename == 'Pat Printer'
|
|
cached = db.session.get(PrintedItem, item).quantityonhand
|
|
ledgersum = sum(r.quantitychange for r in
|
|
PrintedItemTransaction.query.filter_by(
|
|
printeditemid=item).all())
|
|
assert cached == ledgersum
|
|
|
|
|
|
def test_lowstock_alert_fires_on_crossing_only(client, auth_headers, app, item,
|
|
directory_employee, monkeypatch):
|
|
"""One alert when stock CROSSES the threshold downward; restocking above
|
|
rearms it; staying below does not re-fire."""
|
|
sent = []
|
|
import plugins.printedparts.api.routes as printedparts_routes
|
|
monkeypatch.setattr(
|
|
printedparts_routes, '_send_lowstock_alert',
|
|
lambda alerted_item: sent.append(alerted_item.itemcode))
|
|
|
|
def restock(quantity):
|
|
return client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': quantity,
|
|
'badge': directory_employee},
|
|
headers=auth_headers)
|
|
|
|
def take(quantity):
|
|
return client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001',
|
|
'badge': directory_employee,
|
|
'quantity': quantity}, headers=auth_headers)
|
|
|
|
restock(10) # 10 on hand, threshold 5
|
|
assert take(3).status_code == 200 # 7: above threshold, no alert
|
|
assert sent == []
|
|
assert take(3).status_code == 200 # 4: CROSSES 5 -> one alert
|
|
assert sent == ['3DP-9001']
|
|
assert take(2).status_code == 200 # 2: still below, no re-fire
|
|
assert sent == ['3DP-9001']
|
|
restock(20) # 22: rearmed
|
|
assert take(18).status_code == 200 # 4: crosses again -> second alert
|
|
assert sent == ['3DP-9001', '3DP-9001']
|
|
|
|
|
|
def test_alert_recipients_merge_users_and_freetext(client, auth_headers, app,
|
|
item, directory_employee,
|
|
monkeypatch):
|
|
"""Selected shopdb users' account emails merge with the free-text list,
|
|
deduped; inactive users are skipped."""
|
|
import shopdb.api as contract_surface
|
|
captured = {}
|
|
monkeypatch.setattr(contract_surface, 'send_email',
|
|
lambda to, subject, html, text=None:
|
|
captured.setdefault('to', to) or True)
|
|
|
|
with app.app_context():
|
|
from shopdb.api import User
|
|
from werkzeug.security import generate_password_hash
|
|
active = User(username='partslead', email='lead@site.test',
|
|
passwordhash=generate_password_hash('x'), isactive=True)
|
|
inactive = User(username='oldtimer', email='gone@site.test',
|
|
passwordhash=generate_password_hash('x'),
|
|
isactive=False)
|
|
db.session.add_all([active, inactive])
|
|
db.session.commit()
|
|
Setting.set('printedparts_alert_userids',
|
|
f'{active.userid},{inactive.userid}',
|
|
valuetype='string', category='printedparts')
|
|
Setting.set('printedparts_alert_email',
|
|
'extra@site.test, lead@site.test',
|
|
valuetype='string', category='printedparts')
|
|
db.session.commit()
|
|
|
|
client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 10, 'badge': directory_employee},
|
|
headers=auth_headers)
|
|
take = client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001',
|
|
'badge': directory_employee, 'quantity': 6}, headers=auth_headers)
|
|
assert take.status_code == 200 # 4 on hand: crossed threshold 5
|
|
|
|
assert captured['to'] == ['lead@site.test', 'extra@site.test']
|
|
|
|
|
|
def test_retire_hides_and_restore_returns(client, auth_headers, item):
|
|
"""Retire drops the item from the default list and the kiosk; restore
|
|
brings it back with history intact."""
|
|
assert client.delete(f'/api/printedparts/items/{item}',
|
|
headers=auth_headers).status_code == 200
|
|
|
|
listed = client.get('/api/printedparts/items',
|
|
headers=auth_headers).get_json()['data']
|
|
assert all(row['printeditemid'] != item for row in listed)
|
|
kiosk = client.get('/api/printedparts/kiosk/item/3DP-9001', headers=auth_headers)
|
|
assert kiosk.status_code == 404
|
|
|
|
including = client.get('/api/printedparts/items?active=false',
|
|
headers=auth_headers)
|
|
assert any(row['printeditemid'] == item
|
|
for row in including.get_json()['data'])
|
|
|
|
assert client.post(f'/api/printedparts/items/{item}/restore',
|
|
headers=auth_headers).status_code == 200
|
|
assert client.get('/api/printedparts/kiosk/item/3DP-9001', headers=auth_headers).status_code == 200
|
|
|
|
|
|
def test_file_revisions_append_and_download(client, auth_headers, item, tmp_path):
|
|
"""Uploads mint sequential revisions; download returns the original name."""
|
|
import io
|
|
|
|
first = client.post(f'/api/printedparts/items/{item}/files',
|
|
data={'file': (io.BytesIO(b'solid part'), 'clip_v1.stl'),
|
|
'note': 'initial'},
|
|
headers=auth_headers,
|
|
content_type='multipart/form-data')
|
|
assert first.status_code == 201, first.get_json()
|
|
assert first.get_json()['data']['revision'] == 1
|
|
|
|
second = client.post(f'/api/printedparts/items/{item}/files',
|
|
data={'file': (io.BytesIO(b'G1 X0 Y0'), 'clip_v2.gcode')},
|
|
headers=auth_headers,
|
|
content_type='multipart/form-data')
|
|
assert second.get_json()['data']['revision'] == 2
|
|
|
|
bad = client.post(f'/api/printedparts/items/{item}/files',
|
|
data={'file': (io.BytesIO(b'x'), 'malware.exe')},
|
|
headers=auth_headers,
|
|
content_type='multipart/form-data')
|
|
assert bad.status_code == 400
|
|
|
|
listing = client.get(f'/api/printedparts/items/{item}/files',
|
|
headers=auth_headers).get_json()['data']
|
|
assert [f['revision'] for f in listing] == [2, 1]
|
|
|
|
fileid = listing[1]['fileid']
|
|
download = client.get(f'/api/printedparts/files/{fileid}/download', headers=auth_headers)
|
|
assert download.status_code == 200
|
|
assert download.data == b'solid part'
|
|
assert 'clip_v1.stl' in download.headers['Content-Disposition']
|
|
|
|
|
|
def test_alert_role_members_receive(client, auth_headers, app, item,
|
|
directory_employee, monkeypatch):
|
|
"""Every active member of a selected role gets the alert."""
|
|
import shopdb.api as contract_surface
|
|
captured = {}
|
|
monkeypatch.setattr(contract_surface, 'send_email',
|
|
lambda to, subject, html, text=None:
|
|
captured.setdefault('to', to) or True)
|
|
|
|
with app.app_context():
|
|
from shopdb.api import User, Role
|
|
from werkzeug.security import generate_password_hash
|
|
role = Role(rolename='partscrew', description='3D parts crew')
|
|
member = User(username='crewone', email='crewone@site.test',
|
|
passwordhash=generate_password_hash('x'), isactive=True)
|
|
member.roles.append(role)
|
|
db.session.add_all([role, member])
|
|
db.session.commit()
|
|
Setting.set('printedparts_alert_roleids', str(role.roleid),
|
|
valuetype='string', category='printedparts')
|
|
db.session.commit()
|
|
|
|
client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 10, 'badge': directory_employee},
|
|
headers=auth_headers)
|
|
take = client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001',
|
|
'badge': directory_employee, 'quantity': 6}, headers=auth_headers)
|
|
assert take.status_code == 200
|
|
assert captured['to'] == ['crewone@site.test']
|
|
|
|
|
|
def test_gagelabtag_assigned_searched_and_kiosk_resolved(client, auth_headers):
|
|
"""The internal code stays auto-minted; the gage-lab tag is optional,
|
|
unique, searchable, and the kiosk resolves it (exact and bare digits)."""
|
|
created = client.post('/api/printedparts/items',
|
|
json={'itemname': 'Gage block holder',
|
|
'gagelabtag': 'wjrp0117'},
|
|
headers=auth_headers)
|
|
assert created.status_code == 201
|
|
data = created.get_json()['data']
|
|
assert data['gagelabtag'] == 'WJRP0117'
|
|
assert data['itemcode'].startswith('3DP') # internal code untouched
|
|
|
|
duplicate = client.post('/api/printedparts/items',
|
|
json={'itemname': 'Other',
|
|
'gagelabtag': 'WJRP0117'},
|
|
headers=auth_headers)
|
|
assert duplicate.status_code == 409
|
|
|
|
searched = client.get('/api/printedparts/items?search=WJRP0117',
|
|
headers=auth_headers).get_json()['data']
|
|
assert len(searched) == 1
|
|
|
|
by_tag = client.get('/api/printedparts/kiosk/item/WJRP0117', headers=auth_headers)
|
|
assert by_tag.status_code == 200
|
|
by_digits = client.get('/api/printedparts/kiosk/item/117', headers=auth_headers)
|
|
assert by_digits.status_code == 200
|
|
assert by_digits.get_json()['data']['gagelabtag'] == 'WJRP0117'
|
|
|
|
|
|
def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee, auth_headers):
|
|
"""A label QR payload 'TAG|rev' resolves the item by TAG and records the
|
|
revision on the take (traceability of which print-file rev was consumed)."""
|
|
with app.app_context():
|
|
row = PrintedItem(itemcode='3DP-7000', gagelabtag='WJRP7000',
|
|
itemname='Rev part', quantityonhand=10, lowstockthreshold=2)
|
|
db.session.add(row)
|
|
db.session.commit()
|
|
iid = row.printeditemid
|
|
|
|
take = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': 'WJRP7000|3', 'badge': directory_employee, 'quantity': 1}, headers=auth_headers)
|
|
assert take.status_code == 200, take.get_json()
|
|
|
|
with app.app_context():
|
|
txn = PrintedItemTransaction.query.filter_by(
|
|
printeditemid=iid, transactiontype='take').first()
|
|
assert txn is not None
|
|
assert txn.revision == 3
|
|
|
|
# Manual entry (no |rev) records a null revision, still resolves.
|
|
take2 = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': 'WJRP7000', 'badge': directory_employee, 'quantity': 1}, headers=auth_headers)
|
|
assert take2.status_code == 200
|
|
with app.app_context():
|
|
last = (PrintedItemTransaction.query
|
|
.filter_by(printeditemid=iid, transactiontype='take')
|
|
.order_by(PrintedItemTransaction.transactionid.desc()).first())
|
|
assert last.revision is None
|
|
|
|
|
|
def test_lowstock_routes_to_support_team_webhook(client, app, db, directory_employee, auth_headers):
|
|
"""A low-stock crossing posts to the selected support team's webhook URL."""
|
|
from unittest.mock import patch
|
|
from shopdb.core.models import SupportTeam, Setting
|
|
with app.app_context():
|
|
team = SupportTeam(teamname='3D Print Team',
|
|
webhookurl='https://teams.example/3dprint')
|
|
db.session.add(team)
|
|
db.session.commit()
|
|
Setting.set('printedparts_alert_supportteamid', str(team.supportteamid))
|
|
row = PrintedItem(itemcode='3DP-7100', gagelabtag='WJRP7100',
|
|
itemname='Widget', quantityonhand=6, lowstockthreshold=5)
|
|
db.session.add(row)
|
|
db.session.commit()
|
|
|
|
with patch('shopdb.api.send_webhook') as mock_send:
|
|
# take 2 -> 6 crosses to 4 (<= threshold 5) -> alert fires
|
|
resp = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': 'WJRP7100', 'badge': directory_employee, 'quantity': 2}, headers=auth_headers)
|
|
assert resp.status_code == 200, resp.get_json()
|
|
assert mock_send.called
|
|
assert mock_send.call_args.kwargs['url'] == 'https://teams.example/3dprint'
|