"""asset-id.txt names the device, and that name survives a PC swap. Every other identity the collector has for a subordinate device is derived from the PC: reuse looks for a prior link from THIS PC asset, and adoption looks up `-`. Both survive a re-image and neither survives a swap - a new hostname is a new PC asset with no prior link and a predicted number that has never existed, so the same physical device gets a second record while the first keeps its history under a dead PC's name. That is the failure that left 43 measuring tools shadowed by minted twins. These tests pin the fix for BOTH device families, because the part-marker path was modelled on the metrology path as it stood before it was fixed. """ import pytest from shopdb.core.models import Asset, AssetType from shopdb.core.models.relationship import RelationshipType, AssetRelationship KEY = 'deviceid-key' @pytest.fixture def collector_key(app): old = app.config.get('COLLECTOR_API_KEY') app.config['COLLECTOR_API_KEY'] = KEY yield KEY app.config['COLLECTOR_API_KEY'] = old @pytest.fixture def rig(db): for name in ('computer', 'machine', 'measuring_tool'): if not AssetType.query.filter_by(assettype=name).first(): db.session.add(AssetType(assettype=name)) for name in ('controls', 'partof'): if not RelationshipType.query.filter_by(relationshiptype=name).first(): db.session.add(RelationshipType(relationshiptype=name)) db.session.commit() def _report(client, key, hostname, pctype, **extra): payload = {'hostname': hostname, 'pctype': pctype} payload.update(extra) return client.post('/api/collector/computers', json=payload, headers={'X-API-Key': key}) def _asset(db, assetnumber, assettype='machine'): """A bare asset: an operation, or something that is NOT a device.""" at = AssetType.query.filter_by(assettype=assettype).first() asset = Asset(assetnumber=assetnumber, assettypeid=at.assettypeid) db.session.add(asset) db.session.commit() return asset def _marker(db, assetnumber): """A Part Marker the collector did NOT create - the real unit on the floor. The extension row and machine type are what make it a marker; a bare asset of the right number is deliberately refused, which the wrong-type test pins. """ from plugins.machines.models import Machine, MachineType at = AssetType.query.filter_by(assettype='machine').first() mtype = MachineType.query.filter_by(machinetype='Part Marker').first() if mtype is None: mtype = MachineType(machinetype='Part Marker') db.session.add(mtype) db.session.flush() asset = Asset(assetnumber=assetnumber, assettypeid=at.assettypeid) db.session.add(asset) db.session.flush() db.session.add(Machine(assetid=asset.assetid, machinetypeid=mtype.machinetypeid)) db.session.commit() return asset def _tool(db, assetnumber): """A measuring tool the collector did NOT create.""" from plugins.measuringtools.models import MeasuringTool at = AssetType.query.filter_by(assettype='measuring_tool').first() asset = Asset(assetnumber=assetnumber, assettypeid=at.assettypeid) db.session.add(asset) db.session.flush() db.session.add(MeasuringTool(assetid=asset.assetid)) db.session.commit() return asset def _controlled(pcname, label): """Asset numbers this PC controls under a collector label.""" pc = Asset.query.filter(Asset.assetnumber.ilike(pcname)).first() if pc is None: return [] rels = AssetRelationship.query.filter_by( sourceassetid=pc.assetid, label=label, isactive=True).all() return sorted(Asset.query.filter_by(assetid=r.targetassetid).first().assetnumber for r in rels) # --------------------------------------------------------------- part markers def test_a_pc_swap_does_not_mint_a_second_marker(client, db, rig, collector_key): """THE case this exists for. Same physical marker, two different PCs.""" _asset(db, '0613') marker = _marker(db, 'PM-0613-A') first = _report(client, collector_key, 'FMARK100', pctype='gea-shopfloor-partmarker', machinenumber='0613', deviceid='PM-0613-A') assert first.status_code in (200, 201), first.get_data(as_text=True)[:300] # The bay's PC is replaced. New hostname, same marker named in asset-id.txt. second = _report(client, collector_key, 'FMARK200', pctype='gea-shopfloor-partmarker', machinenumber='0613', deviceid='PM-0613-A') assert second.status_code in (200, 201), second.get_data(as_text=True)[:300] assert Asset.query.filter_by(assetnumber='FMARK100-PARTMARKER').first() is None assert Asset.query.filter_by(assetnumber='FMARK200-PARTMARKER').first() is None assert _controlled('FMARK200', 'collector:partmarker') == ['PM-0613-A'] assert Asset.query.filter_by(assetnumber='PM-0613-A').count() == 1 assert marker.assetid == Asset.query.filter_by( assetnumber='PM-0613-A').first().assetid def test_without_the_file_a_swap_still_mints_the_old_way(client, db, rig, collector_key): """The unfixed behaviour, pinned so the file's value stays visible.""" _asset(db, '0614') assert _report(client, collector_key, 'FMARK300', pctype='gea-shopfloor-partmarker', machinenumber='0614').status_code in (200, 201) assert _report(client, collector_key, 'FMARK400', pctype='gea-shopfloor-partmarker', machinenumber='0614').status_code in (200, 201) assert Asset.query.filter_by(assetnumber='FMARK300-PARTMARKER').first() assert Asset.query.filter_by(assetnumber='FMARK400-PARTMARKER').first() def test_an_unknown_device_warns_and_links_nothing(client, db, rig, collector_key): _asset(db, '0616') resp = _report(client, collector_key, 'FMARK500', pctype='gea-shopfloor-partmarker', machinenumber='0616', deviceid='PM-TYPO') assert resp.status_code in (200, 201) assert Asset.query.filter_by(assetnumber='PM-TYPO').first() is None warnings = ' '.join(resp.get_json()['data'].get('warnings', [])) assert 'PM-TYPO' in warnings def test_a_device_of_the_wrong_type_is_refused(client, db, rig, collector_key): """A machine number pasted into asset-id.txt must not become a marker.""" _asset(db, '0617') _asset(db, 'PLAIN-MACHINE') resp = _report(client, collector_key, 'FMARK600', pctype='gea-shopfloor-partmarker', machinenumber='0617', deviceid='PLAIN-MACHINE') assert resp.status_code in (200, 201) assert _controlled('FMARK600', 'collector:partmarker') != ['PLAIN-MACHINE'] warnings = ' '.join(resp.get_json()['data'].get('warnings', [])) assert 'PLAIN-MACHINE' in warnings def test_repeat_cycles_with_the_file_are_stable(client, db, rig, collector_key): _asset(db, '0618') _marker(db, 'PM-0618-A') codes = [_report(client, collector_key, 'FMARK700', pctype='gea-shopfloor-partmarker', machinenumber='0618', deviceid='PM-0618-A').status_code for _ in range(3)] assert codes == [codes[0]] * 3, codes assert _controlled('FMARK700', 'collector:partmarker') == ['PM-0618-A'] # ------------------------------------------------------------ measuring tools def test_the_same_file_serves_a_metrology_bay(client, db, rig, collector_key): """One file, no device type in it: the pc-type decides which sync uses it.""" _tool(db, 'MT-9001') resp = _report(client, collector_key, 'KEYENCE100', pctype='gea-shopfloor-keyence', deviceid='MT-9001') assert resp.status_code in (200, 201), resp.get_data(as_text=True)[:300] assert _controlled('KEYENCE100', 'collector:measuringtool') == ['MT-9001'] assert Asset.query.filter_by(assetnumber='KEYENCE100-KEYENCE').first() is None def test_the_0120_field_name_still_works(client, db, rig, collector_key): """measuringtool-id.txt shipped in 0.12.0; a staged bay keeps reporting.""" _tool(db, 'MT-9002') resp = _report(client, collector_key, 'KEYENCE200', pctype='gea-shopfloor-keyence', measuringtoolid='MT-9002') assert resp.status_code in (200, 201) assert _controlled('KEYENCE200', 'collector:measuringtool') == ['MT-9002'] def test_deviceid_wins_when_both_arrive(client, db, rig, collector_key): _tool(db, 'MT-9003') _tool(db, 'MT-9004') resp = _report(client, collector_key, 'KEYENCE300', pctype='gea-shopfloor-keyence', deviceid='MT-9003', measuringtoolid='MT-9004') assert resp.status_code in (200, 201) assert _controlled('KEYENCE300', 'collector:measuringtool') == ['MT-9003']