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

tests/test_core/test_to_dict_floor.py

@@ -0,0 +1,69 @@
+"""BaseModel.to_dict never publishes a secret-bearing column.
+
+The default serialiser walks __table__.columns and emits every one, so a model
+that does not override it publishes whatever it happens to carry - including
+columns added later by someone not thinking about this method. User is the live
+example: it has no override, so User.to_dict() returned its bcrypt hash. Nothing
+called it in a response path, which was luck.
+
+A blanket allowlist is the stronger rule and would empty the output of every
+model relying on the default, so this pins the floor instead: named columns are
+never emitted, and a model may extend the set.
+"""
+
+import pytest
+
+from shopdb.core.models import User
+from shopdb.core.models.base import BaseModel
+
+
+def test_the_default_serialiser_hides_a_password_hash(app):
+    with app.app_context():
+        user = User(username='floor', email='floor@example.invalid',
+                    passwordhash='$2b$12$notarealhashbutlookslikeone')
+        data = user.to_dict()
+        assert 'passwordhash' not in data
+        # Still a useful dict - this is a filter, not a mute button.
+        assert data['username'] == 'floor'
+        assert data['email'] == 'floor@example.invalid'
+
+
+def test_every_hidden_name_is_actually_dropped(app):
+    """Each entry earns its place: drop one and the column comes back."""
+    with app.app_context():
+        user = User(username='floor2', email='f2@example.invalid',
+                    passwordhash='x')
+        assert 'passwordhash' in {c.name for c in User.__table__.columns}
+        assert 'passwordhash' in BaseModel.HIDDEN_COLUMNS
+        assert 'passwordhash' not in user.to_dict()
+
+
+def test_a_model_can_extend_the_hidden_set(app):
+    """A model with its own secret adds to the floor rather than replacing
+    to_dict wholesale."""
+    with app.app_context():
+        class Extended(User):
+            __abstract__ = True
+            HIDDEN_COLUMNS = BaseModel.HIDDEN_COLUMNS | {'email'}
+
+        assert 'passwordhash' in Extended.HIDDEN_COLUMNS
+        assert 'email' in Extended.HIDDEN_COLUMNS
+
+
+@pytest.mark.parametrize('name', sorted(BaseModel.HIDDEN_COLUMNS))
+def test_no_model_publishes_a_hidden_column_through_the_default(name, app):
+    """Sweep every mapped model: if it carries a hidden column AND relies on the
+    default to_dict, the column must not come out."""
+    with app.app_context():
+        from shopdb.extensions import db
+
+        for mapper in db.Model.registry.mappers:
+            model = mapper.class_
+            table = getattr(model, '__table__', None)
+            if table is None or name not in table.columns:
+                continue
+            if 'to_dict' in vars(model):
+                continue  # overrides the default; its own business
+            assert name in getattr(model, 'HIDDEN_COLUMNS', frozenset()), (
+                '%s carries %s and uses the default to_dict, so it would be '
+                'published' % (model.__name__, name))

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'