Endpoint telemetry, PC Detail rework and a checked backup/restore path

tests/test_plugins/test_endpoints_agentupload.py

@@ -0,0 +1,90 @@
+"""Agent binary upload version gating: PE-fallback and precedence.
+
+The upload route no longer rejects the version at raw input. A typed value is
+only a hint; the dotted-numeric FILE version in the binary's PE resource is the
+authority. The route falls back to it when the typed value is missing or not
+dotted-numeric, and only refuses when neither yields a usable version.
+
+Building a real PE with a RT_VERSION resource inline is impractical (see the
+agentstore unit tests), so these API tests monkeypatch read_pe_file_version to
+stand in for the parser. PE-magic bytes keep looks_like_pe happy so the request
+reaches the version gate.
+"""
+
+import io
+
+from plugins.endpoints.api import agentstore
+from plugins.endpoints.api import agentupload
+from shopdb.config import Config
+
+UPLOAD_URL = '/api/endpoints/siteagent/binaries/{}'
+
+_PE_BYTES = b'MZ' + b'\x00' * 128
+
+
+def _upload(client, headers, monkeypatch, detected, typed=None):
+    monkeypatch.setattr(agentstore, 'read_pe_file_version',
+                        lambda data: detected)
+    form = {'file': (io.BytesIO(_PE_BYTES), 'shopdbagent.exe')}
+    if typed is not None:
+        form['version'] = typed
+    return client.post(
+        UPLOAD_URL.format('agent'),
+        data=form,
+        content_type='multipart/form-data',
+        headers=headers)
+
+
+def test_upload_falls_back_to_pe_version_when_typed_is_missing(
+        client, db, app, auth_headers, monkeypatch):
+    response = _upload(client, auth_headers, monkeypatch, '1.5.0.0')
+    assert response.status_code == 200, response.get_json()
+    payload = response.get_json()['data']
+    assert payload['current'] == '1.5.0.0'
+    assert any(v['version'] == '1.5.0.0' for v in payload['versions'])
+
+
+def test_upload_falls_back_when_typed_is_not_dotted_numeric(
+        client, db, app, auth_headers, monkeypatch):
+    response = _upload(client, auth_headers, monkeypatch, '1.5.0.0',
+                       typed='1.5.0+1a1bf0640f956634dada1b34840')
+    assert response.status_code == 200, response.get_json()
+    assert response.get_json()['data']['current'] == '1.5.0.0'
+
+
+def test_upload_prefers_valid_typed_version_over_pe(
+        client, db, app, auth_headers, monkeypatch):
+    response = _upload(client, auth_headers, monkeypatch, '9.9.9.9',
+                       typed='1.6.3')
+    assert response.status_code == 200, response.get_json()
+    assert response.get_json()['data']['current'] == '1.6.3'
+
+
+def test_upload_rejects_when_neither_typed_nor_pe_is_usable(
+        client, db, app, auth_headers, monkeypatch):
+    response = _upload(client, auth_headers, monkeypatch, None)
+    assert response.status_code == 400
+    assert 'dotted-numeric' in response.get_json()['data']['error']['message']
+
+
+def test_upload_cap_admits_a_self_contained_single_file_publish():
+    """The tray publishes ~69 MB; the cap must clear a real binary.
+
+    This is the bug that presented as "upload does not detect version": the
+    size check returns before the PE is parsed, so an oversized file looks
+    like a parser failure to the operator. Pinned at 100 MB rather than the
+    observed 69 MB so runtime growth does not silently re-break it.
+    """
+    assert agentupload._MAX_UPLOAD_BYTES >= 100 * 1024 * 1024
+
+
+def test_request_cap_stays_above_the_agent_upload_cap():
+    """MAX_CONTENT_LENGTH bounds the whole multipart envelope, not the file.
+
+    Setting the two equal rejects a file sitting exactly at the agent cap,
+    because the boundaries and form fields push the request over. They must
+    move together, with real headroom between them.
+    """
+    envelope = Config.MAX_CONTENT_LENGTH
+    assert envelope > agentupload._MAX_UPLOAD_BYTES
+    assert envelope - agentupload._MAX_UPLOAD_BYTES >= 1024 * 1024

tests/test_plugins/test_endpoints_dellupdates.py

@@ -0,0 +1,168 @@
+"""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)'

tests/test_plugins/test_endpoints_disk_health.py

@@ -0,0 +1,108 @@
+"""Disk health rows: a dying drive must never render as a healthy one.
+
+The agent reports drive health from two sources because neither covers the
+fleet. MSFT_PhysicalDisk is the only one that reports NVMe - verified on a
+two-NVMe box where the classic ATA SMART class returned zero instances while
+both drives were present and healthy. Win32_DiskDrive covers ATA/SATA and hosts
+where the Storage namespace is missing.
+
+That means the panel can receive the same physical drive twice, from sources
+that may disagree, plus rows where health could not be read at all. Every test
+here pins a way this card could start showing a failing or unreadable drive as
+fine, which is the only failure mode that matters: nobody investigates a disk
+that looks healthy.
+"""
+
+from plugins.endpoints.api.routes import _diskhealth_rows
+
+
+def _drive(deviceid, predictfailure, healthstatus='Healthy',
+           source='physicaldisk', model='Samsung SSD'):
+    return {
+        'deviceid': deviceid,
+        'model': model,
+        'mediatype': 'SSD',
+        'bustype': 'NVMe',
+        'healthstatus': healthstatus,
+        'predictfailure': predictfailure,
+        'source': source,
+    }
+
+
+def test_a_failing_drive_sorts_above_healthy_ones():
+    """On a multi-disk box the bad drive must not be below the fold."""
+    rows = _diskhealth_rows([
+        _drive('0', False),
+        _drive('1', True, healthstatus='Unhealthy'),
+        _drive('2', False),
+    ])
+
+    assert [row['deviceid'] for row in rows] == ['1', '0', '2']
+    assert rows[0]['verdict'] == 'Failure predicted'
+
+
+def test_a_drive_that_did_not_answer_reads_as_unknown_not_ok():
+    """Null predictfailure is an unread drive, not a healthy one."""
+    rows = _diskhealth_rows([_drive('0', None, healthstatus='')])
+
+    assert rows[0]['verdict'] == 'Unknown'
+    assert rows[0]['healthstatus'] == 'Unknown'
+
+
+def test_unknown_outranks_ok_so_silence_is_not_buried_under_health():
+    """A drive that stopped answering is a weaker claim than one saying OK."""
+    rows = _diskhealth_rows([
+        _drive('0', False),
+        _drive('1', None),
+    ])
+
+    assert [row['verdict'] for row in rows] == ['Unknown', 'OK']
+
+
+def test_both_sources_survive_for_the_same_hardware():
+    """Two disagreeing views of one drive is the signal, not noise.
+
+    Collapsing them would hide the case worth looking at: one source calling a
+    drive healthy while the other predicts failure.
+    """
+    rows = _diskhealth_rows([
+        _drive('0', False, source='physicaldisk'),
+        _drive('0', True, healthstatus='Pred Fail', source='diskdrive'),
+    ])
+
+    assert len(rows) == 2
+    assert {row['source'] for row in rows} == {'physicaldisk', 'diskdrive'}
+    assert rows[0]['verdict'] == 'Failure predicted'
+
+
+def test_input_order_does_not_change_which_drive_is_shown_first():
+    """The verdict ranking must decide the order, not arrival order."""
+    healthy = _drive('0', False)
+    failing = _drive('1', True)
+
+    forward = _diskhealth_rows([healthy, failing])
+    backward = _diskhealth_rows([failing, healthy])
+
+    assert forward == backward
+    assert forward[0]['deviceid'] == '1'
+
+
+def test_an_nvme_drive_is_reported_rather_than_dropped():
+    """The whole reason for the second source: ATA SMART returns nothing here."""
+    rows = _diskhealth_rows([_drive('0', False, source='physicaldisk')])
+
+    assert rows[0]['bustype'] == 'NVMe'
+    assert rows[0]['verdict'] == 'OK'
+
+
+def test_a_malformed_payload_does_not_take_the_panel_down():
+    rows = _diskhealth_rows([_drive('0', False), 'not a dict', None])
+
+    assert len(rows) == 1
+    assert rows[0]['deviceid'] == '0'
+
+
+def test_a_missing_or_wrong_shaped_payload_is_an_empty_panel():
+    assert _diskhealth_rows(None) == []
+    assert _diskhealth_rows({}) == []
+    assert _diskhealth_rows([]) == []

tests/test_plugins/test_endpoints_ip_history.py

@@ -0,0 +1,138 @@
+"""IP History and IP Configuration, carved out of the old Details Network card.
+
+These read core Communication rows, which accumulate as an ACCIDENTAL history:
+_record_interfaces matches an incoming NIC on MAC then on IP and inserts when
+neither matches, and never deletes. So an address the machine has not held for
+a year sits next to the current one. These tests pin the reading of that, not
+the collecting.
+
+The date columns are the delicate part. modifieddate only moves when a field on
+the row actually changes, so an address still held and reported unchanged keeps
+an old stamp. It is labelled 'Last changed' for that reason and must not drift
+back to 'Last seen'.
+"""
+from datetime import datetime
+from pathlib import Path
+
+from plugins.endpoints.api.routes import (
+    _IPHISTORY_COLUMNS,
+    _iphistory_status,
+    _isostamp,
+)
+
+
+class FakeComm:
+    """Only the columns these two routes read."""
+
+    def __init__(self, ipaddress='10.0.0.1', macaddress='AA:BB',
+                 isprimary=False, ismachinenetwork=False, isdhcp=False,
+                 subnetmask='', gateway='', dns1='', dns2='',
+                 createddate=None, modifieddate=None):
+        self.ipaddress = ipaddress
+        self.macaddress = macaddress
+        self.isprimary = isprimary
+        self.ismachinenetwork = ismachinenetwork
+        self.isdhcp = isdhcp
+        self.subnetmask = subnetmask
+        self.gateway = gateway
+        self.dns1 = dns1
+        self.dns2 = dns2
+        self.createddate = createddate
+        self.modifieddate = modifieddate
+
+
+def test_an_apipa_address_is_named_as_a_dhcp_failure():
+    # 169.254.x.x is not an address the machine was given - it is what Windows
+    # assigns when DHCP failed. Calling it 'Previous' would file a past network
+    # fault as a past subnet.
+    assert _iphistory_status(FakeComm(ipaddress='169.254.122.88')) == 'APIPA (no DHCP)'
+
+
+def test_apipa_wins_over_primary():
+    # A box currently failing DHCP can have its APIPA address flagged primary
+    # by the collector. The fault is the more useful thing to say.
+    comm = FakeComm(ipaddress='169.254.1.1', isprimary=True)
+    assert _iphistory_status(comm) == 'APIPA (no DHCP)'
+
+
+def test_a_normal_address_is_not_mistaken_for_apipa():
+    # Guards a substring match: 169.254 must be a prefix, not found anywhere.
+    assert _iphistory_status(FakeComm(ipaddress='10.169.254.5')) == 'Previous'
+
+
+def test_the_primary_address_is_marked():
+    assert _iphistory_status(FakeComm(isprimary=True)) == 'Primary'
+
+
+def test_a_machine_network_address_is_distinguished_from_a_stale_one():
+    # ismachinenetwork means it talks to the machine, not that it is old.
+    assert _iphistory_status(FakeComm(ismachinenetwork=True)) == 'Machine network'
+
+
+def test_everything_else_reads_as_previous():
+    assert _iphistory_status(FakeComm()) == 'Previous'
+
+
+def test_a_missing_timestamp_yields_null_not_a_crash():
+    # createddate is nullable on legacy rows imported before BaseModel stamps.
+    assert _isostamp(None) is None
+
+
+def test_a_timestamp_is_emitted_as_utc_iso():
+    # Matches every other timestamp this file emits; the client formats it to
+    # the viewer's locale rather than being pinned to the server's.
+    stamp = _isostamp(datetime(2026, 9, 5, 11, 30, 0))
+    assert stamp == '2026-09-05T11:30:00Z'
+
+
+def test_the_second_date_is_not_labelled_last_seen():
+    # The whole reason for the label. modifieddate cannot support 'last seen':
+    # an unchanged address reported daily keeps a months-old stamp.
+    labels = {col['key']: col['label'] for col in _IPHISTORY_COLUMNS}
+    assert labels['lastchanged'] == 'Last changed'
+    assert 'seen' not in labels['lastchanged'].lower()
+    assert labels['firstseen'] == 'First seen'
+
+
+def test_the_history_ordering_compiles_on_mysql():
+    # Regression: the first cut ordered with .nullslast(), which is valid
+    # ANSI/Postgres but blows up on MySQL with a 1064 syntax error. The test
+    # suite runs on SQLite, so it passed and the page still 500'd in the
+    # browser. Compile the real ordering against the MySQL dialect so the
+    # dialect gap is caught here instead of in a container log.
+    from sqlalchemy import select
+    from sqlalchemy.dialects import mysql
+
+    from shopdb.api import Communication
+
+    ordering = (select(Communication.ipaddress)
+                .where(Communication.assetid == 1)
+                .order_by(Communication.modifieddate.is_(None),
+                          Communication.modifieddate.desc(),
+                          Communication.communicationid.desc()))
+
+    compiled = str(ordering.compile(dialect=mysql.dialect()))
+    assert 'NULLS LAST' not in compiled.upper()
+
+
+def test_every_panel_handler_still_has_a_route():
+    # Regression: inserting the ipconfig route directly above panel_applications
+    # ate that function's @endpoints_bp.route line, leaving a handler decorated
+    # with require_permission only. It imported fine and every test passed; the
+    # endpoint was simply gone. Assert each panel_* handler is actually mapped.
+    import re
+
+    from plugins.endpoints.api import routes as routes_module
+
+    source = Path(routes_module.__file__).read_text(encoding='utf-8')
+    handlers = re.findall(r'^def (panel_\w+)', source, re.MULTILINE)
+    assert handlers, 'expected panel handlers in routes.py'
+
+    routed = set()
+    for match in re.finditer(r'^@endpoints_bp\.route\(', source, re.MULTILINE):
+        tail = source[match.start():]
+        found = re.search(r'^def (\w+)', tail, re.MULTILINE)
+        if found:
+            routed.add(found.group(1))
+
+    assert set(handlers) - routed == set()

tests/test_plugins/test_endpoints_link_label.py

@@ -0,0 +1,49 @@
+"""NIC link label: an unknown link must never read as a working one.
+
+Link speed and duplex are paired onto the interface rows from a second WMI
+class, so three states are possible: read successfully, unreadable, and
+disconnected. Verified on a live box that a DISCONNECTED adapter returns empty
+Speed and FullDuplex - two of four NICs on the test machine were in that state.
+
+That is the trap these tests pin. Defaulting an absent speed to 0 or an absent
+duplex to False would report every unplugged NIC as half duplex, which is the
+exact signature of the cable/port fault this field exists to catch. A column
+that cries fault on every disconnected NIC is a column nobody reads.
+"""
+
+from plugins.endpoints.api.routes import _link_label
+
+
+def test_a_gigabit_full_duplex_link_is_not_annotated():
+    """The normal case should be quiet so the abnormal case stands out."""
+    assert _link_label(1000000000, True) == '1 Gb'
+
+
+def test_half_duplex_is_called_out_because_it_is_the_actionable_case():
+    assert _link_label(100000000, False) == '100 Mb half duplex'
+
+
+def test_a_disconnected_adapter_reads_as_unknown_not_half_duplex():
+    """The live-box case: disconnected NICs return empty speed and duplex.
+
+    Reading this as half duplex would flag every unplugged NIC as a fault.
+    """
+    assert _link_label(None, None) == '-'
+
+
+def test_an_unreadable_duplex_on_a_live_link_is_not_assumed_half():
+    """Speed known, duplex not. Unknown duplex must not render as a fault."""
+    assert _link_label(10000000000, None) == '10 Gb'
+
+
+def test_zero_speed_is_unknown_rather_than_a_zero_speed_link():
+    assert _link_label(0, True) == '-'
+
+
+def test_sub_gigabit_speeds_stay_in_megabits():
+    assert _link_label(10000000, True) == '10 Mb'
+
+
+def test_speeds_do_not_render_trailing_zeros():
+    """2.5Gb NICs are common now; the label should not read '2.5000 Gb'."""
+    assert _link_label(2500000000, True) == '2.5 Gb'

tests/test_plugins/test_endpoints_mapped_drives.py

@@ -0,0 +1,186 @@
+"""Mapped-drive merge: two disagreeing Windows sources into one honest row.
+
+Windows reports mapped drives twice - Win32_NetworkConnection (the 'net use'
+table) and Get-SmbMapping - and the two do not agree. Captured on one machine at
+one moment, same two mappings, Get-SmbMapping said OK for both while net use
+said Disconnected and Reconnecting. Trusting the structured source alone would
+have rendered a healthy Q: on a box whose Q: was down.
+
+So the probe collects both without reconciling them and _mapped_drives() decides
+what an operator is shown. Every test here pins a way that decision could
+silently start rendering a broken drive as a working one, which is the only
+failure mode of this card that matters: nobody reports a drive that looks fine.
+"""
+
+from plugins.endpoints.api.routes import _mapped_drives
+
+
+def _netuse(localname, remotename, status):
+    return {'localname': localname, 'remotename': remotename, 'status': status}
+
+
+def _smb(localpath, remotepath, status):
+    return {'localpath': localpath, 'remotepath': remotepath, 'status': status}
+
+
+def test_the_worse_of_two_disagreeing_sources_is_what_gets_shown():
+    """The real capture: Get-SmbMapping says OK, net use says Disconnected.
+
+    The drive was down. Showing OK because one source said so is the exact bug
+    this merge exists to prevent.
+    """
+    rows = _mapped_drives({
+        'netuse': [_netuse('Q:', r'\\server\specdata', 'Disconnected')],
+        'smbmapping': [_smb('Q:', r'\\server\specdata', 'OK')],
+    })
+
+    assert rows == [r'Q: \\server\specdata (Disconnected)']
+
+
+def test_reconnecting_is_worse_than_ok_and_survives_the_merge():
+    """Reconnecting is a real status from the capture, not a synonym for OK."""
+    rows = _mapped_drives({
+        'netuse': [_netuse('S:', r'\\server\shared', 'Reconnecting')],
+        'smbmapping': [_smb('S:', r'\\server\shared', 'OK')],
+    })
+
+    assert rows == [r'S: \\server\shared (Reconnecting)']
+
+
+def test_a_drive_both_sources_call_healthy_is_not_annotated():
+    """Only problems are labelled; tagging every drive '(OK)' buries the bad one."""
+    rows = _mapped_drives({
+        'netuse': [_netuse('H:', r'\\server\home', 'OK')],
+        'smbmapping': [_smb('H:', r'\\server\home', 'OK')],
+    })
+
+    assert rows == [r'H: \\server\home']
+
+
+def test_a_drive_only_one_source_knows_about_is_still_reported():
+    """Get-SmbMapping is absent on some builds and session types.
+
+    Requiring agreement between the sources would hide every mapped drive on
+    those machines rather than showing what is known about them.
+    """
+    rows = _mapped_drives({
+        'netuse': [_netuse('P:', r'\\server\programs', 'Disconnected')],
+        'smbmapping': [],
+    })
+
+    assert rows == [r'P: \\server\programs (Disconnected)']
+
+
+def test_the_smbmapping_key_missing_entirely_degrades_to_netuse():
+    """Old agents predate the second source; the panel must still work.
+
+    The server ships ahead of the agent republish, so this is the live shape
+    until every endpoint has the new probe.
+    """
+    rows = _mapped_drives({
+        'netuse': [_netuse('Q:', r'\\server\specdata', 'OK')],
+    })
+
+    assert rows == [r'Q: \\server\specdata']
+
+
+def test_a_drive_with_no_status_from_either_source_reads_as_unknown():
+    """A blank annotation reads as healthy, so a failed check must not be blank."""
+    rows = _mapped_drives({
+        'netuse': [_netuse('Z:', r'\\server\scratch', None)],
+        'smbmapping': [],
+    })
+
+    assert rows == [r'Z: \\server\scratch (Unknown)']
+
+
+def test_an_unrecognised_status_is_not_assumed_healthy():
+    """Unknown vocabulary ranks below OK deliberately.
+
+    Paired against OK specifically: a status the server does not recognise must
+    LOSE to a healthy one, or an unseen fault renders green. Pairing it against
+    a known-bad status would prove nothing, since the known-bad value would win
+    however unknowns are ranked.
+    """
+    rows = _mapped_drives({
+        'netuse': [_netuse('X:', r'\\server\thing', 'SomeNewStatus')],
+        'smbmapping': [_smb('X:', r'\\server\thing', 'OK')],
+    })
+
+    assert rows == [r'X: \\server\thing (SomeNewStatus)']
+
+
+def test_a_known_bad_status_beats_an_unrecognised_one():
+    """Unavailable outranks an unknown string; the worst named state wins."""
+    rows = _mapped_drives({
+        'netuse': [_netuse('X:', r'\\server\thing', 'Unavailable')],
+        'smbmapping': [_smb('X:', r'\\server\thing', 'SomeNewStatus')],
+    })
+
+    assert rows == [r'X: \\server\thing (Unavailable)']
+
+
+def test_which_source_reported_a_status_does_not_change_the_verdict():
+    """The same pair of statuses must resolve identically whichever side sent it.
+
+    _worse_drive_status ranks by severity and breaks ties by argument order, so
+    if two statuses ever rank EQUAL the answer silently depends on which source
+    happened to report which - and a drive would read healthy or broken based on
+    enumeration order. Asserting both directions is what makes an accidental tie
+    visible; a single-direction test passes on luck.
+    """
+    remote = r'\\server\thing'
+    pairs = [
+        ('OK', 'Disconnected'),
+        ('OK', 'SomeNewStatus'),
+        ('Reconnecting', 'Unavailable'),
+        ('OK', 'Reconnecting'),
+    ]
+
+    for first, second in pairs:
+        forward = _mapped_drives({
+            'netuse': [_netuse('X:', remote, first)],
+            'smbmapping': [_smb('X:', remote, second)],
+        })
+        backward = _mapped_drives({
+            'netuse': [_netuse('X:', remote, second)],
+            'smbmapping': [_smb('X:', remote, first)],
+        })
+        assert forward == backward, (
+            'statuses {!r}/{!r} resolve differently depending on which source '
+            'reported them'.format(first, second))
+
+
+def test_the_two_sources_are_paired_by_drive_letter_not_by_order():
+    """The sources enumerate independently and need not agree on ordering."""
+    rows = _mapped_drives({
+        'netuse': [
+            _netuse('Q:', r'\\server\specdata', 'OK'),
+            _netuse('S:', r'\\server\shared', 'OK'),
+        ],
+        'smbmapping': [
+            _smb('S:', r'\\server\shared', 'Disconnected'),
+            _smb('Q:', r'\\server\specdata', 'OK'),
+        ],
+    })
+
+    assert rows == [
+        r'Q: \\server\specdata',
+        r'S: \\server\shared (Disconnected)',
+    ]
+
+
+def test_no_user_context_at_all_yields_no_rows():
+    """Nobody signed in is a valid state, not a fault."""
+    assert _mapped_drives({}) == []
+
+
+def test_malformed_entries_are_skipped_rather_than_raising():
+    """The payload is agent-supplied; a bad row must not blank the whole card."""
+    rows = _mapped_drives({
+        'netuse': ['not-a-dict', None, {'status': 'OK'},
+                   _netuse('Q:', r'\\server\specdata', 'Disconnected')],
+        'smbmapping': 'not-a-list',
+    })
+
+    assert rows == [r'Q: \\server\specdata (Disconnected)']

tests/test_plugins/test_endpoints_mapped_drives_card.py

@@ -0,0 +1,118 @@
+"""The mapped drives card: structured rows, badges, and row order.
+
+The merge logic itself is pinned in test_endpoints_mapped_drives.py. These
+tests cover what the CARD adds on top: a per-drive badge colour and the row
+order that decides what an operator sees first.
+
+The failure mode this guards is silent: a broken drive rendered green, or
+sorted below the fold, leaves the card looking healthy and tells nobody.
+"""
+from plugins.endpoints.api.routes import (
+    _mapped_drive_rows, _mapped_drives, _mapped_drives_unhealthy_first,
+)
+
+
+def _context(netuse=None, smbmapping=None):
+    context = {}
+    if netuse is not None:
+        context['netuse'] = netuse
+    if smbmapping is not None:
+        context['smbmapping'] = smbmapping
+    return context
+
+
+def _netuse(local, remote, status):
+    return {'localname': local, 'remotename': remote, 'status': status}
+
+
+def test_a_healthy_drive_is_green():
+    rows = _mapped_drive_rows(_context(netuse=[_netuse('Z:', r'\\srv\share', 'OK')]))
+    assert rows[0]['statecolor'] == 'green'
+    assert rows[0]['statuslabel'] == 'OK'
+
+
+def test_a_disconnected_drive_is_red():
+    rows = _mapped_drive_rows(
+        _context(netuse=[_netuse('Z:', r'\\srv\share', 'Disconnected')]))
+    assert rows[0]['statecolor'] == 'red'
+
+
+def test_connected_counts_as_healthy_alongside_ok():
+    """net use and Get-SmbMapping use different words for the same state."""
+    rows = _mapped_drive_rows(_context(netuse=[_netuse('Z:', r'\\srv\s', 'Connected')]))
+    assert rows[0]['statecolor'] == 'green'
+
+
+def test_a_drive_with_no_status_is_red_not_green():
+    """An uncoloured row reads as working. Unknown health is not health."""
+    rows = _mapped_drive_rows(_context(netuse=[_netuse('Z:', r'\\srv\s', None)]))
+    assert rows[0]['statuslabel'] == 'Unknown'
+    assert rows[0]['statecolor'] == 'red'
+
+
+def test_an_unrecognised_status_is_not_coloured_green():
+    rows = _mapped_drive_rows(_context(netuse=[_netuse('Z:', r'\\srv\s', 'Weird')]))
+    assert rows[0]['statecolor'] == 'red'
+
+
+def test_the_worse_of_two_sources_drives_the_badge_colour():
+    """The merge already picks the worse status; the badge must follow it."""
+    rows = _mapped_drive_rows(_context(
+        netuse=[_netuse('Z:', r'\\srv\share', 'OK')],
+        smbmapping=[{'localpath': 'Z:', 'remotepath': r'\\srv\share',
+                     'status': 'Disconnected'}],
+    ))
+    assert len(rows) == 1
+    assert rows[0]['statecolor'] == 'red'
+
+
+def test_the_remote_path_is_carried_as_its_own_field():
+    """The card shows the path separately rather than glued into the title."""
+    rows = _mapped_drive_rows(_context(netuse=[_netuse('Z:', r'\\srv\share', 'OK')]))
+    assert rows[0]['drive'] == 'Z:'
+    assert rows[0]['remote'] == r'\\srv\share'
+
+
+def test_the_summary_line_and_the_card_agree_about_a_broken_drive():
+    """Both are built from the same rows, so they cannot drift apart."""
+    context = _context(netuse=[_netuse('Z:', r'\\srv\share', 'Disconnected')])
+    assert _mapped_drive_rows(context)[0]['statecolor'] == 'red'
+    assert '(Disconnected)' in _mapped_drives(context)[0]
+
+
+def test_a_healthy_drive_is_not_annotated_in_the_summary_line():
+    context = _context(netuse=[_netuse('Z:', r'\\srv\share', 'OK')])
+    assert _mapped_drives(context) == [r'Z: \\srv\share']
+
+
+def test_no_user_context_yields_no_rows():
+    assert _mapped_drive_rows({}) == []
+
+
+def test_broken_drives_sort_above_healthy_ones():
+    """The broken one must be on screen without scrolling past nine good ones."""
+    rows = _mapped_drive_rows(_context(netuse=[
+        _netuse('H:', r'\\srv\home', 'OK'),
+        _netuse('S:', r'\\srv\shared', 'OK'),
+        _netuse('Z:', r'\\srv\dead', 'Disconnected'),
+    ]))
+    ordered = _mapped_drives_unhealthy_first(rows)
+    assert [row['drive'] for row in ordered] == ['Z:', 'H:', 'S:']
+
+
+def test_drive_letter_order_survives_within_each_health_group():
+    """Sorting only on health keeps the list from reshuffling on refresh.
+
+    The unhealthy drives here carry DIFFERENT status words, chosen so that
+    sorting on the status text would reorder them (Disconnected before
+    Unavailable) against drive-letter order. A secondary sort key would pass
+    a test where every broken drive said the same thing.
+    """
+    rows = _mapped_drive_rows(_context(netuse=[
+        _netuse('A:', r'\\srv\a', 'OK'),
+        _netuse('B:', r'\\srv\b', 'Unavailable'),
+        _netuse('C:', r'\\srv\c', 'Connected'),
+        _netuse('D:', r'\\srv\d', 'Disconnected'),
+    ]))
+    ordered = _mapped_drives_unhealthy_first(rows)
+    assert [row['drive'] for row in ordered] == ['B:', 'D:', 'A:', 'C:']

tests/test_plugins/test_endpoints_panel_groups.py

@@ -0,0 +1,74 @@
+"""The Monitors tab split: a panel in a group no tab renders disappears.
+
+This has silently eaten cards twice - once when tracked services moved out of
+'actions', once on the 'metrics' to 'monitors' rename. Both times the backend
+named a group the frontend did not render, and the panel was not shown, not
+logged, and not missed until someone went looking for it.
+
+The split is only correct if four places agree, and only two of them are Python:
+  1. the panels' group here
+  2. RENDERED_GROUPS in PluginAssetPanels.vue
+  3. the tab label map in PCDetail.vue
+  4. the v-show block that gives the tab a layout
+
+These tests pin the Python half so a later edit cannot quietly move a panel into
+a group nothing renders. The frontend half is pinned by its own spec.
+"""
+
+from plugins.endpoints.plugin import EndpointsPlugin
+from shopdb.core.api.pluginui import ASSET_PANEL_GROUPS
+
+
+# Derived from the server tuple rather than hand-copied. This used to be a
+# literal set duplicating RENDERED_GROUPS, which made it a FIFTH copy of the
+# vocabulary and one more thing to forget: it passed for a group the server
+# would rewrite to 'details' anyway. ASSET_PANEL_GROUPS is the value that
+# actually decides what reaches the client, and panelGroups.spec.js pins it
+# against the frontend copy.
+RENDERED_GROUPS = set(ASSET_PANEL_GROUPS)
+
+
+def _panels():
+    return {panel['id']: panel for panel in EndpointsPlugin().get_asset_panels()}
+
+
+def test_no_panel_lands_in_a_group_the_frontend_does_not_render():
+    """The failure mode that ate cards twice: a group with no tab."""
+    for panel in EndpointsPlugin().get_asset_panels():
+        group = panel.get('group', 'details')
+        assert group in RENDERED_GROUPS, (
+            f"panel {panel['id']} is in group {group!r}, which no tab renders"
+        )
+
+
+def test_tracked_services_and_processes_moved_to_their_own_group():
+    panels = _panels()
+
+    assert panels['endpoints-tracked-services']['group'] == 'serviceprocess'
+    assert panels['endpoints-tracked-processes']['group'] == 'serviceprocess'
+
+
+def test_the_service_process_tab_is_not_left_empty():
+    """A tab whose panels all moved away reads as broken, not as absent."""
+    groups = [panel.get('group', 'details')
+              for panel in EndpointsPlugin().get_asset_panels()]
+
+    assert 'serviceprocess' in groups
+
+
+def test_system_monitors_keeps_panels_of_its_own():
+    """The rename is only safe if Monitors still has content after the move."""
+    groups = [panel.get('group', 'details')
+              for panel in EndpointsPlugin().get_asset_panels()]
+
+    assert 'monitors' in groups
+
+
+def test_disk_health_stayed_with_the_system_health_panels():
+    """Disk health is the box's own health, not something running on it.
+
+    It is now a section of the Storage card rather than a panel of its own,
+    so the group assertion follows it there.
+    """
+    assert 'endpoints-diskhealth' not in _panels()
+    assert _panels()['endpoints-storage']['group'] == 'monitors'

tests/test_plugins/test_endpoints_plugin.py

@@ -69,12 +69,12 @@ def test_endpoints_asset_panels_are_computer_scoped():
     """Every asset panel targets computer assets only, with an endpoint."""
     plugin = EndpointsPlugin()
     panels = plugin.get_asset_panels()
-    assert len(panels) == 7
+    assert len(panels) == 14
     for panel in panels:
         assert panel['assettypes'] == ['computer']
         assert panel['endpoint'].startswith('/api/endpoints/')
         assert '{assetid}' in panel['endpoint']
-        assert panel['render'] in ('keyvalue', 'list', 'table', 'operatorreports')
+        assert panel['render'] in ('keyvalue', 'list', 'table', 'tabs', 'operatorreports')
 
 
 def test_services_and_processes_are_split_into_two_filtered_cards():
@@ -220,18 +220,21 @@ def test_known_payload_keys_cover_persisted_and_relayed_fields():
     assert persisted_or_relayed <= KNOWN_PAYLOAD_KEYS
 
 
-def test_new_hardware_telemetry_keys_are_logged_not_silently_known():
-    """Disk/bios/memory inventory is collected at the agent before the server
-    persists it. Those keys must be OUTSIDE the known set so ingest logs them
-    rather than dropping them silently."""
+def test_hardware_telemetry_keys_are_persisted_not_merely_logged():
+    """Disk/bios/memory/processor/osinfo inventory was collected at the agent and
+    logged as unknown keys for several versions while the shapes were proven
+    against real endpoint traffic. They have columns now (endpoints migration
+    0013), so they must be INSIDE the known set - leaving them out would log
+    every report as carrying unpersisted keys while the ingest quietly stores
+    them, which is the most misleading combination available."""
     from plugins.endpoints.plugin import KNOWN_PAYLOAD_KEYS
 
-    for key in ('disks', 'bios', 'memory', 'processor', 'osinfo', 'domain'):
-        assert key not in KNOWN_PAYLOAD_KEYS
+    for key in ('disks', 'bios', 'memory', 'processor', 'osinfo', 'domain',
+                'pendingreboot', 'usercontext'):
+        assert key in KNOWN_PAYLOAD_KEYS
 
-    # 'applications' USED to be in this list. It is persisted now - that is what
-    # the installed-software work added - so it belongs in the known set, and
-    # asserting its absence would fail the moment the feature it tests works.
+    # 'applications' followed this same staging path earlier and is likewise
+    # persisted, to its own endpointapplications table rather than a state blob.
     assert 'applications' in KNOWN_PAYLOAD_KEYS
 
 

tests/test_plugins/test_endpoints_storage_sections.py

@@ -0,0 +1,51 @@
+"""The Storage panel serves volumes and disk health as two sections.
+
+Visual grouping only. The two tables share no key - storage is per volume and
+health is per physical device - so nothing is reconciled across them. These
+tests pin that the sections stay independent and that neither disappears when
+empty, because a dropped section takes its tab with it and reads as "no disks"
+rather than "nothing was reported".
+"""
+import pytest
+
+from plugins.endpoints.api.routes import _storage_rows, _STORAGE_COLUMNS, _DISKHEALTH_COLUMNS
+
+
+def test_freepercent_is_derived_when_an_older_agent_omits_it():
+    rows = _storage_rows([{'drive': 'C:', 'sizebytes': 1000, 'freebytes': 250}])
+    assert rows[0]['freepercent'] == '25.0%'
+
+
+def test_an_agent_supplied_freepercent_is_preferred_over_recomputing():
+    rows = _storage_rows([
+        {'drive': 'C:', 'sizebytes': 1000, 'freebytes': 250, 'freepercent': 24.9},
+    ])
+    assert rows[0]['freepercent'] == '24.9%'
+
+
+def test_a_zero_size_volume_does_not_divide_by_zero():
+    rows = _storage_rows([{'drive': 'C:', 'sizebytes': 0, 'freebytes': 0}])
+    assert rows[0]['freepercent'] is None
+
+
+def test_free_percent_is_none_rather_than_zero_when_free_is_unknown():
+    """Zero free space and unknown free space must not look identical."""
+    rows = _storage_rows([{'drive': 'C:', 'sizebytes': 1000}])
+    assert rows[0]['freepercent'] is None
+
+
+def test_non_dict_entries_are_skipped():
+    rows = _storage_rows([{'drive': 'C:', 'sizebytes': 1000, 'freebytes': 500}, 'junk', None])
+    assert len(rows) == 1
+
+
+@pytest.mark.parametrize('value', [None, {}, 'x'])
+def test_a_missing_snapshot_yields_no_rows(value):
+    assert _storage_rows(value) == []
+
+
+def test_the_two_section_column_sets_share_no_key():
+    """The point of two sections rather than one merged table."""
+    storage = {column['key'] for column in _STORAGE_COLUMNS}
+    health = {column['key'] for column in _DISKHEALTH_COLUMNS}
+    assert not storage & health