Make a partly-undone import retryable, and refuse a plan it cannot deliver

tests/test_winc_import_routes.py

@@ -8,6 +8,7 @@ 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
+from datetime import datetime
 
 import pytest
 
@@ -268,3 +269,153 @@ def test_a_preview_of_one_file_does_not_approve_another(client, db,
                               'planhash': approved})
     assert response.status_code == 400
     assert Asset.query.filter_by(name='Something Else Entirely').first() is None
+
+
+# --- an undo that hit a conflict must stay retryable -------------------------
+#
+# undo_batch used to write 'undone' whatever happened, and its early return
+# refuses a second undo on an undone batch - so the one entry that needed a
+# retry was the only one that could never get it.
+
+def test_a_conflicted_undo_is_marked_partial_and_can_be_retried(
+        client, db, auth_headers, app):
+    from shopdb.core.models import Asset, ImportBatch, BATCH_PARTIAL, BATCH_UNDONE
+
+    assetid, _ = _seed_machine(db)
+    committed = _commit(client, auth_headers, extra={'acknowledge': 'businessunit'})
+    assert committed.status_code == 200
+    batchid = committed.get_json()['data']['batchid']
+
+    # A human edits a cell the import wrote. Undo must refuse to clobber it.
+    with app.app_context():
+        asset = db.session.get(Asset, assetid)
+        asset.name = 'Edited By A Human'
+        db.session.commit()
+
+    first = client.post(UNDO, json={'batchid': batchid}, headers=auth_headers)
+    assert first.status_code == 200
+    body = first.get_json()['data']
+    assert body['conflicts'], 'expected the human edit to conflict'
+
+    with app.app_context():
+        assert db.session.get(ImportBatch, batchid).status == BATCH_PARTIAL
+
+    # The human resolves it. The retry must actually do something - this is
+    # what the old 'undone' status made impossible.
+    with app.app_context():
+        db.session.get(Asset, assetid).name = 'Renamed Machine'
+        db.session.commit()
+
+    second = client.post(UNDO, json={'batchid': batchid}, headers=auth_headers)
+    assert second.status_code == 200
+    assert second.get_json()['data']['reverted'] >= 1, 'retry did nothing'
+
+    with app.app_context():
+        assert db.session.get(ImportBatch, batchid).status == BATCH_UNDONE
+
+
+def test_a_clean_undo_is_still_marked_undone(client, db, auth_headers, app):
+    from shopdb.core.models import ImportBatch, BATCH_UNDONE
+
+    _seed_machine(db)
+    committed = _commit(client, auth_headers, extra={'acknowledge': 'businessunit'})
+    batchid = committed.get_json()['data']['batchid']
+
+    response = client.post(UNDO, json={'batchid': batchid}, headers=auth_headers)
+    assert response.status_code == 200
+    assert not response.get_json()['data']['conflicts']
+    with app.app_context():
+        batch = db.session.get(ImportBatch, batchid)
+        assert batch.status == BATCH_UNDONE
+        assert batch.undoneat is not None
+
+
+# --- the guards themselves, not just the paths through them ------------------
+
+def test_undo_refuses_a_business_unit_something_else_points_at(
+        client, db, auth_headers, app):
+    """_UNDO_REFERENCES named only assets, so a BU referenced from anywhere
+    else was hard-deleted anyway. The guard is a SELECT COUNT(*), so SQLite
+    exercises it exactly as MySQL does."""
+    from shopdb.core.models import BusinessUnit, DashboardDefault
+
+    _seed_machine(db)
+    committed = _commit(client, auth_headers, extra={'acknowledge': 'businessunit'})
+    assert committed.status_code == 200
+    batchid = committed.get_json()['data']['batchid']
+
+    with app.app_context():
+        created = BusinessUnit.query.filter_by(businessunit='Large Parts').first()
+        assert created is not None, 'the import should have created the BU'
+        db.session.add(DashboardDefault(businessunitid=created.businessunitid))
+        db.session.commit()
+        buid = created.businessunitid
+
+    response = client.post(UNDO, json={'batchid': batchid}, headers=auth_headers)
+    assert response.status_code == 200
+    reasons = [c['reason'] for c in response.get_json()['data']['conflicts']]
+    assert any('dashboarddefaults' in r for r in reasons), reasons
+
+    with app.app_context():
+        assert db.session.get(BusinessUnit, buid) is not None, \
+            'the BU was deleted despite still being referenced'
+
+
+def test_undo_refuses_a_batch_from_another_importer(client, db, auth_headers, app):
+    from shopdb.core.models import ImportBatch, BATCH_APPLIED
+
+    with app.app_context():
+        db.session.add(ImportBatch(batchid='not-ours', kind='some-other-import',
+                                   planhash='0' * 64, rowcount=0,
+                                   createdat=datetime.utcnow(),
+                                   status=BATCH_APPLIED))
+        db.session.commit()
+
+    response = client.post(UNDO, json={'batchid': 'not-ours'}, headers=auth_headers)
+    assert response.status_code == 400, response.get_json()
+    assert 'different importer' in response.get_json()['data']['error']['message']
+
+
+# --- a group the writer cannot create must block, visibly ---------------------
+#
+# _PACKET_GROUPS reports on location, but apply_plan only ever creates business
+# units: it sets locationid only for a location that already RESOLVED. So a plan
+# needing a new location used to preview clean, get acknowledged, commit, and
+# silently do none of it.
+
+# 'WM-T6/U6' is an alias override onto the canonical location 'T6'. Seeding the
+# machine without a T6 location is what makes the location NEEDS_CREATE rather
+# than UNRESOLVED - an unaliased bay is merely unresolved, which already blocked.
+NEEDS_LOCATION_CSV = (
+    'assetnum,description,location,eq1\r\n'
+    'W0083A,Renamed Machine,WM-T6/U6,lg parts\r\n'
+)
+
+
+def test_preview_blocks_and_names_a_group_it_cannot_create(client, db,
+                                                           auth_headers):
+    _seed_machine(db)
+
+    response = _upload(client, auth_headers, PREVIEW,
+                       csv_text=NEEDS_LOCATION_CSV)
+    assert response.status_code == 200, response.get_json()
+    report = response.get_json()['data']
+
+    assert report['needscreate'].get('location') == {'T6': 1}, report
+    assert report['blocking'] is True, report
+    # Named, not merely blocked: the page renders this list, and without it the
+    # banner says "blocked" with nothing on screen to act on.
+    assert report['uncreatable'].get('location') == {'T6': 1}, report
+
+
+def test_commit_refuses_a_group_it_cannot_create(client, db, auth_headers):
+    from shopdb.core.models.asset import Asset
+
+    assetid, _ = _seed_machine(db)
+    response = _commit(client, auth_headers, csv_text=NEEDS_LOCATION_CSV,
+                       extra={'acknowledge': 'businessunit,location'})
+    assert response.status_code == 400, response.get_json()
+
+    with client.application.app_context():
+        assert db.session.get(Asset, assetid).name == 'Old Name', \
+            'nothing may be written'