Add reversible Maximo/WINC machine-adoption import

Machine data arrives as a Maximo asset export (validated against the WINC
site) that must update existing machines in place, not create canonical
per-table rows, so it does not fit the ShopDB-native/Azure/Thor CSV adapter
path. Bulk edits to live machines also need an escape hatch, so the whole
commit is journaled and can be undone as one batch.

Add a dedicated dry-run analyzer, approval packet, resolver/suggest helpers,
and a reversible writer/undo backed by a new importbatches/importjournalentries
schema (with migration). Expose preview/commit/batches/undo settings routes and
a matching frontend source card and panel, and surface the batch/journal
contract through shopdb.api for plugins.

tests/test_winc_import_routes.py

@@ -0,0 +1,210 @@
+"""WINC machine-adoption import routes: preview / commit / undo.
+
+WINC is not a canonical-CSV source adapter (that path is covered by
+test_import_sources). It updates existing machines only, needs explicit
+acknowledgment before creating a business unit, and every commit is journalled
+so it can be undone. These tests drive the real routes end to end against the
+in-memory schema: seed a machine asset, upload an export that matches it, and
+assert what preview reports, what commit writes, and that undo puts it back.
+"""
+import io
+
+import pytest
+
+
+def _seed_machine(db, assetnumber='0083A Debur'):
+    """Create one machine asset the export can match on, plus a J3 location.
+
+    Returns (assetid, locationid). The workstation key of '0083A Debur' is
+    '0083A', which the export row 'W0083A' reduces to, so the two crosswalk.
+    """
+    from shopdb.core.models import AssetType, AssetStatus, Location
+    from shopdb.core.models.asset import Asset
+    from plugins.machines.models.machine import Machine
+
+    assettype = AssetType(assettype='machine')
+    status = AssetStatus(status='In Use')
+    location = Location(locationname='J3')
+    db.session.add_all([assettype, status, location])
+    db.session.flush()
+
+    asset = Asset(
+        assetnumber=assetnumber,
+        name='Old Name',
+        assettypeid=assettype.assettypeid,
+        statusid=status.statusid,
+    )
+    db.session.add(asset)
+    db.session.flush()
+
+    db.session.add(Machine(assetid=asset.assetid))
+    db.session.commit()
+    return asset.assetid, location.locationid
+
+
+# An export whose single in-scope row matches the seeded machine. 'WM-J3'
+# resolves to the J3 location; 'lg parts' aliases to the 'Large Parts' business
+# unit (present or needs-create depending on the test).
+EXPORT_CSV = (
+    'assetnum,description,location,eq1\r\n'
+    'W0083A,Renamed Machine,WM-J3,lg parts\r\n'
+    'FACILITIES-1,Not a machine,,\r\n'
+)
+
+
+def _upload(client, headers, url, csv_text=EXPORT_CSV, extra=None):
+    data = {'files': (io.BytesIO(csv_text.encode('utf-8')), 'winc.csv')}
+    if extra:
+        data.update(extra)
+    return client.post(url, data=data, content_type='multipart/form-data',
+                       headers=headers)
+
+
+PREVIEW = '/api/settings/asset-import/winc/preview'
+COMMIT = '/api/settings/asset-import/winc/commit'
+UNDO = '/api/settings/asset-import/winc/undo'
+BATCHES = '/api/settings/asset-import/winc/batches'
+
+
+def test_preview_reports_plan_without_writing(client, db, auth_headers):
+    _seed_machine(db)
+
+    response = _upload(client, auth_headers, PREVIEW)
+    assert response.status_code == 200, response.get_json()
+    report = response.get_json()['data']
+
+    # One workstation row in scope, the facilities row filtered out.
+    assert report['committed'] is False
+    assert report['rowcount'] == 1
+    assert report['summary']['in_scope'] == 1
+    assert report['summary']['non_workstation_rows'] == 1
+    # 'Large Parts' does not exist yet, so it is flagged for acknowledgment.
+    assert report['needscreate']['businessunit'] == {'Large Parts': 1}
+    assert report['blocking'] is False
+
+    # A preview must not have touched the asset.
+    from shopdb.core.models.asset import Asset
+    assert Asset.query.first().name == 'Old Name'
+
+
+def test_preview_rejects_non_winc_upload(client, db, auth_headers):
+    _seed_machine(db)
+    response = _upload(client, auth_headers, PREVIEW,
+                       csv_text='foo,bar\r\n1,2\r\n')
+    assert response.status_code == 400
+    assert 'assetnum' in response.get_json()['data']['error']['message']
+
+
+def test_commit_requires_acknowledgment_for_needs_create(client, db,
+                                                         auth_headers):
+    _seed_machine(db)
+    # No acknowledge field: the needs-create business unit blocks the write.
+    response = _upload(client, auth_headers, COMMIT)
+    assert response.status_code == 400
+    message = response.get_json()['data']['error']['message'].lower()
+    assert 'acknowledge' in message
+
+    from shopdb.core.models.asset import Asset
+    from shopdb.core.models import BusinessUnit
+    assert Asset.query.first().name == 'Old Name'
+    assert BusinessUnit.query.count() == 0
+
+
+def test_commit_applies_plan_and_creates_business_unit(client, db,
+                                                       auth_headers):
+    assetid, locationid = _seed_machine(db)
+
+    response = _upload(client, auth_headers, COMMIT,
+                       extra={'acknowledge': 'businessunit'})
+    assert response.status_code == 200, response.get_json()
+    report = response.get_json()['data']
+    assert report['committed'] is True
+    assert report['batchid']
+
+    from shopdb.core.models.asset import Asset
+    from shopdb.core.models import BusinessUnit
+    asset = db.session.get(Asset, assetid)
+    assert asset.name == 'Renamed Machine'
+    assert asset.locationid == locationid
+    created = BusinessUnit.query.filter_by(businessunit='Large Parts').one()
+    assert asset.businessunitid == created.businessunitid
+
+
+def test_commit_uses_existing_business_unit_when_present(client, db,
+                                                         auth_headers):
+    from shopdb.core.models import BusinessUnit
+    existing = BusinessUnit(businessunit='Large Parts')
+    db.session.add(existing)
+    db.session.commit()
+    existing_id = existing.businessunitid
+
+    assetid, _ = _seed_machine(db)
+
+    # Already resolved, so no needs-create and no acknowledgment needed.
+    response = _upload(client, auth_headers, COMMIT)
+    assert response.status_code == 200, response.get_json()
+
+    from shopdb.core.models.asset import Asset
+    asset = db.session.get(Asset, assetid)
+    assert asset.businessunitid == existing_id
+    # No duplicate business unit was created.
+    assert BusinessUnit.query.filter_by(businessunit='Large Parts').count() == 1
+
+
+def test_undo_reverses_a_committed_batch(client, db, auth_headers):
+    assetid, _ = _seed_machine(db)
+
+    committed = _upload(client, auth_headers, COMMIT,
+                        extra={'acknowledge': 'businessunit'})
+    batchid = committed.get_json()['data']['batchid']
+
+    from shopdb.core.models.asset import Asset
+    from shopdb.core.models import BusinessUnit
+    assert db.session.get(Asset, assetid).name == 'Renamed Machine'
+
+    response = client.post(UNDO, json={'batchid': batchid}, headers=auth_headers)
+    assert response.status_code == 200, response.get_json()
+    result = response.get_json()['data']
+    assert result['batchid'] == batchid
+    assert result['reverted'] > 0
+    assert result['conflicts'] == []
+
+    db.session.expire_all()
+    asset = db.session.get(Asset, assetid)
+    assert asset.name == 'Old Name'
+    assert asset.locationid is None
+    assert asset.businessunitid is None
+    # The importer-created business unit was hard-deleted on undo.
+    assert BusinessUnit.query.filter_by(businessunit='Large Parts').count() == 0
+
+
+def test_undo_requires_batchid(client, db, auth_headers):
+    response = client.post(UNDO, json={}, headers=auth_headers)
+    assert response.status_code == 400
+    assert 'batchid' in response.get_json()['data']['error']['message'].lower()
+
+
+def test_batches_lists_committed_then_undone(client, db, auth_headers):
+    _seed_machine(db)
+    committed = _upload(client, auth_headers, COMMIT,
+                        extra={'acknowledge': 'businessunit'})
+    batchid = committed.get_json()['data']['batchid']
+
+    listed = client.get(BATCHES, headers=auth_headers).get_json()['data']
+    batch = next(b for b in listed['batches'] if b['batchid'] == batchid)
+    assert batch['status'] == 'applied'
+    assert batch['kind'] == 'winc-machine-adoption'
+
+    client.post(UNDO, json={'batchid': batchid}, headers=auth_headers)
+    listed = client.get(BATCHES, headers=auth_headers).get_json()['data']
+    batch = next(b for b in listed['batches'] if b['batchid'] == batchid)
+    assert batch['status'] == 'undone'
+    assert batch['undoneat'] is not None
+
+
+def test_winc_routes_require_admin(client, db, member_headers):
+    _seed_machine(db)
+    for url in (PREVIEW, COMMIT):
+        response = _upload(client, member_headers, url)
+        assert response.status_code in (401, 403)
+    assert client.get(BATCHES, headers=member_headers).status_code in (401, 403)