shopdb-flask/tests/test_plugins/test_endpoints_reconciler.py

"""Warn-only reconciler: drift computation and its exposure on the panels.

The reconciler runs server-side on agent check-in. For every monitor a site
policy attaches to an endpoint's PC type that is EXPECTED to be running, it flags
DRIFT when the latest snapshot does not report it running. It never starts or
stops anything - drift is surfaced for an operator to act on manually.

These tests cover the pure computer (compute_drift), the check-in write path
(apply_collector_payload persisting driftjson), the panel exposure, and the
observe-only seam (monitor_expects_running) that keeps a non-keepalive monitor
out of the drift list.
"""

from shopdb.core.models import Asset, AssetType, Setting


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


def _computer(db, hostname, computertype):
    """Create a computer Asset + Computer of the given PC type; return the pair."""
    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=computertype).first()
    if not ctype:
        ctype = ComputerType(computertype=computertype)
        db.session.add(ctype)
        db.session.flush()

    asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
                  isactive=True)
    db.session.add(asset)
    db.session.flush()
    comp = Computer(assetid=asset.assetid, hostname=hostname,
                    computertypeid=ctype.computertypeid)
    db.session.add(comp)
    db.session.flush()
    return asset, ctype


def _attach_monitor_to_pctype(db, monitor, computertypeid):
    """Attach a monitor to a PC type via an active Business/Configuration Policy."""
    from plugins.computers.models import (
        ConfigurationPolicy, ConfigurationPolicyAction,
        BusinessPolicy, BusinessPolicyItem,
    )

    if not Setting.query.filter_by(
            key='computers_configpolicies_enabled').first():
        db.session.add(
            Setting(key='computers_configpolicies_enabled', value='true'))

    policy = ConfigurationPolicy(
        name='Coverage {}'.format(monitor.endpointmonitorid),
        scopejson={'type': 'pctype', 'id': computertypeid},
    )
    db.session.add(policy)
    db.session.flush()
    db.session.add(ConfigurationPolicyAction(
        configurationpolicyid=policy.configurationpolicyid,
        action='monitor:{}'.format(monitor.endpointmonitorid),
        mode='gate',
    ))
    business = BusinessPolicy(name='BP {}'.format(monitor.endpointmonitorid),
                             priority=100)
    db.session.add(business)
    db.session.flush()
    db.session.add(BusinessPolicyItem(
        businesspolicyid=business.businesspolicyid,
        configurationpolicyid=policy.configurationpolicyid,
        sortorder=0,
    ))
    db.session.commit()


def _process_monitor(db, displayname, processname):
    from plugins.endpoints.models import EndpointMonitor

    monitor = EndpointMonitor(
        kind='process', displayname=displayname, processname=processname,
        source='local', isactive=True,
    )
    db.session.add(monitor)
    db.session.flush()
    return monitor


# ---------------------------------------------------------------------------
# Pure computer
# ---------------------------------------------------------------------------

def test_compute_drift_flags_attached_monitor_not_reported_running(
        client, db, app):
    """An expected-running monitor absent from the snapshot is drift."""
    from plugins.endpoints.services.reconciler import compute_drift

    asset, ctype = _computer(db, 'DRIFT01', 'CMM PC')
    monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
    _attach_monitor_to_pctype(db, monitor, ctype.computertypeid)

    drift = compute_drift(asset.assetid, [])
    assert len(drift) == 1
    assert drift[0]['name'] == 'dncMain.exe'
    assert drift[0]['kind'] == 'process'
    assert drift[0]['expected'] == 'running'
    assert drift[0]['observed'] == 'notrunning'


def test_compute_drift_empty_when_monitor_reported_running(client, db, app):
    """A monitor the snapshot reports running is not drift."""
    from plugins.endpoints.services.reconciler import compute_drift

    asset, ctype = _computer(db, 'DRIFT02', 'CMM PC')
    monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
    _attach_monitor_to_pctype(db, monitor, ctype.computertypeid)

    reported = [{'name': 'dncMain.exe', 'kind': 'process', 'state': 'running'}]
    assert compute_drift(asset.assetid, reported) == []


def test_compute_drift_empty_when_no_monitor_attached(client, db, app):
    """No attached monitors means nothing to reconcile."""
    from plugins.endpoints.services.reconciler import compute_drift

    asset, _ = _computer(db, 'DRIFT03', 'CMM PC')
    assert compute_drift(asset.assetid, []) == []


def test_monitor_expects_running_is_the_observe_only_seam():
    """The 'should be up' decision lives in one function; inactive is excluded."""
    from plugins.endpoints.services.reconciler import monitor_expects_running

    class _M:
        isactive = True

    active = _M()
    inactive = _M()
    inactive.isactive = False
    assert monitor_expects_running(active) is True
    assert monitor_expects_running(inactive) is False


# ---------------------------------------------------------------------------
# Check-in write path
# ---------------------------------------------------------------------------

def test_checkin_persists_drift_on_endpoint_state(client, db, app):
    """apply_collector_payload caches drift on the state in the same commit."""
    from plugins.endpoints.plugin import EndpointsPlugin
    from plugins.endpoints.models import EndpointState

    asset, ctype = _computer(db, 'CHKIN01', 'CMM PC')
    monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
    _attach_monitor_to_pctype(db, monitor, ctype.computertypeid)

    plugin = EndpointsPlugin()
    plugin.apply_collector_payload({'hostname': 'CHKIN01', 'services': []})
    db.session.commit()

    state = EndpointState.query.filter_by(assetid=asset.assetid).first()
    assert state.driftjson is not None
    import json
    drift = json.loads(state.driftjson)
    assert [d['name'] for d in drift] == ['dncMain.exe']


def test_checkin_clears_drift_when_process_returns(client, db, app):
    """A later report that shows the process running clears the cached drift."""
    from plugins.endpoints.plugin import EndpointsPlugin
    from plugins.endpoints.models import EndpointState
    import json

    asset, ctype = _computer(db, 'CHKIN02', 'CMM PC')
    monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
    _attach_monitor_to_pctype(db, monitor, ctype.computertypeid)

    plugin = EndpointsPlugin()
    plugin.apply_collector_payload({'hostname': 'CHKIN02', 'services': []})
    db.session.commit()

    plugin.apply_collector_payload({
        'hostname': 'CHKIN02',
        'services': [{'name': 'dncMain.exe', 'kind': 'process',
                      'state': 'running'}],
    })
    db.session.commit()

    state = EndpointState.query.filter_by(assetid=asset.assetid).first()
    assert json.loads(state.driftjson) == []


# ---------------------------------------------------------------------------
# Panel exposure
# ---------------------------------------------------------------------------

def test_status_panel_surfaces_drift_warning(client, db, app, auth_headers):
    """The status panel adds a Needs attention field when drift is cached."""
    from plugins.endpoints.plugin import EndpointsPlugin

    asset, ctype = _computer(db, 'PANEL01', 'CMM PC')
    monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
    _attach_monitor_to_pctype(db, monitor, ctype.computertypeid)

    plugin = EndpointsPlugin()
    plugin.apply_collector_payload({'hostname': 'PANEL01', 'services': []})
    db.session.commit()

    response = client.get(STATUS_URL.format(asset.assetid), headers=auth_headers)
    assert response.status_code == 200
    labels = [f['label'] for f in response.get_json()['data']['fields']]
    assert 'Needs attention' in labels


def test_services_panel_marks_drifted_row(client, db, app, auth_headers):
    """The synthetic Unknown row for an expected-running monitor is drift-flagged."""
    from plugins.endpoints.plugin import EndpointsPlugin

    asset, ctype = _computer(db, 'PANEL02', 'CMM PC')
    monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
    _attach_monitor_to_pctype(db, monitor, ctype.computertypeid)

    plugin = EndpointsPlugin()
    plugin.apply_collector_payload({'hostname': 'PANEL02', 'services': []})
    db.session.commit()

    response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers)
    assert response.status_code == 200
    rows = response.get_json()['data']
    edenc = next((r for r in rows if r['name'] == 'eDNC'), None)
    assert edenc is not None
    assert edenc.get('drift') is True