ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
81ad221e6af1fe9aa05d6dab738f07bb499139f0
shopdb-flask/tests/test_plugins/test_printedparts_ledger.py
cproudlock aa8028f96e Let the row refuse the write both kiosks were deciding in Python 
Wave 1 of the remediation plan for the 2026-09-03 feature test run. Two kiosks,
one shape of defect, one critical consequence.

Both kiosks read the state, decided in Python, and then wrote. Nothing stops two
requests interleaving between the read and the write, so both see the same
pre-state and both proceed. Reproduced on MySQL 5.6 over HTTP, five concurrent
takes of 5 from a bin holding 5: all five were accepted and quantityonhand
landed on -20, because each request wrote its own arithmetic. The ledger had
recorded every one of them, so the cached quantity and the ledger disagreed
permanently, which is the number the shop floor actually reads. The USB kiosk
had the same shape: concurrent scans of one drive all succeeded and inserted
several open custody rows, so one drive was recorded out to two people and the
documented 409 never fired.

The comparison now lives in the WHERE clause, where the database settles it at
write time. printedparts._move_quantity issues
UPDATE ... WHERE quantityonhand >= :n and treats rowcount 0 as the refusal,
raising InsufficientStock, which take and adjust turn back into the message they
already gave - carrying the quantity the database holds now, not the stale one
the request read. The low-stock alert derives its before-value from what the
write produced for the same reason. Both USB paths claim the device with
ischeckedout (self-hosted) or status (external cmmc_usb) in the WHERE clause;
the external one claims BEFORE writing the log line, so a refused claim leaves
nothing to undo. Check-in is guarded the same way.

After the fix, the same five concurrent takes: one accepted, four refused,
quantityonhand 0 and the ledger sum 0.

Check-in also closes EVERY open row for the device rather than the newest.
One device can only be out once, so a second open row is corruption, and
closing only the newest left the older ones open forever - the device then read
as returned while its custody history said otherwise.

Separately, usbcheckouts now records who handed a drive back
(checkinsso/checkinname, migration usb0004checkinidentity). Returning for a
colleague is a supported flow that the kiosk asks to confirm, but the row it
wrote attributed the check-in to the HOLDER, so the custody history and the
audit report both named somebody who was not there. Nullable and deliberately
not backfilled: copying the holder's badge into old rows would manufacture
evidence for the exact claim this stops making.

Four tests. Both guards verified by planting: strip the condition out of either
UPDATE and its test fails.

Scope: product
2026-09-03 13:58:26 -04:00

597 lines
28 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'
Reference in New Issue View Git Blame Copy Permalink