aa4bfcd41c
printeditemfiles lands as the plugin's first incremental migration (0002 on the plugin chain - the ADR-008 payoff). Revisions are append-only per item: upload assigns the next number, records the uploader from the JWT, enforces an extension allowlist and a 100 MB cap; download serves the original filename; a permission-gated delete covers wrong-file mistakes. The detail page gains the revision table with a current badge. Unique storedfilename is sized 191 so the index fits MySQL's 767-byte prefix - the per-plugin chain does not apply the core env's ROW_FORMAT hook. Alert recipients gain roles: Role joins the 0.13.0 surface, a role picker on the settings page, and every active member of the selected roles is folded into the deduped recipient list.
329 lines
14 KiB
Python
329 lines
14 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_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}')
|
|
assert lookup.status_code == 200
|
|
|
|
take = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 2})
|
|
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})
|
|
assert too_many.status_code == 400
|
|
|
|
unknown = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1})
|
|
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})
|
|
|
|
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})
|
|
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').get_json()['data']
|
|
assert all(row['printeditemid'] != item for row in listed)
|
|
kiosk = client.get('/api/printedparts/kiosk/item/3DP-9001')
|
|
assert kiosk.status_code == 404
|
|
|
|
including = client.get('/api/printedparts/items?active=false')
|
|
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').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').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')
|
|
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})
|
|
assert take.status_code == 200
|
|
assert captured['to'] == ['crewone@site.test']
|