mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-25 18:40:06 -04:00
Surface coarse SMART health for USB-bridge drives (#498)
* Surface coarse SMART health for USB-bridge drives USB enclosures that reject ATA pass-through (e.g. Seagate Expansion) are invisible to pySMART/DeviceList, so the drive shows no SMART at all on the Controller page. Fall back to smartctl over the SCSI transport, which many such bridges still answer, to report a coarse PASS/FAIL health plus drive identity (model/serial/capacity). The health (-d scsi -H) and identity (-d scsi -i) reads are issued as separate calls: combining them makes smartctl suppress the health line when the identity read reports "SMART support is: Unavailable" (which these bridges do). The entry is marked smart_limited (labelled "(limited)" in both UIs) so a green PASS is not mistaken for full monitoring — these drives expose no attributes or temperature, and some bridges report "OK" statically. * Harden SCSI health fallback (Greptile review) - _scsi_health: treat any unrecognised status (e.g. "UNKNOWN", "N/A") as unreadable (None) instead of a false FAIL. Only "OK" -> PASS and "FAIL*" -> FAIL; anything else omits the drive rather than showing a red FAIL that contradicts the smart_limited disclaimer. - get_all_smart_status: isolate the SCSI supplement loop in its own try/except so an error there cannot discard the pySMART results already collected. --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
9f0e3fa0aa
commit
dac849f9f6
4 changed files with 298 additions and 14 deletions
|
|
@ -45,16 +45,31 @@ def get_all_smart_status() -> list[dict]:
|
|||
return []
|
||||
|
||||
try:
|
||||
devices = DeviceList()
|
||||
results = []
|
||||
|
||||
for device in devices.devices:
|
||||
covered = set()
|
||||
for device in DeviceList().devices:
|
||||
results.append(_get_device_smart(device))
|
||||
|
||||
return results
|
||||
covered.add(f"/dev/{device.name}")
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
# Drives behind USB enclosures that reject ATA pass-through (e.g. Seagate
|
||||
# Expansion, WD easystore) are invisible to pySMART/DeviceList. smartctl
|
||||
# --scan still lists them; add a coarse SCSI health entry so the drive is
|
||||
# at least represented instead of silently unmonitored. Isolated from the
|
||||
# pySMART read above: a failure here must not discard those results.
|
||||
try:
|
||||
for path, _interface in _scan_devices():
|
||||
if path in covered:
|
||||
continue
|
||||
limited = _limited_device_smart(path)
|
||||
if limited is not None:
|
||||
results.append(limited)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def _get_device_smart(device) -> dict:
|
||||
"""Extract SMART data from a pySMART device."""
|
||||
|
|
@ -180,6 +195,117 @@ def _scan_devices() -> list[tuple[str, Optional[str]]]:
|
|||
return devices
|
||||
|
||||
|
||||
def _read_scsi_limited(path: str) -> tuple:
|
||||
"""Read the coarse SCSI health (and identity) for a USB-bridge drive.
|
||||
|
||||
USB enclosures that reject ATA pass-through expose no SMART attributes,
|
||||
but many still answer the SCSI health query. Both the health check and
|
||||
INQUIRY are non-media SCSI commands, so they do not spin up a parked
|
||||
drive.
|
||||
|
||||
The health and identity reads MUST be issued as separate smartctl calls:
|
||||
when ``-i`` reports "SMART support is: Unavailable" (which these bridges
|
||||
do), combining it with ``-H`` makes smartctl suppress the health line,
|
||||
even though the standalone ``-d scsi -H`` query answers fine.
|
||||
|
||||
Returns ``(health, model, serial, capacity)`` where health is 'PASS',
|
||||
'FAIL', or None. This is the only signal available for such drives; it
|
||||
carries no attributes or temperature, and some bridges report 'OK'
|
||||
statically, so it is a weak signal — callers mark it ``smart_limited``.
|
||||
"""
|
||||
health = _scsi_health(path)
|
||||
if health is None:
|
||||
# No health signal means nothing useful; skip the identity read too.
|
||||
return (None, None, None, None)
|
||||
model, serial, capacity = _scsi_identity(path)
|
||||
return (health, model, serial, capacity)
|
||||
|
||||
|
||||
def _scsi_health(path: str) -> Optional[str]:
|
||||
"""Return 'PASS'/'FAIL'/None from ``smartctl -d scsi -H`` (alone)."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["smartctl", "-d", "scsi", "-H", path],
|
||||
capture_output=True, text=True, timeout=SMARTCTL_STANDBY_TIMEOUT,
|
||||
)
|
||||
except (subprocess.SubprocessError, OSError):
|
||||
return None
|
||||
for line in result.stdout.splitlines():
|
||||
key, sep, val = line.partition(":")
|
||||
if sep and key.strip().lower() == "smart health status":
|
||||
val = val.strip().upper()
|
||||
# Standard SCSI health is "OK" or "FAILED!". Some bridges emit
|
||||
# non-standard strings (e.g. "UNKNOWN", "N/A"); treat anything we
|
||||
# do not recognise as unreadable (None) rather than a false FAIL.
|
||||
if val == "OK":
|
||||
return "PASS"
|
||||
if val.startswith("FAIL"):
|
||||
return "FAIL"
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _scsi_identity(path: str) -> tuple:
|
||||
"""Return ``(model, serial, capacity)`` from ``smartctl -d scsi -i``."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["smartctl", "-d", "scsi", "-i", path],
|
||||
capture_output=True, text=True, timeout=SMARTCTL_STANDBY_TIMEOUT,
|
||||
)
|
||||
except (subprocess.SubprocessError, OSError):
|
||||
return (None, None, None)
|
||||
vendor = product = serial = capacity = None
|
||||
for line in result.stdout.splitlines():
|
||||
key, sep, val = line.partition(":")
|
||||
if not sep:
|
||||
continue
|
||||
key = key.strip().lower()
|
||||
val = val.strip()
|
||||
if key == "vendor":
|
||||
vendor = val
|
||||
elif key == "product":
|
||||
product = val
|
||||
elif key == "serial number":
|
||||
serial = val
|
||||
elif key == "user capacity":
|
||||
# e.g. "26,000,658,267,648 bytes [26.0 TB]" -> "26.0 TB"
|
||||
bracket = re.search(r'\[([^\]]+)\]', val)
|
||||
capacity = bracket.group(1) if bracket else (val or None)
|
||||
model = " ".join(p for p in (vendor, product) if p) or None
|
||||
return (model, serial, capacity)
|
||||
|
||||
|
||||
def _limited_device_smart(path: str) -> Optional[dict]:
|
||||
"""Build a coarse SMART payload for a drive pySMART cannot read.
|
||||
|
||||
Used for USB-bridge drives without ATA pass-through. Only the SCSI
|
||||
health is available; attributes, temperature and power-on hours are
|
||||
unknown. The payload is shaped like ``_get_device_smart`` so the UIs can
|
||||
render it uniformly, but is marked ``smart_limited`` so a green PASS is
|
||||
not mistaken for full monitoring. Returns None when even the SCSI health
|
||||
is unavailable (nothing useful to show).
|
||||
"""
|
||||
health, model, serial, capacity = _read_scsi_limited(path)
|
||||
if health is None:
|
||||
return None
|
||||
return {
|
||||
"device": path.rsplit("/", 1)[-1],
|
||||
"path": path,
|
||||
"model": model,
|
||||
"serial": serial,
|
||||
"capacity": capacity,
|
||||
"assessment": None, # the drive's own ATA self-assessment is unreachable
|
||||
"health": health,
|
||||
"temperature": None,
|
||||
"power_on_hours": None,
|
||||
"reallocated_sectors": None,
|
||||
"pending_sectors": None,
|
||||
"offline_uncorrectable": None,
|
||||
"smart_enabled": False,
|
||||
"smart_limited": True,
|
||||
}
|
||||
|
||||
|
||||
def _drive_is_asleep(path: str) -> bool:
|
||||
"""Return True if the drive at ``path`` is in standby/sleep.
|
||||
|
||||
|
|
|
|||
|
|
@ -1786,7 +1786,8 @@
|
|||
|
||||
for (const drive of drives) {
|
||||
const assessmentClass = getHealthClass(drive.health || drive.assessment);
|
||||
const assessmentText = drive.health || drive.assessment || 'Unknown';
|
||||
const assessmentText = (drive.health || drive.assessment || 'Unknown')
|
||||
+ (drive.smart_limited ? ' (limited)' : '');
|
||||
const tempText = drive.temperature !== null ? `${drive.temperature}°C` : '--';
|
||||
|
||||
html += `
|
||||
|
|
@ -1859,7 +1860,7 @@
|
|||
</tr>
|
||||
<tr>
|
||||
<td style="color: #888;">Assessment</td>
|
||||
<td class="${assessmentClass}">${drive.assessment || 'Unknown'}</td>
|
||||
<td class="${assessmentClass}">${drive.assessment || drive.health || 'Unknown'}${drive.smart_limited ? ' (limited — USB enclosure reports health only)' : ''}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td style="color: #888;">Temperature</td>
|
||||
|
|
|
|||
|
|
@ -15,11 +15,33 @@ from controller.lib.smart import (
|
|||
_get_attribute,
|
||||
_parse_raw_int,
|
||||
_derive_health,
|
||||
_read_scsi_limited,
|
||||
_limited_device_smart,
|
||||
_drive_is_asleep,
|
||||
HDD_HIGH_TEMPERATURE,
|
||||
HDD_CRITICAL_TEMPERATURE,
|
||||
)
|
||||
|
||||
# Real smartctl output from a Seagate Expansion USB enclosure that rejects
|
||||
# ATA pass-through. The health (`-d scsi -H`) and identity (`-d scsi -i`)
|
||||
# reads MUST be separate calls: combining `-i` makes smartctl suppress the
|
||||
# health line.
|
||||
SCSI_HEALTH_OUTPUT = """\
|
||||
=== START OF READ SMART DATA SECTION ===
|
||||
SMART Health Status: OK
|
||||
"""
|
||||
|
||||
SCSI_IDENTITY_OUTPUT = """\
|
||||
=== START OF INFORMATION SECTION ===
|
||||
Vendor: Seagate
|
||||
Product: Expansion HDD
|
||||
Revision: 1802
|
||||
User Capacity: 26,000,658,267,648 bytes [26.0 TB]
|
||||
Serial number: 00000000NT17S4W2
|
||||
Device type: disk
|
||||
SMART support is: Unavailable - device lacks SMART capability.
|
||||
"""
|
||||
|
||||
|
||||
def _attr(name, raw):
|
||||
a = mock.Mock()
|
||||
|
|
@ -73,12 +95,66 @@ class TestGetAllSmartStatus:
|
|||
mock_device_list = mock.Mock()
|
||||
mock_device_list.devices = [mock_device]
|
||||
|
||||
with mock.patch("controller.lib.smart.is_smart_available", return_value=True):
|
||||
with mock.patch("controller.lib.smart.DeviceList", return_value=mock_device_list):
|
||||
result = get_all_smart_status()
|
||||
assert len(result) == 1
|
||||
assert result[0]["device"] == "sda"
|
||||
assert result[0]["model"] == "Samsung SSD"
|
||||
with mock.patch("controller.lib.smart.is_smart_available", return_value=True), \
|
||||
mock.patch("controller.lib.smart.DeviceList", return_value=mock_device_list), \
|
||||
mock.patch("controller.lib.smart._scan_devices", return_value=[]):
|
||||
result = get_all_smart_status()
|
||||
assert len(result) == 1
|
||||
assert result[0]["device"] == "sda"
|
||||
assert result[0]["model"] == "Samsung SSD"
|
||||
|
||||
def test_supplements_with_scsi_health_for_bridge_drive(self):
|
||||
"""A scan-visible drive that pySMART/DeviceList cannot read (USB
|
||||
enclosure) is added via the coarse SCSI health fallback."""
|
||||
ata = mock.Mock()
|
||||
ata.name = "sdc"
|
||||
ata.model = "WDC"
|
||||
ata.serial = "S2"
|
||||
ata.capacity = "14.0 TB"
|
||||
ata.assessment = "PASS"
|
||||
ata.smart_enabled = True
|
||||
ata.attributes = []
|
||||
|
||||
device_list = mock.Mock()
|
||||
device_list.devices = [ata]
|
||||
limited = {"device": "sda", "path": "/dev/sda", "health": "PASS",
|
||||
"smart_limited": True}
|
||||
|
||||
with mock.patch("controller.lib.smart.is_smart_available", return_value=True), \
|
||||
mock.patch("controller.lib.smart.DeviceList", return_value=device_list), \
|
||||
mock.patch("controller.lib.smart._scan_devices",
|
||||
return_value=[("/dev/sdc", "sat"), ("/dev/sda", "sat")]), \
|
||||
mock.patch("controller.lib.smart._limited_device_smart",
|
||||
return_value=limited) as limited_fn:
|
||||
result = get_all_smart_status()
|
||||
|
||||
# sdc came from DeviceList (covered); only sda hits the fallback.
|
||||
limited_fn.assert_called_once_with("/dev/sda")
|
||||
devices = {d["device"] for d in result}
|
||||
assert devices == {"sdc", "sda"}
|
||||
|
||||
def test_supplement_failure_preserves_pysmart_results(self):
|
||||
"""An error in the SCSI supplement must not discard the drives already
|
||||
collected from pySMART."""
|
||||
ata = mock.Mock()
|
||||
ata.name = "sdb"
|
||||
ata.model = "Seagate"
|
||||
ata.serial = "S1"
|
||||
ata.capacity = "8.0 TB"
|
||||
ata.assessment = "PASS"
|
||||
ata.smart_enabled = True
|
||||
ata.attributes = []
|
||||
|
||||
device_list = mock.Mock()
|
||||
device_list.devices = [ata]
|
||||
|
||||
with mock.patch("controller.lib.smart.is_smart_available", return_value=True), \
|
||||
mock.patch("controller.lib.smart.DeviceList", return_value=device_list), \
|
||||
mock.patch("controller.lib.smart._scan_devices",
|
||||
side_effect=RuntimeError("scan blew up")):
|
||||
result = get_all_smart_status()
|
||||
|
||||
assert [d["device"] for d in result] == ["sdb"]
|
||||
|
||||
|
||||
class TestGetDeviceSmart:
|
||||
|
|
@ -376,3 +452,80 @@ class TestBuildSmartStats:
|
|||
|
||||
Dev.assert_not_called()
|
||||
assert stats["drives"] == []
|
||||
|
||||
|
||||
class TestReadScsiLimited:
|
||||
"""Tests for _read_scsi_limited — coarse health for USB-bridge drives.
|
||||
|
||||
Health and identity are two separate smartctl calls (combining `-H -i`
|
||||
suppresses the health line on these bridges), so subprocess.run is mocked
|
||||
with a side_effect sequence: first the `-H` read, then the `-i` read.
|
||||
"""
|
||||
|
||||
def test_parses_ok_health_and_identity(self):
|
||||
runs = [mock.Mock(stdout=SCSI_HEALTH_OUTPUT),
|
||||
mock.Mock(stdout=SCSI_IDENTITY_OUTPUT)]
|
||||
with mock.patch("controller.lib.smart.subprocess.run", side_effect=runs):
|
||||
health, model, serial, capacity = _read_scsi_limited("/dev/sda")
|
||||
assert health == "PASS"
|
||||
assert model == "Seagate Expansion HDD"
|
||||
assert serial == "00000000NT17S4W2"
|
||||
assert capacity == "26.0 TB"
|
||||
|
||||
def test_parses_failed_health(self):
|
||||
runs = [mock.Mock(stdout="SMART Health Status: FAILED\n"),
|
||||
mock.Mock(stdout=SCSI_IDENTITY_OUTPUT)]
|
||||
with mock.patch("controller.lib.smart.subprocess.run", side_effect=runs):
|
||||
health, _model, _serial, _capacity = _read_scsi_limited("/dev/sda")
|
||||
assert health == "FAIL"
|
||||
|
||||
def test_no_health_line_skips_identity_and_returns_none(self):
|
||||
"""No health line → don't bother reading identity; report nothing."""
|
||||
run = mock.Mock(stdout="Vendor: WD\nProduct: Thing\n")
|
||||
with mock.patch("controller.lib.smart.subprocess.run",
|
||||
return_value=run) as run_fn:
|
||||
result = _read_scsi_limited("/dev/sda")
|
||||
assert result == (None, None, None, None)
|
||||
# Only the health read happened; identity was skipped.
|
||||
run_fn.assert_called_once()
|
||||
|
||||
def test_unrecognised_status_is_treated_as_unreadable(self):
|
||||
"""A non-standard bridge status (e.g. UNKNOWN) must NOT show as a red
|
||||
FAIL; treat it as unreadable so the drive is omitted, not falsely
|
||||
failed."""
|
||||
run = mock.Mock(stdout="SMART Health Status: UNKNOWN\n")
|
||||
with mock.patch("controller.lib.smart.subprocess.run",
|
||||
return_value=run) as run_fn:
|
||||
result = _read_scsi_limited("/dev/sda")
|
||||
assert result == (None, None, None, None)
|
||||
run_fn.assert_called_once()
|
||||
|
||||
def test_smartctl_error_returns_all_none(self):
|
||||
with mock.patch("controller.lib.smart.subprocess.run",
|
||||
side_effect=OSError("boom")):
|
||||
assert _read_scsi_limited("/dev/sda") == (None, None, None, None)
|
||||
|
||||
|
||||
class TestLimitedDeviceSmart:
|
||||
"""Tests for _limited_device_smart — the limited payload shape."""
|
||||
|
||||
def test_builds_limited_payload(self):
|
||||
with mock.patch("controller.lib.smart._read_scsi_limited",
|
||||
return_value=("PASS", "Seagate Expansion HDD",
|
||||
"NT17S4W2", "26.0 TB")):
|
||||
payload = _limited_device_smart("/dev/sda")
|
||||
assert payload["device"] == "sda"
|
||||
assert payload["path"] == "/dev/sda"
|
||||
assert payload["health"] == "PASS"
|
||||
assert payload["model"] == "Seagate Expansion HDD"
|
||||
assert payload["smart_limited"] is True
|
||||
# Attributes are unreachable through the bridge.
|
||||
assert payload["assessment"] is None
|
||||
assert payload["temperature"] is None
|
||||
assert payload["pending_sectors"] is None
|
||||
|
||||
def test_returns_none_when_health_unavailable(self):
|
||||
"""No SCSI health → nothing useful to show, omit the drive."""
|
||||
with mock.patch("controller.lib.smart._read_scsi_limited",
|
||||
return_value=(None, None, None, None)):
|
||||
assert _limited_device_smart("/dev/sda") is None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue