bvandeusenandClaude Opus 5.5
38ae99fb8e
Agent 1.27.0 sends each temperature sensor's hwmon chip. Steward keeps the chip per host and label beside the reading, so a sensor's metric key stays its label, and the strip and the heat alert name it as "NVMe · Sensor 2" rather than a bare "Sensor 2". get_thresholds (viewer) and set_thresholds (admin, audited) read and change the warn/crit cutoffs CPU, memory, disk, load and temperature are graded by, as Settings → Thresholds does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
"""get_thresholds / set_thresholds: the cutoffs host readings are graded by,
|
|
read and changed as Settings → Thresholds does."""
|
|
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace as NS
|
|
|
|
import pytest
|
|
|
|
from steward.mcp.tools import thresholds
|
|
from steward.models.users import UserRole
|
|
|
|
|
|
@pytest.fixture
|
|
def store(monkeypatch):
|
|
"""The settings table, the app and the audit log, in memory."""
|
|
saved: dict = {"thresholds.temp_crit": 85}
|
|
app = NS(config={})
|
|
audited: list = []
|
|
|
|
class _Db:
|
|
@asynccontextmanager
|
|
async def begin(self):
|
|
yield
|
|
|
|
@asynccontextmanager
|
|
async def db_session():
|
|
yield _Db()
|
|
|
|
async def get_all_settings(_db):
|
|
return dict(saved)
|
|
|
|
async def set_setting(_db, key, value):
|
|
saved[key] = value
|
|
|
|
async def audit_mcp(*args):
|
|
audited.append(args)
|
|
|
|
monkeypatch.setattr(thresholds, "db_session", db_session)
|
|
monkeypatch.setattr(thresholds, "get_all_settings", get_all_settings)
|
|
monkeypatch.setattr(thresholds, "set_setting", set_setting)
|
|
monkeypatch.setattr(thresholds, "mcp_app", lambda: app)
|
|
monkeypatch.setattr(thresholds, "audit_mcp", audit_mcp)
|
|
return NS(saved=saved, app=app, audited=audited)
|
|
|
|
|
|
async def test_the_cutoffs_read_with_their_units_and_defaults(store):
|
|
got = await thresholds.get_thresholds()
|
|
assert set(got) == {"cpu", "mem", "disk", "load", "temp"}
|
|
assert got["temp"] == {"warn": 70, "crit": 85, "unit": "°C", "measures": "hottest real thermometer",
|
|
"default": {"warn": 70, "crit": 85}}
|
|
|
|
|
|
async def test_raising_one_cutoff_keeps_the_other_and_grades_by_it_at_once(store):
|
|
got = await thresholds.set_thresholds({"temp": {"crit": 90}})
|
|
assert got["temp"]["crit"] == 90 and got["temp"]["warn"] == 70
|
|
assert store.saved == {"thresholds.temp_crit": 90} # warn isn't written
|
|
assert store.app.config["THRESHOLDS"]["temp"]["crit"] == 90
|
|
(_action, _kind, _id, detail), = store.audited
|
|
assert detail["changes"] == {"thresholds.temp_crit": {"from": 85, "to": 90}}
|
|
|
|
|
|
@pytest.mark.parametrize("cutoffs, refusal", [
|
|
({}, "Nothing to change"),
|
|
({"fan": {"crit": 1}}, "No reading named fan"),
|
|
({"temp": {"high": 90}}, "warn and/or crit"),
|
|
({"temp": {}}, "warn and/or crit"),
|
|
({"temp": {"crit": "hot"}}, "must be a number"),
|
|
({"temp": {"crit": 0}}, "more than 0"),
|
|
({"temp": {"crit": 60}}, r"warn \(70\) must be below its crit \(60\)"),
|
|
])
|
|
async def test_refusals_change_nothing(store, cutoffs, refusal):
|
|
with pytest.raises(ValueError, match=refusal):
|
|
await thresholds.set_thresholds(cutoffs)
|
|
assert store.saved == {"thresholds.temp_crit": 85} and not store.audited
|
|
|
|
|
|
def test_the_tools_are_registered_with_their_roles(app):
|
|
mcp = app.mcp_server
|
|
for name, role, read_only in (("get_thresholds", UserRole.viewer, True),
|
|
("set_thresholds", UserRole.admin, False)):
|
|
assert mcp.tool_roles[name] is role, name
|
|
assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is read_only, name
|