ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Collector: machine number identifies the machine, not the PC; ntlars empty-value fix

Browse Source
This commit is contained in:
Cameron Proudlock
2026-08-10 14:26:19 -04:00
parent 41a4a96f1e
commit 502728c9f6
5 changed files with 431 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
tests/test_core/test_collector_contract.py
View File

@@ -145,7 +145,12 @@ def test_complete_asset_payload_maps_enrollment_data(client, db, collector_key,
with client.application.app_context():
comp = Computer.query.filter(Computer.hostname.ilike('WJSF1234')).first()
assert comp is not None
assert comp.asset.assetnumber == '0615' # machinenumber -> assetnumber
# The PC keeps its OWN assetnumber (hostname). machinenumber names the
# MACHINE, and machines are assets too (ADR-001), so assigning it here
# hit the unique index on assets.assetnumber and returned 500 to the bay
# on every retry. The machine number now builds a controls link instead,
# which is what the reported number was actually for.
assert comp.asset.assetnumber == 'WJSF1234'
assert comp.asset.serialnumber == 'SN-ENROLL'
assert comp.computertype.computertype == 'Shopfloor PC' # pctype mapped
assert comp.vendor.vendor == 'Dell' # created
@@ -528,3 +533,192 @@ def test_unknown_protocol_warns_and_does_not_create_one(client, db, collector_ke
assert any('TeamViewer' in w for w in body['warnings'])
assert AccessProtocol.query.filter(
AccessProtocol.name.ilike('TeamViewer')).first() is None
# =============================================================================
# PC -> machine link, and the PC-swap case
# =============================================================================
@pytest.fixture
def machine_3015(db):
"""A machine asset that already owns the number a PC will report."""
from shopdb.core.models import AssetType, Asset, RelationshipType
mt = AssetType(assettype='machine', pluginname='machines',
tablename='machines', description='Machines')
db.session.add(mt)
db.session.add(RelationshipType(relationshiptype='controls',
isdirectional=True))
db.session.flush()
asset = Asset(assetnumber='3015', name='Machine 3015',
assettypeid=mt.assettypeid, statusid=1)
db.session.add(asset)
db.session.commit()
return asset
def _links(machineassetid, activeonly=True):
from shopdb.core.models import AssetRelationship
q = AssetRelationship.query.filter_by(targetassetid=machineassetid)
if activeonly:
q = q.filter_by(isactive=True)
return q.all()
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
index and returned 500 to the bay on every retry, forever."""
r = client.post('/api/collector/computers',
json={'hostname': 'FFBWTH63', 'machinenumber': '3015'},
headers={'X-API-Key': collector_key})
assert r.status_code == 200, r.get_json()
def test_pc_gets_its_own_assetnumber_not_the_machines(
client, db, collector_key, computer_assettype, machine_3015):
from plugins.computers.models import Computer
client.post('/api/collector/computers',
json={'hostname': 'FFBWTH63', 'machinenumber': '3015'},
headers={'X-API-Key': collector_key})
comp = Computer.query.filter(Computer.hostname.ilike('FFBWTH63')).first()
assert comp.asset.assetnumber == 'FFBWTH63'
def test_reported_machine_number_creates_the_link(
client, db, collector_key, computer_assettype, machine_3015):
client.post('/api/collector/computers',
json={'hostname': 'FFBWTH63', 'machinenumber': '3015'},
headers={'X-API-Key': collector_key})
assert len(_links(machine_3015.assetid)) == 1
def test_replacing_a_pc_hands_the_machine_over(
client, db, collector_key, computer_assettype, machine_3015):
"""The point of the whole thing: the new PC takes the bay and the retired
one stops being shown as its controller."""
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
owner = db.session.get(Asset, active[0].sourceassetid)
assert owner.assetnumber == 'NEWPC'
# and it says so, so a person can go and look at the old one
assert any('taken over' in w for w in r.get_json()['data']['warnings'])
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})
client.post('/api/collector/computers',
json={'hostname': 'NEWPC', 'machinenumber': '3015'},
headers={'X-API-Key': collector_key})
assert len(_links(machine_3015.assetid, activeonly=False)) == 2
def test_the_old_pc_status_is_left_alone(
client, db, collector_key, computer_assettype, machine_3015):
"""The collector cannot tell if it was shelved, broken or re-imaged, so it
must not guess - and must not overwrite what a person set."""
from plugins.computers.models import Computer
client.post('/api/collector/computers',
json={'hostname': 'OLDPC', 'machinenumber': '3015'},
headers={'X-API-Key': collector_key})
old = Computer.query.filter(Computer.hostname.ilike('OLDPC')).first()
before = old.asset.statusid
client.post('/api/collector/computers',
json={'hostname': 'NEWPC', 'machinenumber': '3015'},
headers={'X-API-Key': collector_key})
db.session.refresh(old.asset)
assert old.asset.statusid == before
def test_unknown_machine_number_warns_and_invents_nothing(
client, db, collector_key, computer_assettype, machine_3015):
from shopdb.core.models import Asset
r = client.post('/api/collector/computers',
json={'hostname': 'FFBWTH63', 'machinenumber': '9911'},
headers={'X-API-Key': collector_key})
assert any('9911' in w for w in r.get_json()['data']['warnings'])
assert Asset.query.filter(Asset.assetnumber == '9911').first() is None
def test_the_placeholder_machine_number_links_nothing(
client, db, collector_key, computer_assettype, machine_3015):
"""9999 is the imaging-time placeholder, not a real bay."""
r = client.post('/api/collector/computers',
json={'hostname': 'FFBWTH63', 'machinenumber': '9999'},
headers={'X-API-Key': collector_key})
assert r.status_code == 200
assert _links(machine_3015.assetid) == []
def test_supersede_alerts_email_and_webhook(client, db, collector_key,
computer_assettype, machine_3015,
monkeypatch):
"""A person has to find out. Email goes to the site alert_recipients and the
webhook fires, the same path the toner alerts use."""
from shopdb.core.models import Setting
import shopdb.api as shopdbapi
# 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.
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))
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()
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})
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]
# 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]
def test_alert_failure_never_breaks_the_collector(client, db, collector_key,
computer_assettype,
machine_3015, monkeypatch):
"""A broken mail server must not stop a bay reporting its inventory."""
import shopdb.api as shopdbapi
def boom(*a, **k):
raise RuntimeError('smtp down')
monkeypatch.setattr(shopdbapi, 'send_email', boom)
monkeypatch.setattr(shopdbapi, 'send_webhook', boom)
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})
assert r.status_code == 200
assert len(_links(machine_3015.assetid)) == 1

47
tests/test_plugins/test_backups.py
View File

@@ -710,3 +710,50 @@ def test_panel_label_is_rendered_in_the_site_zone(bk_app, bk_plugin):
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
# 16:30 UTC on 2026-08-07 is 12:30 EDT.
assert '12:30' in _label(revision)
# =============================================================================
# REGEDIT4 / part marker exports
# =============================================================================
REGEDIT4REG = (
'REGEDIT4\r\n\r\n'
r'[HKEY_LOCAL_MACHINE\SOFTWARE\GE Aircraft Engines\DNC\General]' '\r\n'
'"MachineNo"="WJPRT"\r\n'
'"Cnc"="MARKER"\r\n\r\n'
r'[HKEY_LOCAL_MACHINE\SOFTWARE\GE Aircraft Engines\DNC\Mark]' '\r\n'
'"Port Id"="COM4"\r\n'
'"Baud"="9600"\r\n'
'"KRelay1"=\r\n'
)
def test_regedit4_header_is_accepted():
"""The part marker's per-machine file is REGEDIT4, not 'Windows Registry
Editor Version 5.00'."""
projection = ntlars.parse(REGEDIT4REG.encode('utf-8'))
assert registry.NtlarsKind.embeddedmachineno(projection) == 'WJPRT'
def test_value_with_an_empty_right_hand_side_does_not_fail_the_file():
"""A real export contains "KRelay1=" with nothing after the sign. Raising on
it would mean that machine could never be backed up at all."""
projection = ntlars.parse(REGEDIT4REG.encode('utf-8'))
mark = next(k for k in projection['keys'] if k['path'] == 'Mark')
assert mark['values']['KRelay1'] == {'type': 'REG_SZ', 'data': ''}
def test_part_marker_serial_port_survives_the_roundtrip():
"""The COM port the marker hangs off is the point of keeping this config."""
first = ntlars.parse(REGEDIT4REG.encode('utf-8'))
second = ntlars.parse(ntlars.render(first, dialect='wow6432node'))
mark = next(k for k in second['keys'] if k['path'] == 'Mark')
assert mark['values']['Port Id']['data'] == 'COM4'
def test_a_marker_export_shows_the_mark_tab():
card = dncinfo.build(ntlars.parse(REGEDIT4REG.encode('utf-8')), assetid=0)
labels = [s['label'] for s in card['sections']]
assert any('MARK' in label for label in labels)
mark = next(s for s in card['sections'] if 'MARK' in s['label'])
assert any(f['label'] == 'Port Id' and f['value'] == 'COM4' for f in mark['fields'])