shopdb-flask/tests/test_winc_import_routes.py

"""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
from datetime import datetime

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)


def _commit(client, headers, csv_text=EXPORT_CSV, extra=None, planhash=None):
    """Preview, then commit the plan the preview returned.

    This is the real flow, and the tests have to use it: the commit route now
    requires the planhash from the preview. Committing without previewing was
    what made the approval meaningless - the writer compared a freshly built
    packet against the plan it was built from, so its drift check could not
    fail and an admin could approve one plan and apply another.

    Pass planhash explicitly to simulate approving one plan and committing a
    different one.
    """
    if planhash is None:
        preview = _upload(client, headers, PREVIEW, csv_text=csv_text)
        planhash = preview.get_json()['data']['planhash']
    payload = dict(extra or {})
    payload['planhash'] = planhash
    return _upload(client, headers, COMMIT, csv_text=csv_text, extra=payload)


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 = _commit(client, auth_headers)
    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 = _commit(client, auth_headers, 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 = _commit(client, auth_headers)
    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 = _commit(client, auth_headers, 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 = _commit(client, auth_headers, 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)


# --- an approved plan has to be the plan that gets applied --------------------
#
# The writer's drift check compared a packet against the plan it was built from,
# so it could never fail. "Approved" meant only that the admin ticked the right
# needs-create group NAMES - the plan itself could be anything.

def test_commit_refuses_a_plan_that_was_not_the_one_previewed(client, db,
                                                              auth_headers):
    _seed_machine(db)
    response = _commit(client, auth_headers,
                       extra={'acknowledge': 'businessunit'},
                       planhash='a' * 64)
    assert response.status_code == 400
    assert 'previewed' in response.get_json()['data']['error']['message']


def test_commit_refuses_without_a_planhash_at_all(client, db, auth_headers):
    """Committing without previewing is what made the approval vacuous."""
    _seed_machine(db)
    response = _upload(client, auth_headers, COMMIT,
                       extra={'acknowledge': 'businessunit'})
    assert response.status_code == 400
    assert 'planhash' in response.get_json()['data']['error']['message']


def test_a_preview_of_one_file_does_not_approve_another(client, db,
                                                        auth_headers):
    """The concrete attack: preview file A, commit file B."""
    from shopdb.core.models import Asset

    _seed_machine(db)
    other_csv = (
        'assetnum,description,location,eq1\r\n'
        'W0083A,Something Else Entirely,WM-J3,lg parts\r\n'
    )
    approved = _upload(client, auth_headers, PREVIEW).get_json()['data']['planhash']
    response = _upload(client, auth_headers, COMMIT, csv_text=other_csv,
                       extra={'acknowledge': 'businessunit',
                              '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'