211 lines
8.0 KiB
Python
211 lines
8.0 KiB
Python
"""Dashboard card: PCs that were reporting and have stopped.
|
|
|
|
Deliberately narrow. A PC that never reported at all is usually a record made by
|
|
hand or imported, not a bay that broke, and a PC that is not In Use is silent on
|
|
purpose - that is the status doing its job. Both would bury the real signal: a
|
|
machine that was working, is not now, and nobody has marked as anything else.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from shopdb.core.models import Asset, AssetStatus, AssetType, Setting
|
|
|
|
URL = '/api/computers/dashboard/quiet'
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
def _inuse(db):
|
|
"""The card joins on the status NAME, so a bare statusid will not do."""
|
|
status = AssetStatus.query.filter_by(status='In Use').first()
|
|
if not status:
|
|
status = AssetStatus(status='In Use', isactive=True)
|
|
db.session.add(status)
|
|
db.session.flush()
|
|
return status
|
|
|
|
|
|
def _pc(db, hostname, lastreported='now', isactive=True):
|
|
from plugins.computers.models import Computer
|
|
|
|
atype = AssetType.query.filter_by(assettype='computer').first()
|
|
if not atype:
|
|
atype = AssetType(assettype='computer')
|
|
db.session.add(atype)
|
|
db.session.flush()
|
|
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
|
statusid=_inuse(db).statusid, isactive=isactive)
|
|
db.session.add(asset)
|
|
db.session.flush()
|
|
comp = Computer(assetid=asset.assetid, hostname=hostname)
|
|
if lastreported == 'now':
|
|
comp.lastreporteddate = _now()
|
|
elif lastreported is None:
|
|
comp.lastreporteddate = None
|
|
else:
|
|
comp.lastreporteddate = _now() - timedelta(hours=lastreported)
|
|
db.session.add(comp)
|
|
db.session.commit()
|
|
return comp
|
|
|
|
|
|
def test_a_reporting_pc_is_not_listed(client, db, auth_headers):
|
|
_pc(db, 'BUSYPC')
|
|
assert client.get(URL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
|
|
def test_a_pc_gone_quiet_is_listed_with_how_long(client, db, auth_headers):
|
|
_pc(db, 'QUIETPC', lastreported=72)
|
|
[row] = client.get(URL, headers=auth_headers).get_json()['data']
|
|
assert row['hostname'] == 'QUIETPC'
|
|
assert row['quietdays'] == 3
|
|
|
|
|
|
def test_a_pc_that_never_reported_is_not_listed(client, db, auth_headers):
|
|
"""Usually a hand-made or imported record rather than a bay that broke.
|
|
Listing them buries the PCs that actually changed state."""
|
|
_pc(db, 'NEVERPC', lastreported=None)
|
|
assert client.get(URL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
|
|
def test_a_pc_not_in_use_is_not_listed(client, db, auth_headers):
|
|
"""A PC in Repair or Retired is silent on purpose - that is the status
|
|
doing its job, not a fault to chase."""
|
|
from shopdb.core.models import AssetStatus
|
|
repair = AssetStatus.query.filter_by(status='In Repair').first()
|
|
if not repair:
|
|
repair = AssetStatus(status='In Repair', isactive=True)
|
|
db.session.add(repair)
|
|
db.session.flush()
|
|
comp = _pc(db, 'SHELVED', lastreported=200)
|
|
comp.asset.statusid = repair.statusid
|
|
db.session.commit()
|
|
assert client.get(URL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
|
|
def test_the_longest_silence_comes_first(client, db, auth_headers):
|
|
_pc(db, 'DAY2', lastreported=48)
|
|
_pc(db, 'DAY9', lastreported=216)
|
|
rows = client.get(URL, headers=auth_headers).get_json()['data']
|
|
assert [r['hostname'] for r in rows] == ['DAY9', 'DAY2']
|
|
|
|
|
|
def test_the_window_is_a_setting_not_a_hardcode(client, db, auth_headers):
|
|
"""Every site will disagree with 24 hours (ADR-015)."""
|
|
_pc(db, 'OVERNIGHT', lastreported=10)
|
|
assert client.get(URL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
db.session.add(Setting(key='computers_quietreporthours', value='4'))
|
|
db.session.commit()
|
|
rows = client.get(URL, headers=auth_headers).get_json()['data']
|
|
assert [r['hostname'] for r in rows] == ['OVERNIGHT']
|
|
|
|
|
|
def test_a_bad_setting_falls_back_rather_than_failing_the_card(client, db,
|
|
auth_headers):
|
|
db.session.add(Setting(key='computers_quietreporthours', value='soon'))
|
|
db.session.commit()
|
|
_pc(db, 'QUIETPC', lastreported=72)
|
|
assert client.get(URL, headers=auth_headers).status_code == 200
|
|
|
|
|
|
def test_a_retired_pc_is_not_nagged_about(client, db, auth_headers):
|
|
"""A decommissioned PC is silent on purpose. Listing it would train people
|
|
to ignore the card, which is the failure mode this whole board avoids."""
|
|
_pc(db, 'RETIRED', lastreported=None, isactive=False)
|
|
assert client.get(URL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
|
|
def test_the_card_is_permission_gated(client, db, member_headers):
|
|
assert client.get(URL, headers=member_headers).status_code == 403
|
|
|
|
|
|
def test_the_widget_uses_the_new_contract(app):
|
|
from plugins.computers.plugin import ComputersPlugin
|
|
|
|
widget = ComputersPlugin().get_dashboard_widgets()[0]
|
|
assert 'component' not in widget
|
|
assert widget['render'] == 'exceptions'
|
|
assert widget['permission'] == 'computers.view'
|
|
assert widget['empty'] == 'hide'
|
|
|
|
|
|
# =============================================================================
|
|
# Machine numbers claimed by two PCs - a fault, unless the devices are modelled
|
|
# =============================================================================
|
|
|
|
SHAREDURL = '/api/computers/dashboard/sharedmachines'
|
|
|
|
|
|
def _machine(db, assetnumber):
|
|
mt = AssetType.query.filter_by(assettype='machine').first()
|
|
if not mt:
|
|
mt = AssetType(assettype='machine')
|
|
db.session.add(mt)
|
|
db.session.flush()
|
|
asset = Asset(assetnumber=assetnumber, assettypeid=mt.assettypeid,
|
|
statusid=1)
|
|
db.session.add(asset)
|
|
db.session.commit()
|
|
return asset
|
|
|
|
|
|
def _controls(db, pchostname, machineasset, label='collector:machine'):
|
|
from shopdb.core.models import AssetRelationship, RelationshipType
|
|
controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
|
|
if not controls:
|
|
controls = RelationshipType(relationshiptype='controls', isdirectional=True)
|
|
db.session.add(controls)
|
|
db.session.flush()
|
|
comp = _pc(db, pchostname)
|
|
db.session.add(AssetRelationship(
|
|
sourceassetid=comp.assetid, targetassetid=machineasset.assetid,
|
|
relationshiptypeid=controls.relationshiptypeid, label=label,
|
|
isactive=True))
|
|
db.session.commit()
|
|
return comp
|
|
|
|
|
|
def test_one_pc_per_machine_is_not_a_finding(client, db, auth_headers):
|
|
machine = _machine(db, '3015')
|
|
_controls(db, 'ONLYPC', machine)
|
|
assert client.get(SHAREDURL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
|
|
def test_two_pcs_on_one_number_is_reported(client, db, auth_headers):
|
|
machine = _machine(db, '2026')
|
|
_controls(db, 'PCONE', machine)
|
|
_controls(db, 'PCTWO', machine)
|
|
[row] = client.get(SHAREDURL, headers=auth_headers).get_json()['data']
|
|
assert row['machinenumber'] == '2026'
|
|
assert row['pccount'] == 2
|
|
assert 'PCONE' in row['pcs'] and 'PCTWO' in row['pcs']
|
|
|
|
|
|
def test_a_modelled_shared_number_is_not_a_finding(client, db, auth_headers):
|
|
"""Part markers legitimately share an operation number. Modelled correctly,
|
|
each device is its own asset filed partof the operation - so the operation
|
|
has child assets, and that is what tells a real fault from correct data."""
|
|
from shopdb.core.models import AssetRelationship, RelationshipType
|
|
operation = _machine(db, '0615')
|
|
_controls(db, 'MARKERPCA', operation)
|
|
_controls(db, 'MARKERPCB', operation)
|
|
|
|
partof = RelationshipType(relationshiptype='partof', isdirectional=True)
|
|
db.session.add(partof)
|
|
db.session.flush()
|
|
marker = _machine(db, 'MARKERPCA-PARTMARKER')
|
|
db.session.add(AssetRelationship(
|
|
sourceassetid=marker.assetid, targetassetid=operation.assetid,
|
|
relationshiptypeid=partof.relationshiptypeid,
|
|
label='collector:partmarker', isactive=True))
|
|
db.session.commit()
|
|
|
|
assert client.get(SHAREDURL, headers=auth_headers).get_json()['data'] == []
|
|
|
|
|
|
def test_shared_machines_is_permission_gated(client, db, member_headers):
|
|
assert client.get(SHAREDURL, headers=member_headers).status_code == 403
|