271 lines
11 KiB
Python
271 lines
11 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
|
|
|
|
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
|