Blank asset numbers, contested device claims, and honest toner forecasting

tests/test_plugins/test_collector_deviceid.py

@@ -102,8 +102,25 @@ def _controlled(pcname, label):
 
 # --------------------------------------------------------------- part markers
 
+def _retire(db, hostname):
+    """Move a PC off In Use, which is the one-step way to yield its device."""
+    from shopdb.core.models import AssetStatus
+    retired = AssetStatus.query.filter_by(status='Retired').first()
+    if retired is None:
+        retired = AssetStatus(status='Retired')
+        db.session.add(retired)
+        db.session.flush()
+    pc = Asset.query.filter(Asset.assetnumber.ilike(hostname)).first()
+    pc.statusid = retired.statusid
+    db.session.commit()
+
+
 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."""
+    """THE case this exists for. Same physical marker, two different PCs.
+
+    Neither PC mints a twin. Who HOLDS the marker is settled separately, by the
+    two tests below - this one pins only that the physical unit stays one row.
+    """
     _asset(db, '0613')
     marker = _marker(db, 'PM-0613-A')
 
@@ -120,12 +137,51 @@ def test_a_pc_swap_does_not_mint_a_second_marker(client, db, rig, collector_key)
 
     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_a_live_incumbent_keeps_the_marker_and_the_challenger_is_dormant(
+        client, db, rig, collector_key):
+    """Two PCs naming one device must not both hold it actively.
+
+    Before this, neither device path looked at who else held the target, so a
+    replaced PC kept its link forever and a copied asset-id.txt claimed the same
+    marker from every bay, silently.
+    """
+    _asset(db, '0613')
+    _marker(db, 'PM-0613-A')
+    _report(client, collector_key, 'FMARK100', pctype='gea-shopfloor-partmarker',
+            machinenumber='0613', deviceid='PM-0613-A')
+    resp = _report(client, collector_key, 'FMARK200',
+                   pctype='gea-shopfloor-partmarker',
+                   machinenumber='0613', deviceid='PM-0613-A')
+
+    assert _controlled('FMARK100', 'collector:partmarker') == ['PM-0613-A']
+    assert _controlled('FMARK200', 'collector:partmarker') == []
+    warnings = ' '.join(resp.get_json()['data'].get('warnings', []))
+    assert 'FMARK100' in warnings and 'PM-0613-A' in warnings
+
+
+def test_handover_completes_once_the_incumbent_yields(client, db, rig,
+                                                      collector_key):
+    """The swap case as it actually happens: the old PC is retired or goes quiet."""
+    _asset(db, '0613')
+    _marker(db, 'PM-0613-A')
+    _report(client, collector_key, 'FMARK100', pctype='gea-shopfloor-partmarker',
+            machinenumber='0613', deviceid='PM-0613-A')
+    _retire(db, 'FMARK100')
+
+    _report(client, collector_key, 'FMARK200', pctype='gea-shopfloor-partmarker',
+            machinenumber='0613', deviceid='PM-0613-A')
+
+    assert _controlled('FMARK200', 'collector:partmarker') == ['PM-0613-A']
+    # Archived, never deleted: "which PC drove this in June" stays answerable.
+    assert _controlled('FMARK100', 'collector:partmarker') == []
+    assert AssetRelationship.query.filter_by(label='collector:partmarker').count() >= 2
+
+
 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."""
@@ -150,6 +206,11 @@ def test_an_unknown_device_warns_and_links_nothing(client, db, rig,
     assert Asset.query.filter_by(assetnumber='PM-TYPO').first() is None
     warnings = ' '.join(resp.get_json()['data'].get('warnings', []))
     assert 'PM-TYPO' in warnings
+    # LINKS NOTHING, which is what the name claims. This used to warn and then
+    # mint FMARK500-PARTMARKER anyway - the twin the whole path exists to stop -
+    # and the test passed because it only checked that PM-TYPO was not created.
+    assert Asset.query.filter_by(assetnumber='FMARK500-PARTMARKER').first() is None
+    assert _controlled('FMARK500', 'collector:partmarker') == []
 
 
 def test_a_device_of_the_wrong_type_is_refused(client, db, rig, collector_key):
@@ -160,7 +221,8 @@ def test_a_device_of_the_wrong_type_is_refused(client, db, rig, collector_key):
                    pctype='gea-shopfloor-partmarker',
                    machinenumber='0617', deviceid='PLAIN-MACHINE')
     assert resp.status_code in (200, 201)
-    assert _controlled('FMARK600', 'collector:partmarker') != ['PLAIN-MACHINE']
+    assert _controlled('FMARK600', 'collector:partmarker') == []
+    assert Asset.query.filter_by(assetnumber='FMARK600-PARTMARKER').first() is None
     warnings = ' '.join(resp.get_json()['data'].get('warnings', []))
     assert 'PLAIN-MACHINE' in warnings