655 lines
30 KiB
Python
655 lines
30 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
|
|
|
|
|
|
KIOSK_TOKEN = 'kiosk-device-token-for-tests'
|
|
# The header a provisioned screen sends. A module constant rather than the
|
|
# fixture's return value, so a test does not have to name the fixture just to
|
|
# reach the header.
|
|
KIOSK_HEADERS = {'X-Kiosk-Token': KIOSK_TOKEN}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def kiosk(app, db):
|
|
"""Provision the kiosk the way an administrator would, and hand back the
|
|
header a provisioned screen sends. Without this every kiosk route 404s,
|
|
which is the point: the routes are the only unauthenticated write in the
|
|
product, so they are off until a site opts in."""
|
|
with app.app_context():
|
|
Setting.set('printedparts_kiosk_enabled', 'true')
|
|
Setting.set('printedparts_kiosk_token', KIOSK_TOKEN)
|
|
db.session.commit()
|
|
return KIOSK_HEADERS
|
|
|
|
|
|
def test_catalog_reads_require_view_permission(client, member_headers, item):
|
|
"""Browsing the catalog is printedparts.view-gated; a PROVISIONED kiosk
|
|
reads without an account."""
|
|
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',
|
|
headers=KIOSK_HEADERS).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_decrement_only(client, auth_headers, app, item,
|
|
directory_employee, kiosk):
|
|
"""A provisioned kiosk needs no account 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, **KIOSK_HEADERS})
|
|
assert lookup.status_code == 200
|
|
|
|
take = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 2},
|
|
headers={**auth_headers, **KIOSK_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, **KIOSK_HEADERS})
|
|
assert too_many.status_code == 400
|
|
|
|
unknown = client.post('/api/printedparts/kiosk/take', json={
|
|
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1},
|
|
headers={**auth_headers, **KIOSK_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, **KIOSK_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, **KIOSK_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, **KIOSK_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, **KIOSK_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, **KIOSK_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, **KIOSK_HEADERS})
|
|
assert by_tag.status_code == 200
|
|
by_digits = client.get('/api/printedparts/kiosk/item/117', headers={**auth_headers, **KIOSK_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, **KIOSK_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, **KIOSK_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, **KIOSK_HEADERS})
|
|
assert resp.status_code == 200, resp.get_json()
|
|
assert mock_send.called
|
|
assert mock_send.call_args.kwargs['url'] == 'https://teams.example/3dprint'
|
|
|
|
|
|
# --- the kiosk is off until a site opts in -----------------------------------
|
|
#
|
|
# These are the cases the Darkwing scan turned into an exploit path. Every one
|
|
# of them returned 200 and wrote to the ledger before the gate existed.
|
|
|
|
def _clear_kiosk(app, db):
|
|
with app.app_context():
|
|
Setting.set('printedparts_kiosk_enabled', 'false')
|
|
Setting.set('printedparts_kiosk_token', '')
|
|
db.session.commit()
|
|
|
|
|
|
def test_kiosk_is_shut_when_the_site_has_not_enabled_it(client, app, db, item):
|
|
_clear_kiosk(app, db)
|
|
assert client.get('/api/printedparts/kiosk/item/3DP-9001').status_code == 404
|
|
assert client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001', 'badge': '1',
|
|
'quantity': 1}).status_code == 404
|
|
|
|
|
|
def test_kiosk_is_shut_without_the_device_token(client, item):
|
|
"""Enabled is not enough: an anonymous caller has no token."""
|
|
assert client.get('/api/printedparts/kiosk/item/3DP-9001').status_code == 404
|
|
assert client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001', 'badge': '1',
|
|
'quantity': 1}).status_code == 404
|
|
|
|
|
|
def test_kiosk_is_shut_with_the_wrong_device_token(client, item):
|
|
wrong = {'X-Kiosk-Token': KIOSK_TOKEN + 'x'}
|
|
assert client.get('/api/printedparts/kiosk/item/3DP-9001',
|
|
headers=wrong).status_code == 404
|
|
assert client.post('/api/printedparts/kiosk/take', headers=wrong,
|
|
json={'itemcode': '3DP-9001', 'badge': '1',
|
|
'quantity': 1}).status_code == 404
|
|
|
|
|
|
def test_a_single_take_cannot_drain_a_bin(client, app, item, auth_headers,
|
|
directory_employee):
|
|
"""quantity was bounded only by what was on hand, so one request emptied
|
|
the bin. It is capped per transaction now."""
|
|
from plugins.printedparts.api.routes import KIOSK_MAX_TAKE_QUANTITY
|
|
|
|
client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': KIOSK_MAX_TAKE_QUANTITY + 10,
|
|
'badge': directory_employee}, headers=auth_headers)
|
|
over = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
|
|
json={'itemcode': '3DP-9001',
|
|
'badge': directory_employee,
|
|
'quantity': KIOSK_MAX_TAKE_QUANTITY + 1})
|
|
assert over.status_code == 400
|
|
assert str(KIOSK_MAX_TAKE_QUANTITY) in over.get_json()['data']['error']['message']
|
|
|
|
|
|
def test_true_is_not_a_quantity(client, item, directory_employee):
|
|
"""isinstance(True, int) is True in Python, so a JSON true used to sail
|
|
through the type check and take one."""
|
|
resp = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
|
|
json={'itemcode': '3DP-9001',
|
|
'badge': directory_employee, 'quantity': True})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_the_badge_route_is_not_an_enumeration_oracle(client, item, auth_headers,
|
|
directory_employee):
|
|
"""An unknown badge and a malformed one must be indistinguishable, or the
|
|
route tells an anonymous caller which numbers are real employees.
|
|
|
|
Stock first: the on-hand check runs BEFORE the badge is resolved, so an
|
|
empty bin would answer 400 for both and the test would pass without ever
|
|
reaching the code it is about."""
|
|
client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 5, 'badge': directory_employee},
|
|
headers=auth_headers)
|
|
unknown = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
|
|
json={'itemcode': '3DP-9001', 'badge': '999999999',
|
|
'quantity': 1})
|
|
malformed = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
|
|
json={'itemcode': '3DP-9001', 'badge': 'not-a-badge',
|
|
'quantity': 1})
|
|
assert unknown.status_code == malformed.status_code == 422
|
|
assert (unknown.get_json()['data']['error']['message']
|
|
== malformed.get_json()['data']['error']['message'])
|
|
|
|
|
|
# --- the race ---------------------------------------------------------------
|
|
#
|
|
# The take read quantityonhand, compared it in Python, and then wrote. Five
|
|
# concurrent takes of 5 from a bin holding 5 all passed that comparison and all
|
|
# wrote their own 5 - 5 = 0, so quantityonhand ended up permanently out of step
|
|
# with the ledger that had recorded every one of them. The comparison now lives
|
|
# in the UPDATE's WHERE clause, where it is atomic.
|
|
|
|
def test_a_take_that_passed_its_precheck_is_still_refused_by_the_row(
|
|
client, auth_headers, app, item, directory_employee, kiosk,
|
|
monkeypatch):
|
|
"""Simulate the race window: the request holds an item that says 5 on hand
|
|
while the row has already been emptied by somebody else."""
|
|
from plugins.printedparts.api import routes
|
|
from plugins.printedparts.models import PrintedItem
|
|
|
|
stocked = client.post(f'/api/printedparts/items/{item}/restock',
|
|
json={'quantity': 5, 'badge': directory_employee},
|
|
headers=auth_headers)
|
|
assert stocked.status_code == 200
|
|
|
|
real_find = routes._kiosk_find_item
|
|
|
|
def stale_find(itemcode):
|
|
"""What the losing request sees: its own read, taken before the winner
|
|
committed, still shows a full bin.
|
|
|
|
Detached first, so setting the attribute changes only this request's
|
|
view and never reaches the database - the test is about a stale
|
|
reader, not about a second writer.
|
|
"""
|
|
found = real_find(itemcode)
|
|
if found is not None:
|
|
db.session.expunge(found)
|
|
found.quantityonhand = 5
|
|
return found
|
|
|
|
# Somebody else empties the bin first, through the kiosk, so the ledger
|
|
# records it the way a real competing take would.
|
|
winner = client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001', 'quantity': 5,
|
|
'badge': directory_employee},
|
|
headers=KIOSK_HEADERS)
|
|
assert winner.status_code == 200, winner.get_json()
|
|
|
|
monkeypatch.setattr(routes, '_kiosk_find_item', stale_find)
|
|
taken = client.post('/api/printedparts/kiosk/take',
|
|
json={'itemcode': '3DP-9001', 'quantity': 5,
|
|
'badge': directory_employee},
|
|
headers=KIOSK_HEADERS)
|
|
assert taken.status_code == 400, taken.get_json()
|
|
assert 'on hand' in taken.get_json()['data']['error']['message']
|
|
|
|
with app.app_context():
|
|
after = db.session.query(PrintedItem).filter_by(
|
|
printeditemid=item).first()
|
|
assert after.quantityonhand == 0, 'stock must not go negative'
|
|
ledger = sum(t.quantitychange for t in after.transactions)
|
|
assert ledger == after.quantityonhand, \
|
|
'quantityonhand and the ledger must agree'
|
|
|
|
|
|
# --- the number on the bin is the number you type ---------------------------
|
|
#
|
|
# Keypad entry matched the digits ANYWHERE in the code, and the shipped prefix
|
|
# contains one: '3DP0004' collapsed to 30004, so an operator typing the 4
|
|
# printed on the label was told no part matches that barcode, and had to work
|
|
# out that the leading 3 belonged to the prefix.
|
|
|
|
def test_the_keypad_matches_the_number_printed_on_the_label(client, db,
|
|
auth_headers,
|
|
kiosk,
|
|
directory_employee):
|
|
from plugins.printedparts.models import PrintedItem
|
|
|
|
created = client.post('/api/printedparts/items', headers=auth_headers,
|
|
json={'itemname': 'Keypad clip'})
|
|
assert created.status_code == 201, created.get_json()
|
|
itemcode = created.get_json()['data']['itemcode']
|
|
assert itemcode.startswith('3DP'), itemcode
|
|
|
|
number = str(int(itemcode[3:]))
|
|
found = client.get('/api/printedparts/kiosk/item/%s' % number,
|
|
headers=KIOSK_HEADERS)
|
|
assert found.status_code == 200, (number, found.get_json())
|
|
assert found.get_json()['data']['itemcode'] == itemcode
|
|
|
|
|
|
def test_the_full_code_still_works(client, db, auth_headers, kiosk):
|
|
created = client.post('/api/printedparts/items', headers=auth_headers,
|
|
json={'itemname': 'Full code clip'})
|
|
itemcode = created.get_json()['data']['itemcode']
|
|
|
|
found = client.get('/api/printedparts/kiosk/item/%s' % itemcode,
|
|
headers=KIOSK_HEADERS)
|
|
assert found.status_code == 200, found.get_json()
|
|
|
|
|
|
def test_an_ambiguous_number_is_still_refused(client, db, auth_headers, kiosk):
|
|
"""Two items whose trailing numbers agree must not resolve to a guess -
|
|
taking from the wrong bin is worse than being told to scan the label."""
|
|
from plugins.printedparts.models import PrintedItem
|
|
|
|
first = client.post('/api/printedparts/items', headers=auth_headers,
|
|
json={'itemname': 'Ambiguous one'}).get_json()['data']
|
|
item = db.session.get(PrintedItem, first['printeditemid'])
|
|
item.gagelabtag = 'WJRP%04d' % 7777
|
|
db.session.commit()
|
|
|
|
second = client.post('/api/printedparts/items', headers=auth_headers,
|
|
json={'itemname': 'Ambiguous two'}).get_json()['data']
|
|
other = db.session.get(PrintedItem, second['printeditemid'])
|
|
other.gagelabtag = 'GAGE%04d' % 7777
|
|
db.session.commit()
|
|
|
|
refused = client.get('/api/printedparts/kiosk/item/7777',
|
|
headers=KIOSK_HEADERS)
|
|
assert refused.status_code == 404, refused.get_json()
|