Add the printedparts plugin: 3D-printed parts storefront and kiosk

Catalog with photos and append-only print-file revisions, a stock
ledger attributing every take/restock/adjust to a badge-resolved
employee, a touch kiosk (scan bin, scan badge, keypad quantity) with a
deliberately open decrement-only take endpoint, 1x0.5in CODE128 bin
labels, low-stock email alerts to selected users, roles, or addresses,
and stock/consumption/by-person reports with a cache-vs-ledger
reconcile. Ships disabled; enable per site. Contract 0.13.0 adds the
mailer and the User/Role models to the plugin surface.

tests/test_plugins/test_printedparts_ledger.py

@@ -0,0 +1,340 @@
+"""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_catalog_reads_require_view_permission(client, member_headers, item):
+    """Browsing the catalog is printedparts.view-gated; the kiosk stays open."""
+    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').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_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']
+
+
+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})
+    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')
+    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').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')
+    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})
+    assert take.status_code == 200
+    assert captured['to'] == ['crewone@site.test']