Files
shopdb-flask/tests/test_plugins/test_printedparts_ledger.py
cproudlock 6ed3da1b64
Some checks failed
CI / backend (push) Failing after 1m43s
CI / naming (push) Successful in 2s
CI / frontend (push) Successful in 8s
CI / migrations-mysql (push) Failing after 8s
printedparts stage 7: the kiosk - scan bin, scan badge, keypad, take
Two open endpoints: an item lookup by scanned code and the take POST -
the product's first unauthenticated write, held to the decision
record's bar (decrement-only, badge-attributed server-side, bounded,
physically rate-limited; justification in the plugin README). The
/parts-kiosk route is a full-screen no-auth view beside /shopfloor: a
hidden always-focused input consumes keyboard-wedge scans for
whichever step is active, TouchKeypad (net-new 3x4 grid) takes the
quantity, and a success screen resets after a few seconds. Manual
type-in fallbacks cover damaged labels. Kiosk test proves open access,
the over-take guard, the badge policy, and cache==ledger afterward.
2026-07-17 07:49:13 -04:00

169 lines
7.1 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