Machine claim vs handover; alerts gated off; ntlars empty-value fix

2026-08-10 14:49:17 -04:00
parent 502728c9f6
commit d40be9ca68
3 changed files with 400 additions and 42 deletions
--- a/tests/test_core/test_collector_contract.py
+++ b/tests/test_core/test_collector_contract.py
@@ -565,6 +565,52 @@ def _links(machineassetid, activeonly=True):
    return q.all()


+def _go_quiet(db, hostname, hours=48):
+    """Backdate a PC's last report so it reads as gone from the floor.
+
+    A PC pulled off a machine simply stops reporting, and there is no other
+    signal. Tests that want a completed handover have to age the incumbent
+    past MACHINE_CLAIM_QUIET_HOURS; without this it is a live PC and the
+    challenger is correctly held off.
+    """
+    from datetime import datetime, timedelta, timezone
+    from plugins.computers.models import Computer
+
+    comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first()
+    comp.lastreporteddate = (datetime.now(timezone.utc).replace(tzinfo=None)
+                             - timedelta(hours=hours))
+    db.session.commit()
+    return comp
+
+
+def _enable_machinelink_alerts(db):
+    """Turn the PC-to-machine alerts on.
+
+    They ship OFF: several part markers legitimately share one machine number
+    (0613, 0615, WJPRT), so on this site the alerts would fire on correct data.
+    A test that wants to see one has to say so.
+    """
+    from shopdb.core.models import Setting
+    db.session.add(Setting(key='computers_machinelink_alerts', value='true'))
+    db.session.commit()
+
+
+def _set_status(db, hostname, statusname):
+    """Put a PC's asset on a named status, the way a person would in the UI."""
+    from shopdb.core.models import AssetStatus
+    from plugins.computers.models import Computer
+
+    status = AssetStatus.query.filter_by(status=statusname).first()
+    if not status:
+        status = AssetStatus(status=statusname, isactive=True)
+        db.session.add(status)
+        db.session.flush()
+    comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first()
+    comp.asset.statusid = status.statusid
+    db.session.commit()
+    return comp
+
+
 def test_pc_reporting_an_existing_machine_number_does_not_500(
        client, db, collector_key, computer_assettype, machine_3015):
    """The machine owns assetnumber 3015. Naming the PC 3015 too hit the unique
@@ -601,6 +647,7 @@ def test_replacing_a_pc_hands_the_machine_over(
    client.post('/api/collector/computers',
                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})
+    _go_quiet(db, 'OLDPC')
    r = client.post('/api/collector/computers',
                    json={'hostname': 'NEWPC', 'machinenumber': '3015'},
                    headers={'X-API-Key': collector_key})
@@ -614,12 +661,102 @@ def test_replacing_a_pc_hands_the_machine_over(
    assert any('taken over' in w for w in r.get_json()['data']['warnings'])


+def test_a_live_pc_keeps_its_machine_against_a_new_claim(
+        client, db, collector_key, computer_assettype, machine_3015):
+    """A PC imaged for machine 3015 carries that number before it ever reaches
+    the floor. While the PC actually running the machine is still reporting,
+    the claim is not a handover and must not move the link."""
+    from shopdb.core.models import Asset
+    client.post('/api/collector/computers',
+                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+    r = client.post('/api/collector/computers',
+                    json={'hostname': 'NEWPC', 'machinenumber': '3015'},
+                    headers={'X-API-Key': collector_key})
+
+    active = _links(machine_3015.assetid)
+    assert len(active) == 1
+    assert db.session.get(Asset, active[0].sourceassetid).assetnumber == 'OLDPC'
+    assert any('still held by' in w for w in r.get_json()['data']['warnings'])
+
+    # The claim is recorded, dormant, so nothing about it is lost.
+    assert len(_links(machine_3015.assetid, activeonly=False)) == 2
+
+
+def test_a_contested_claim_does_not_flap_or_realert(
+        client, db, collector_key, computer_assettype, machine_3015,
+        monkeypatch):
+    """Both PCs keep reporting on their own schedules. The old code let them
+    trade the link on every pass and alerted each time."""
+    from shopdb.core.models import Asset
+    import shopdb.api as shopdbapi
+
+    alerts = []
+    monkeypatch.setattr(shopdbapi, 'send_alert',
+                        lambda subj, html, **k: alerts.append(subj))
+    _enable_machinelink_alerts(db)
+
+    for _ in range(3):
+        client.post('/api/collector/computers',
+                    json={'hostname': 'OLDPC', 'machinenumber': '3015'},
+                    headers={'X-API-Key': collector_key})
+        client.post('/api/collector/computers',
+                    json={'hostname': 'NEWPC', 'machinenumber': '3015'},
+                    headers={'X-API-Key': collector_key})
+
+    active = _links(machine_3015.assetid)
+    assert len(active) == 1
+    assert db.session.get(Asset, active[0].sourceassetid).assetnumber == 'OLDPC'
+    assert len(alerts) == 1, 'contested claim must alert once, not every cycle'
+
+
+def test_retiring_the_old_pc_hands_the_machine_over_at_once(
+        client, db, collector_key, computer_assettype, machine_3015):
+    """The one-step way to force a swap without waiting out the quiet window:
+    a person says the old PC is no longer in use."""
+    from shopdb.core.models import Asset
+    client.post('/api/collector/computers',
+                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+    client.post('/api/collector/computers',
+                json={'hostname': 'NEWPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+    _set_status(db, 'OLDPC', 'Retired')
+
+    client.post('/api/collector/computers',
+                json={'hostname': 'NEWPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+
+    active = _links(machine_3015.assetid)
+    assert len(active) == 1
+    assert db.session.get(Asset, active[0].sourceassetid).assetnumber == 'NEWPC'
+
+
+def test_a_pc_off_overnight_does_not_lose_its_machine(
+        client, db, collector_key, computer_assettype, machine_3015):
+    """Shorter than the quiet window is not gone. A machine must not change
+    hands because a PC was switched off for the night or the network dropped."""
+    from shopdb.core.models import Asset
+    client.post('/api/collector/computers',
+                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+    _go_quiet(db, 'OLDPC', hours=14)
+    client.post('/api/collector/computers',
+                json={'hostname': 'NEWPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+
+    active = _links(machine_3015.assetid)
+    assert len(active) == 1
+    assert db.session.get(Asset, active[0].sourceassetid).assetnumber == 'OLDPC'
+
+
 def test_the_old_link_is_archived_not_deleted(
        client, db, collector_key, computer_assettype, machine_3015):
    """History has to survive: 'which PC ran 3015 in June' stays answerable."""
    client.post('/api/collector/computers',
                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})
+    _go_quiet(db, 'OLDPC')
    client.post('/api/collector/computers',
                json={'hostname': 'NEWPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})
@@ -636,6 +773,7 @@ def test_the_old_pc_status_is_left_alone(
                headers={'X-API-Key': collector_key})
    old = Computer.query.filter(Computer.hostname.ilike('OLDPC')).first()
    before = old.asset.statusid
+    _go_quiet(db, 'OLDPC')
    client.post('/api/collector/computers',
                json={'hostname': 'NEWPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})
@@ -674,33 +812,37 @@ def test_supersede_alerts_email_and_webhook(client, db, collector_key,
    # Patch on shopdb.api, which is where the plugin imports them from - the
    # names are bound there at import time, so patching shopdb.utils.mailer
    # would silently have no effect and the real (disabled) mailer would run.
+    # send_alert is the site's configured fan-out: alert_recipients by email
+    # plus the alert webhook. Patch the fan-out, not its two legs - resolving
+    # recipients here by hand is exactly the bug this replaced.
    sent = {}
-    monkeypatch.setattr(
-        shopdbapi, 'send_email',
-        lambda to, subj, html, **k: sent.update(email=(to, subj, html)))
-    monkeypatch.setattr(shopdbapi, 'send_webhook',
-                        lambda subj, text, **k: sent.update(webhook=subj))
+    monkeypatch.setattr(shopdbapi, 'send_alert',
+                        lambda subj, html, **k: sent.update(alert=(subj, html)))
    db.session.add(Setting(key='alert_recipients', value='it@example.com'))
    db.session.add(Setting(key='site_base_url', value='https://shopdb/shopdb'))
    db.session.commit()
+    _enable_machinelink_alerts(db)

    client.post('/api/collector/computers',
                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})
+    _go_quiet(db, 'OLDPC')
    client.post('/api/collector/computers',
                json={'hostname': 'NEWPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})

-    assert 'webhook' in sent, 'webhook not fired on supersede'
-    assert 'email' in sent, 'no email sent on supersede'
-    assert sent['email'][0] == ['it@example.com']
-    assert '3015' in sent['email'][1] and 'OLDPC' in sent['email'][1]
+    assert 'alert' in sent, 'no alert raised on supersede'
+    subject, html = sent['alert']
+    assert '3015' in subject and 'OLDPC' in subject and 'NEWPC' in subject
+    # Must be the handover alert, not the contested-claim one. Both name the
+    # same three things, so the subjects alone do not tell them apart.
+    assert 'replaced' in subject, subject

    # The link must be keyed on computerid: /pcs/:id resolves through
    # GET /api/computers/<computerid>, so an assetid opens the wrong PC.
    from plugins.computers.models import Computer
    newpc = Computer.query.filter_by(hostname='NEWPC').first()
-    assert '/pcs/{}"'.format(newpc.computerid) in sent['email'][2]
+    assert '/pcs/{}"'.format(newpc.computerid) in html


 def test_alert_failure_never_breaks_the_collector(client, db, collector_key,
@@ -711,14 +853,41 @@ def test_alert_failure_never_breaks_the_collector(client, db, collector_key,

    def boom(*a, **k):
        raise RuntimeError('smtp down')
-    monkeypatch.setattr(shopdbapi, 'send_email', boom)
-    monkeypatch.setattr(shopdbapi, 'send_webhook', boom)
+    monkeypatch.setattr(shopdbapi, 'send_alert', boom)
+    _enable_machinelink_alerts(db)

    client.post('/api/collector/computers',
                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
                headers={'X-API-Key': collector_key})
+    _go_quiet(db, 'OLDPC')
    r = client.post('/api/collector/computers',
                    json={'hostname': 'NEWPC', 'machinenumber': '3015'},
                    headers={'X-API-Key': collector_key})
    assert r.status_code == 200
    assert len(_links(machine_3015.assetid)) == 1
+
+
+def test_machinelink_alerts_are_off_by_default(client, db, collector_key,
+                                               computer_assettype,
+                                               machine_3015, monkeypatch):
+    """Several part markers legitimately share a machine number (0613, 0615,
+    WJPRT), so on this site a handover alert would fire on correct data. The
+    link, the archive and the collector warning all still happen - only the
+    sending is gated, and it ships off."""
+    import shopdb.api as shopdbapi
+
+    alerts = []
+    monkeypatch.setattr(shopdbapi, 'send_alert',
+                        lambda subj, html, **k: alerts.append(subj))
+
+    client.post('/api/collector/computers',
+                json={'hostname': 'OLDPC', 'machinenumber': '3015'},
+                headers={'X-API-Key': collector_key})
+    _go_quiet(db, 'OLDPC')
+    r = client.post('/api/collector/computers',
+                    json={'hostname': 'NEWPC', 'machinenumber': '3015'},
+                    headers={'X-API-Key': collector_key})
+
+    assert alerts == []
+    assert len(_links(machine_3015.assetid)) == 1
+    assert any('taken over' in w for w in r.get_json()['data']['warnings'])