ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a1802a2bf53b277f751b48c72f7d6774ae9d810f
shopdb-flask/tests/test_plugins/test_printedparts_ledger.py
cproudlock a1802a2bf5 Match the number on the bin, and let an unset kiosk say so 
The last two kiosk findings.

Keypad entry collected EVERY digit in an item's 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. The number on a label is the number at the
end, so the trailing digit run is what is compared now. An ambiguous number is
still refused rather than guessed - taking from the wrong bin is worse than
being asked to scan the label.

A parts kiosk that had never been provisioned drew its normal "Scan the barcode
on the bin" screen and answered "Not found" to every scan, which reads as "we
do not stock that part" rather than "this screen was never set up". The server
cannot say which: the routes answer a bare 404 on purpose, so a caller without
the device token does not learn the door exists, and that stays. But the BROWSER
knows whether it holds a token, and saying so discloses nothing - an
unprovisioned screen now says it is not set up and how to fix it.

For the case the client genuinely cannot distinguish - kiosk switched off versus
a part that is not stocked, both a 404 by design - three misses in a row add a
line suggesting the kiosk may have been switched off. Truthful without the
server having to leak anything.

Three tests, verified by planting the all-digits match back.

Scope: product
2026-09-03 17:00:49 -04:00

655 lines
30 KiB
Python
Raw Blame History

"""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()
Reference in New Issue View Git Blame Copy Permalink