422 lines
17 KiB
Python
422 lines
17 KiB
Python
"""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'
|