199 lines
7.6 KiB
Python
199 lines
7.6 KiB
Python
"""Dashboard card: PCs that stopped reporting, and PCs that never started.
|
|
|
|
Two populations, deliberately not merged. A PC that reported and went quiet is
|
|
probably off, moved or broken. A PC that has NEVER reported is worse - not
|
|
enrolled, or enrolled against the wrong pc-type, so nothing enforces anything on
|
|
it and no backup of it exists. That one hides indefinitely because nothing about
|
|
it fails loudly.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from shopdb.core.models import Asset, AssetType, Setting
|
|
|
|
URL = '/api/computers/dashboard/quiet'
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
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=1, 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['state'] == 'stopped reporting'
|
|
assert row['quietdays'] == 3
|
|
|
|
|
|
def test_a_pc_that_never_reported_is_called_out_separately(client, db,
|
|
auth_headers):
|
|
"""Never-reported is a different fault from gone-quiet: nothing is
|
|
enforcing anything on that PC and no backup of it exists."""
|
|
_pc(db, 'NEVERPC', lastreported=None)
|
|
[row] = client.get(URL, headers=auth_headers).get_json()['data']
|
|
assert row['state'] == 'never reported'
|
|
assert row['lastreported'] is None
|
|
|
|
|
|
def test_never_reported_sorts_above_the_merely_quiet(client, db, auth_headers):
|
|
"""The order someone should work down the list."""
|
|
_pc(db, 'QUIETPC', lastreported=48)
|
|
_pc(db, 'NEVERPC', lastreported=None)
|
|
rows = client.get(URL, headers=auth_headers).get_json()['data']
|
|
assert [r['hostname'] for r in rows] == ['NEVERPC', 'QUIETPC']
|
|
|
|
|
|
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
|