# tests/plugins/host_agent/test_drive_health.py """#5789: each drive's health, read with smartctl by the root helper, judged by Steward, recorded as metrics, listed on the host page. The helper's half reads smartctl's JSON, which is the same on every distribution and for SATA, SAS and NVMe drives; the fixtures are trimmed from real `smartctl --json -a` output (7.2/7.3). """ import importlib.util import json import os import subprocess from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace import pytest from plugins.host_agent import agent as a from plugins.host_agent import drives as dh from tests._templates import render_plugin_template ROOT = Path(__file__).resolve().parents[3] FILES = ROOT / "plugins" / "host_agent" / "files" NVME = {"smartctl": {"version": [7, 2], "exit_status": 0}, "device": {"name": "/dev/nvme0", "type": "nvme", "protocol": "NVMe"}, "model_name": "Samsung SSD 980 PRO 1TB", "user_capacity": {"bytes": 1000204886016}, "smart_status": {"passed": True, "nvme": {"value": 0}}, "temperature": {"current": 41}, "power_on_time": {"hours": 9120}, "nvme_smart_health_information_log": {"critical_warning": 0, "temperature": 41, "available_spare": 100, "available_spare_threshold": 10, "percentage_used": 7, "media_errors": 0}} ATA = {"smartctl": {"version": [7, 3], "exit_status": 0}, "device": {"name": "/dev/sda", "type": "sat", "protocol": "ATA"}, "model_name": "WDC WD40EFRX-68N32N0", "user_capacity": {"bytes": 4000787030016}, "smart_support": {"available": True, "enabled": True}, "smart_status": {"passed": True}, "temperature": {"current": 34}, "ata_smart_attributes": {"table": [ {"id": 5, "name": "Reallocated_Sector_Ct", "raw": {"value": 8}}, {"id": 9, "name": "Power_On_Hours", "raw": {"value": 30211}}, {"id": 197, "name": "Current_Pending_Sector", "raw": {"value": 0}}, {"id": 198, "name": "Offline_Uncorrectable", "raw": {"value": 0}}]}} VIRTUAL = {"smartctl": {"version": [7, 2], "exit_status": 4}, "device": {"name": "/dev/sdb", "type": "scsi", "protocol": "SCSI"}, "scsi_model_name": "QEMU HARDDISK", "smart_support": {"available": False}} GONE = {"smartctl": {"version": [7, 2], "exit_status": 2, "messages": [{"string": "Smartctl open device: /dev/sdc failed: No such device", "severity": "error"}]}} @pytest.fixture def maint(): spec = importlib.util.spec_from_file_location("steward_maint", FILES / "maint.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod # ── the helper reads ───────────────────────────────────────────────────────── def test_an_nvme_drive_s_figures_are_read(maint): got = maint.drive_summary("/dev/nvme0", 0, NVME) assert got == {"device": "nvme0", "protocol": "NVMe", "model": "Samsung SSD 980 PRO 1TB", "capacity_bytes": 1000204886016, "passed": True, "temp_c": 41, "power_on_hours": 9120, "critical_warning": 0, "life_used_pct": 7, "media_errors": 0, "spare_pct": 100, "spare_threshold_pct": 10} def test_a_sata_drive_s_bad_sectors_are_read_from_its_attributes(maint): got = maint.drive_summary("/dev/sda", 0, ATA) assert (got["reallocated"], got["pending"], got["offline_uncorrectable"]) == (8, 0, 0) assert got["passed"] is True and got["temp_c"] == 34 and "life_used_pct" not in got def test_a_drive_with_no_smart_and_one_that_can_t_be_opened_say_so(maint): assert maint.drive_summary("/dev/sdb", 4, VIRTUAL)["available"] is False got = maint.drive_summary("/dev/sdc", 2, GONE) assert got == {"device": "sdc", "error": "Smartctl open device: /dev/sdc failed: No such device"} assert maint.drive_summary("/dev/sdd", None, None)["error"] == "smartctl couldn't read it" @pytest.mark.parametrize("name, kind, device", [ ("/dev/sda", "sat", "sda"), ("/dev/nvme0", "nvme", "nvme0"), ("/dev/bus/0", "megaraid,3", "bus0-megaraid3"), # VDNT-Docker02's PERC (#5806) ("/dev/bus/0", "sat+megaraid,12", "bus0-megaraid12"), ("/dev/sg1", "cciss,0", "sg1-cciss0"), ("/dev/sdc", None, "sdc"), ]) def test_each_drive_has_a_name_of_its_own(maint, name, kind, device): assert maint.drive_name(name, kind) == device def test_disks_behind_one_controller_stay_apart(maint, monkeypatch): def fake(args): if args == ["--scan-open"]: return 0, {"devices": [{"name": "/dev/bus/0", "type": "megaraid,0"}, {"name": "/dev/bus/0", "type": "megaraid,1"}]} return 0, ATA monkeypatch.setattr(maint, "smartctl_version", lambda: (7, 4)) monkeypatch.setattr(maint, "smartctl_json", fake) assert [d["device"] for d in maint.do_drives()["drives"]] == ["bus0-megaraid0", "bus0-megaraid1"] def test_a_name_seen_twice_is_still_two_drives(): """What agent 1.20.0 sent from a PERC: every disk named "0".""" report = {"drives": [{"device": "0", "passed": True}, {"device": "0", "passed": False}, {"device": "0", "passed": True}]} assert sorted(d["device"] for d in dh.drives_of(report)) == ["0", "0#2", "0#3"] rows = {(m, r) for m, r, _ in dh.metric_rows(report, "box") if m == "drive_failing"} assert len(rows) == 3 def test_a_failing_disk_s_figures_still_count(maint): """Exit bit 3 says the disk is failing; its JSON is the point, not an error.""" failing = {**ATA, "smart_status": {"passed": False}} assert maint.drive_summary("/dev/sda", 8, failing)["passed"] is False def test_no_smartctl_or_an_old_one_is_reported_not_parsed(maint, monkeypatch): monkeypatch.setattr(maint, "smartctl_version", lambda: None) assert maint.do_drives()["error"] == "smartctl not installed" monkeypatch.setattr(maint, "smartctl_version", lambda: (6, 6)) got = maint.do_drives() assert got["smartctl"] == "6.6" and "too old" in got["error"] and "drives" not in got def test_the_helper_finds_the_drives_itself(maint, monkeypatch): calls = [] def fake(args): calls.append(args) if args == ["--scan-open"]: return 0, {"devices": [{"name": "/dev/sda", "type": "sat"}, {"name": "/dev/nvme0", "type": "nvme"}, {"name": "not-a-device"}]} return 0, {"/dev/sda": ATA, "/dev/nvme0": NVME}[args[-1]] monkeypatch.setattr(maint, "smartctl_version", lambda: (7, 2)) monkeypatch.setattr(maint, "smartctl_json", fake) got = maint.do_drives() assert [d["device"] for d in got["drives"]] == ["sda", "nvme0"] and got["smartctl"] == "7.2" assert calls[1:] == [["-a", "-d", "sat", "/dev/sda"], ["-a", "-d", "nvme", "/dev/nvme0"]] def test_a_drives_request_writes_the_report_and_no_job_result(maint, tmp_path, monkeypatch): monkeypatch.setattr(maint, "DRIVES", str(tmp_path / "drives.json")) monkeypatch.setattr(maint, "RESULTS", str(tmp_path / "results")) monkeypatch.setattr(maint, "do_drives", lambda: {"checked_at": "t", "drives": []}) req = {"id": "drives-1", "action": "drives"} path = tmp_path / "r.json" path.write_text(json.dumps(req)) assert maint.read_request(str(path)) == req # on the helper's list assert maint.handle(req)["state"] == "done" assert json.loads((tmp_path / "drives.json").read_text()) == {"checked_at": "t", "drives": []} assert oct(os.stat(tmp_path / "drives.json").st_mode & 0o777) == "0o644" assert not (tmp_path / "results").exists() # not a maintenance job # ── the agent asks hourly and carries the answer ───────────────────────────── def _helper(tmp_path): unit, requests = tmp_path / "steward-maint.path", tmp_path / "requests" unit.write_text("[Path]\n") requests.mkdir() return dict(unit=str(unit), requests_dir=str(requests), drives_path=str(tmp_path / "drives.json")) def test_the_agent_asks_at_start_then_hourly(tmp_path): paths, state = _helper(tmp_path), {} a.refresh_drives(state, now_mono=100.0, **paths) assert [json.loads((tmp_path / "requests" / n).read_text())["action"] for n in os.listdir(tmp_path / "requests")] == ["drives"] for n in os.listdir(tmp_path / "requests"): os.unlink(tmp_path / "requests" / n) a.refresh_drives(state, now_mono=100.0 + 3599, **paths) assert os.listdir(tmp_path / "requests") == [] a.refresh_drives(state, now_mono=100.0 + 3600, **paths) assert len(os.listdir(tmp_path / "requests")) == 1 def test_the_agent_carries_the_newest_report(tmp_path): paths, state = _helper(tmp_path), {} a.refresh_drives(state, now_mono=0.0, **paths) assert "drives" not in state # nothing read yet (tmp_path / "drives.json").write_text(json.dumps({"checked_at": "t1", "drives": []})) a.refresh_drives(state, now_mono=1.0, **paths) assert state["drives"]["checked_at"] == "t1" (tmp_path / "drives.json").write_text("{half") # mid-write os.utime(tmp_path / "drives.json", (5, 5)) a.refresh_drives(state, now_mono=2.0, **paths) assert state["drives"]["checked_at"] == "t1" def test_without_the_helper_the_agent_says_what_drive_health_needs(tmp_path): state = {} a.refresh_drives(state, now_mono=0.0, unit=str(tmp_path / "none.path"), requests_dir=str(tmp_path / "none"), drives_path=str(tmp_path / "d.json")) assert "maintenance helper" in state["drives"]["error"] def test_the_agent_and_the_helper_agree_on_the_report_s_place(maint): assert a.DRIVES_PATH == maint.DRIVES and "drives" in a.MAINT_ACTIONS # ── Steward judges ─────────────────────────────────────────────────────────── @pytest.mark.parametrize("drive, state", [ ({"device": "sda", "passed": True, "reallocated": 0}, "ok"), ({"device": "sda", "passed": False}, "failing"), ({"device": "nvme0", "passed": True, "critical_warning": 4}, "failing"), ({"device": "nvme0", "passed": True, "spare_pct": 5, "spare_threshold_pct": 10}, "failing"), ({"device": "nvme0", "passed": True, "life_used_pct": 93}, "worn"), ({"device": "sda", "passed": True, "reallocated": 8}, "watch"), ({"device": "sdb", "available": False}, "unavailable"), ({"device": "sdc", "error": "No such device"}, "unreadable"), ]) def test_each_drive_is_judged(drive, state): assert dh.judge(drive)["state"] == state def test_what_needs_attention_is_listed_first(): report = {"drives": [{"device": "sda", "passed": True}, {"device": "sdb", "available": False}, {"device": "nvme0", "passed": False}, {"device": "sdc", "error": "x"}]} assert [d["device"] for d in dh.drives_of(report)] == ["nvme0", "sdc", "sda", "sdb"] assert dh.drives_of({"error": "smartctl not installed"}) == [] == dh.drives_of(None) def test_each_read_drive_is_a_set_of_metrics_and_the_host_rolls_them_up(): report = {"drives": [{"device": "nvme0", "passed": True, "life_used_pct": 7, "media_errors": 0, "temp_c": 41}, {"device": "sda", "passed": False, "reallocated": 8, "pending": 2}, {"device": "sdb", "available": False}]} rows = {(m, r): v for m, r, v in dh.metric_rows(report, "box")} assert rows == { ("drive_failing", "box:drive:sda"): 1.0, ("drive_bad_sectors", "box:drive:sda"): 10.0, ("drive_failing", "box:drive:nvme0"): 0.0, ("drive_life_used_pct", "box:drive:nvme0"): 7.0, ("drive_bad_sectors", "box:drive:nvme0"): 0.0, ("drive_temp_c", "box:drive:nvme0"): 41.0, ("drives_failing", "box"): 1.0, ("drive_life_used_max_pct", "box"): 7.0} def test_a_host_with_no_drive_read_records_no_roll_up(): """A VM's virtual disks, or smartctl missing: 0 failing would claim a look that never happened.""" assert dh.metric_rows({"drives": [{"device": "vda", "available": False}]}, "box") == [] assert dh.metric_rows({"error": "smartctl not installed"}, "box") == [] def test_every_drive_metric_is_offered_to_rules_with_a_plain_label(): from plugins.host_agent import get_alert_metrics from steward.core.formatting import METRICS, metric_unit assert set(dh.DRIVE_METRICS) <= set(get_alert_metrics()["host_agent"]) assert all(m in METRICS for m in dh.DRIVE_METRICS) assert metric_unit("drive_temp_c") == "°C" and metric_unit("drive_life_used_max_pct") == "%" def test_ingest_records_the_drives_and_keeps_the_report_once(): from plugins.host_agent.routes import _expand_sample_to_metrics, apply_drives now = datetime(2026, 10, 15, tzinfo=timezone.utc) report = {"checked_at": "2026-10-15T00:00:00+00:00", "smartctl": "7.2", "drives": [{"device": "sda", "passed": True}]} rows = _expand_sample_to_metrics({"drives": report}, "box", now) assert {(r.metric_name, r.resource_name) for r in rows} == { ("drive_failing", "box:drive:sda"), ("drives_failing", "box")} reg = SimpleNamespace(drive_health=None, drive_health_at=None) assert apply_drives(reg, {"drives": report}, now) is True assert reg.drive_health == report and reg.drive_health_at == now assert apply_drives(reg, {"drives": dict(report)}, now) is False # unchanged: no write assert apply_drives(reg, {}, now) is False # ── installed on any common distribution ───────────────────────────────────── def test_the_dependency_installer_knows_the_common_package_managers(): script = (FILES / "install-deps.sh").read_text() for manager in ("apt-get", "dnf", "yum", "zypper", "pacman", "apk"): assert f"have {manager}" in script, manager assert "--no-install-recommends" in script # no mail server pulled in on Debian subprocess.run(["sh", "-n", str(FILES / "install-deps.sh")], check=True) def test_the_dependency_installer_never_fails_the_install(tmp_path): """No package manager it knows (PATH holds none): it says so and exits 0.""" bin_dir = tmp_path / "bin" bin_dir.mkdir() done = subprocess.run(["/bin/sh", str(FILES / "install-deps.sh")], capture_output=True, text=True, env={"PATH": str(bin_dir)}) assert done.returncode == 0 assert "WARNING: could not install smartmontools" in done.stdout def test_install_sh_and_every_agent_playbook_run_it(): install = (ROOT / "plugins" / "host_agent" / "templates" / "install.sh.j2").read_text() assert '"$AGENT_DIR/install-deps.sh" || true' in install for name in ("install", "provision", "update"): book = (ROOT / "steward" / "ansible" / "bundled" / "host_agent" / f"{name}.yml").read_text() assert "files/install-deps.sh" in book and "command: /usr/local/lib/steward-agent/install-deps.sh" in book, name # ── the host page ──────────────────────────────────────────────────────────── async def test_the_host_page_lists_each_drive_failing_first(app): report = {"drives": [{"device": "nvme0", "passed": True, "life_used_pct": 7, "temp_c": 41, "model": "Samsung SSD 980 PRO 1TB"}, {"device": "sda", "passed": False, "reallocated": 8, "model": "WDC WD40EFRX"}]} async with app.test_request_context("/plugins/host_agent/h1/metrics"): page = await render_plugin_template( app, "host_agent", "_host_metrics.html", host=SimpleNamespace(address="10.0.0.2"), stale=False, reg=SimpleNamespace(distro=None, kernel=None, arch=None, agent_version=None, last_seen_at=datetime.now(timezone.utc)), hostlvl={"cpu_pct": 5.0}, cores=[], nets={}, disks_io={}, temps={}, mounts={}, drives=dh.drives_of(report), drives_error=None, drives_at=datetime.now(timezone.utc)) page = " ".join(page.split()) assert "Drive health" in page assert page.index('badge-crit">Failing') < page.index('badge-ok">OK') assert "its SMART self-assessment failed" in page and "8 bad sectors" in page assert "7% life used · 0 bad sectors" not in page and "7% life used · 41°C" in page async def test_the_host_page_says_why_drives_weren_t_read(app): async with app.test_request_context("/plugins/host_agent/h1/metrics"): page = await render_plugin_template( app, "host_agent", "_host_metrics.html", host=SimpleNamespace(address="10.0.0.2"), stale=False, reg=SimpleNamespace(distro=None, kernel=None, arch=None, agent_version=None, last_seen_at=datetime.now(timezone.utc)), hostlvl={"cpu_pct": 5.0}, cores=[], nets={}, disks_io={}, temps={}, mounts={}, drives=[], drives_error="smartctl not installed", drives_at=None) assert "Not read: smartctl not installed." in page