shopdb-flask/tests/test_plugins/test_endpoints_dellupdates.py

"""Pending Dell Command | Update reporting: ingest and status-card exposure.

The agent's shopdb-dellupdates utility reads the log the \\ShopDB-DellCommandScan
scheduled task writes and reports a short summary on the normal telemetry
check-in. These tests cover the ingest write path and the three-state rendering
on the status panel, which is the part most likely to be broken by a well-meaning
simplification: "never scanned" must NOT render as "No".
"""

import json

from shopdb.core.models import Asset, AssetType


STATUS_URL = '/api/endpoints/asset/{}/status'


def _computer(db, hostname):
    """Create a computer Asset + Computer row; return the asset."""
    from plugins.computers.models import Computer, ComputerType

    atype = AssetType.query.filter_by(assettype='computer').first()
    if not atype:
        atype = AssetType(assettype='computer')
        db.session.add(atype)
        db.session.flush()

    ctype = ComputerType.query.filter_by(computertype='CMM PC').first()
    if not ctype:
        ctype = ComputerType(computertype='CMM PC')
        db.session.add(ctype)
        db.session.flush()

    asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
                  isactive=True)
    db.session.add(asset)
    db.session.flush()

    db.session.add(Computer(assetid=asset.assetid, hostname=hostname,
                            computertypeid=ctype.computertypeid))
    db.session.flush()
    return asset


def _checkin(db, hostname, payload):
    """Run a check-in for hostname with the extra payload keys merged in."""
    from plugins.endpoints.plugin import EndpointsPlugin

    body = {'hostname': hostname}
    body.update(payload)
    EndpointsPlugin().apply_collector_payload(body)
    db.session.commit()


def _status_fields(client, asset, auth_headers):
    response = client.get(STATUS_URL.format(asset.assetid), headers=auth_headers)
    assert response.status_code == 200
    return {f['label']: f['value']
            for f in response.get_json()['data']['fields']}


def test_checkin_persists_the_dell_summary(client, db, app, auth_headers):
    """A reported dellupdates payload is stored on the endpoint state."""
    from plugins.endpoints.models.endpoint import EndpointState

    _computer(db, 'DELL01')
    _checkin(db, 'DELL01', {'dellupdates': {
        'status': 'ok', 'count': 1, 'truncated': False,
        'updates': [{'name': 'Dell BIOS', 'severity': 'Urgent',
                     'version': '1.2.3'}],
    }})

    state = EndpointState.query.filter_by(
        assetid=Asset.query.filter_by(assetnumber='DELL01').first().assetid
    ).first()
    stored = json.loads(state.dellupdatesjson)
    assert stored['count'] == 1
    assert stored['updates'][0]['name'] == 'Dell BIOS'


def test_pending_updates_show_yes_with_a_count(client, db, app, auth_headers):
    """A scan with pending items renders Yes plus how many."""
    asset = _computer(db, 'DELL02')
    _checkin(db, 'DELL02', {'dellupdates': {
        'status': 'ok', 'count': 3, 'truncated': False, 'updates': [],
    }})

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'Yes (3)'


def test_a_clean_scan_shows_no(client, db, app, auth_headers):
    """A scan that found nothing renders No, not a hidden row."""
    asset = _computer(db, 'DELL03')
    _checkin(db, 'DELL03', {'dellupdates': {
        'status': 'ok', 'count': 0, 'truncated': False, 'updates': [],
    }})

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'No'


def test_a_never_scanned_endpoint_says_so(client, db, app, auth_headers):
    """Never scanned renders as its own visible state, not as No and not hidden.

    Two failure modes are being guarded at once: showing 'No' would claim the
    machine is current when nobody ever checked, and hiding the row entirely
    reads to an operator as a broken panel.
    """
    asset = _computer(db, 'DELL04')
    _checkin(db, 'DELL04', {})

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'Never scanned'


def test_an_unknown_status_says_never_scanned(client, db, app, auth_headers):
    """The utility's own 'never scanned' answer renders as never scanned."""
    asset = _computer(db, 'DELL05')
    _checkin(db, 'DELL05', {'dellupdates': {
        'status': 'unknown', 'count': 0, 'truncated': False, 'updates': [],
    }})

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'Never scanned'


def test_a_failed_scan_read_is_distinct_from_never_scanned(client, db, app,
                                                           auth_headers):
    """An unreadable scan log is its own state, not silence and not clean."""
    asset = _computer(db, 'DELL06')
    _checkin(db, 'DELL06', {'dellupdates': {
        'status': 'error', 'count': 0, 'truncated': False, 'updates': [],
    }})

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'Unknown (scan unreadable)'


def test_a_malformed_blob_does_not_break_the_panel(client, db, app, auth_headers):
    """A corrupt cached payload degrades to the unknown state, not a 500."""
    from plugins.endpoints.models.endpoint import EndpointState

    asset = _computer(db, 'DELL07')
    _checkin(db, 'DELL07', {})

    state = EndpointState.query.filter_by(assetid=asset.assetid).first()
    state.dellupdatesjson = 'not json at all'
    db.session.commit()

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'Never scanned'


def test_an_omitted_key_preserves_the_last_result(client, db, app, auth_headers):
    """A check-in without the key does not erase a previously reported scan.

    The agent omits dellupdates entirely when the collector utility is absent,
    so overwriting to null would discard a still-valid result.
    """
    asset = _computer(db, 'DELL08')
    _checkin(db, 'DELL08', {'dellupdates': {
        'status': 'ok', 'count': 2, 'truncated': False, 'updates': [],
    }})
    _checkin(db, 'DELL08', {})

    fields = _status_fields(client, asset, auth_headers)
    assert fields['Pending Dell updates'] == 'Yes (2)'