ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Import the wilm endpoint telemetry, PC Detail rework and backup overlay

Browse Source 
Ports 37 of Benjamin Baker's commits from the wilm branch, mirrored from the
corporate GitHub at 7905917c. wilm rides the publication lineage and shares no
ancestor with wj, so this is a filtered tree patch rather than a merge:
94 files, and every path the publication export strips (docs/, tools/, mcp/,
CLAUDE.md, start-*.sh, .gitea/, the three excluded tests) is held back, because
a raw diff would delete each of them from this tree.

What comes across:

* Endpoint telemetry, phase 2. Physical disk health from two sources, NIC link
  state, boot type, Defender health, processor/memory/chassis inventory, the
  listening TCP set, mapped network drives and Windows Update history. Five
  additive migrations, 0012 through 0016, off the 0011 head.
* Mapped drives are collected by a probe running in the interactive user's
  session, because they do not exist in the agent's session 0. Two Windows
  sources can disagree about one drive and the worse status wins, so a
  half-broken share is never rendered as healthy.
* Windows events are reported once each, keyed on the channel's own record id,
  instead of being re-sent whenever a report overlapped the last one.
* PC Detail rework: Details cards flow, the hand-written Network card gives way
  to a full address history, Storage and disk health merge into one card with
  two sections, Operator Reports and Corporate Tools get their own tabs.
* An opt-in restic backup overlay with a third container, restore gated on the
  dump's own data revision rather than the code head, and backup/restore history.
* The agent's version is read from the binary's PE version rather than trusted
  from the csproj, and the Fleet view gains an Agent Version column.
* The enqueue route now asks caller_holds instead of the owner's permissions,
  which closes the scoped-token elevation on run_script, update_app and
  system_restart.

Held back deliberately:

* agent/installer/install-shopdbagent.ps1 stays deleted, as upstream deleted it.
  The installer that reaches a bay is generated by _install_script() in
  plugins/endpoints/api/siteagent.py; the standalone script had stopped being
  shipped and was collecting fixes no endpoint ever ran.
* The .gitignore hunk adding docs/*.md. Correct on the publication lineage,
  where docs/ becomes the wiki and its absence is how the gates detect a
  published checkout. Here it would silently ignore every new doc.
* Their deletions of docs/DEV-BUILD-TEST.md and docs/PLUGIN-HOOKS.md. Upstream
  those were stubs; here PLUGIN-HOOKS.md is the 31KB document CLAUDE.md points
  readers at.
* Five zero-byte files committed by an accidental `git add -A`, a run log from
  the author's own machine, and docs/notes/, which is working scratch that says
  so in its own header and belongs in devnotes/ on this side.

Carried with the import so it lands green: the 17 new routes are described in
docs/api-inventory.json with the spec and project map regenerated, and two new
frontend specs read fixtures through fileURLToPath(URL) which node rejects under
jsdom's own URL global.

Suite 2591 passed / 9 skipped; frontend 391 passed; naming, style and doc gates
clean.

Scope: product
This commit is contained in:
cproudlock
2026-09-06 16:47:24 -04:00
parent 9947a477b8
commit 6e36df841e
96 changed files with 12395 additions and 761 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
tests/test_plugins/test_endpoints_agentupload.py Normal file
View File

@@ -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

168
tests/test_plugins/test_endpoints_dellupdates.py Normal file
View File

@@ -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)'

108
tests/test_plugins/test_endpoints_disk_health.py Normal file
View File

@@ -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([]) == []

138
tests/test_plugins/test_endpoints_ip_history.py Normal file
View File

@@ -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()

49
tests/test_plugins/test_endpoints_link_label.py Normal file
View File

@@ -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'

186
tests/test_plugins/test_endpoints_mapped_drives.py Normal file
View File

@@ -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)']

118
tests/test_plugins/test_endpoints_mapped_drives_card.py Normal file
View File

@@ -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:']

74
tests/test_plugins/test_endpoints_panel_groups.py Normal file
View File

@@ -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'

25
tests/test_plugins/test_endpoints_plugin.py
View File

@@ -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

51
tests/test_plugins/test_endpoints_storage_sections.py Normal file
View File

@@ -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