df918ed38f
Contract 0.12.0: send_email/send_alert join the plugin surface (the mailer was core-only), PLUGIN-HOOKS and status docs updated, manifest pins the new floor. The alert fires inside _ledger_write only when a decrement CROSSES the item's threshold - one alert per depletion, rearmed by restocking above - and is best-effort after the commit so mail trouble can never fail a take. Recipients come from printedparts_alert_email, falling back to the site alert_recipients. on_enable re-seeds settings idempotently so existing installs pick up new keys. Crossing/rearm semantics proven by test.
203 lines
8.7 KiB
Python
203 lines
8.7 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']
|