bvandeusenandClaude Opus 5.5
0742b7a2b7
#5061 "Metric catalog belongs to what emits: plugins register their metrics; names match what is recorded". - The static METRIC_CATALOG is gone. metric_catalog() is core's own sources (the four monitor types and "steward", the app's open_fds/open_fds_pct, which were recorded but never offered) plus each loaded plugin's get_alert_metrics() export. A disabled plugin's sources leave the picker. - traefik offers what it records: error_rate_4xx_pct and error_rate_5xx_pct. The old "error_rate" was never recorded, so a rule on it could not fire. - tests/core/test_metric_catalog.py walks every MetricBatch.add / record_metric call (traefik's tuple loop included) and fails on a recorded metric the catalog lacks; it shows on a synthetic writer that it can fail. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
105 lines
4.6 KiB
Python
105 lines
4.6 KiB
Python
"""The monitoring-config write tools: their validation refusals and their roles.
|
|
|
|
Monitor-config validation is covered in tests/core/test_monitors.py (the form
|
|
and the tools share it). The writes themselves, their audit rows and the
|
|
rename warning run against Postgres in
|
|
tests/integration/test_mcp_monitoring_writes.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from steward.alerts import service as alert_svc
|
|
from steward.mcp.tools import alerts
|
|
from steward.models.alerts import AlertOperator
|
|
from steward.models.users import UserRole
|
|
|
|
NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)
|
|
|
|
OPERATOR_TOOLS = {"create_monitor", "update_monitor", "set_monitor_enabled", "create_alert_rule",
|
|
"update_alert_rule", "acknowledge_alert", "create_maintenance_window",
|
|
"end_maintenance_window"}
|
|
ADMIN_TOOLS = {"delete_monitor", "delete_alert_rule"}
|
|
|
|
|
|
@pytest.fixture
|
|
def known(monkeypatch):
|
|
"""A source with one resource and its catalog metrics, without a database:
|
|
the host_agent plugin is loaded (its get_alert_metrics hook writes the
|
|
catalog) and has reported one host."""
|
|
import plugins.host_agent
|
|
from steward.core import plugin_manager
|
|
monkeypatch.setattr(plugin_manager, "get_loaded_plugin_modules",
|
|
lambda: {"host_agent": plugins.host_agent})
|
|
|
|
async def resources(db, src):
|
|
return ["nas"] if src == "host_agent" else []
|
|
monkeypatch.setattr(alert_svc, "resources_for", resources)
|
|
|
|
|
|
async def _fields(**over):
|
|
args = dict(name="cpu hot", source_module="host_agent", resource_name="nas",
|
|
metric_name="cpu_pct", operator=">", threshold="90", consecutive_failures_required=2)
|
|
args.update(over)
|
|
return await alert_svc.rule_fields(None, **args)
|
|
|
|
|
|
async def test_rule_fields_accepts_the_form_choices(known):
|
|
f = await _fields()
|
|
assert f["operator"] is AlertOperator.gt and f["threshold"] == 90.0
|
|
assert f["consecutive_failures_required"] == 2
|
|
|
|
|
|
@pytest.mark.parametrize("over, why", [
|
|
({"name": " "}, "needs a name"),
|
|
({"source_module": "zabbix"}, "'zabbix' is not one of icmp, tcp"),
|
|
({"resource_name": "web01"}, "host_agent has no resource 'web01'. Known: nas"),
|
|
({"metric_name": "cpu"}, "does not report 'cpu'. Its metrics: cpu_pct"),
|
|
({"operator": "=>"}, "'=>' is not one of >, <"),
|
|
({"threshold": "high"}, "threshold must be a number"),
|
|
({"consecutive_failures_required": 0}, "at least 1"),
|
|
])
|
|
async def test_rule_fields_refuses_with_a_reason(known, over, why):
|
|
with pytest.raises(ValueError, match=why):
|
|
await _fields(**over)
|
|
|
|
|
|
def test_window_fields():
|
|
f = alert_svc.window_fields(name="patching", start_at=NOW, end_at=NOW + timedelta(hours=1),
|
|
scope_source="", scope_resource="nas")
|
|
assert f["scope_source"] is None and f["scope_resource"] is None # resource needs a source
|
|
with pytest.raises(ValueError, match="must end after it starts"):
|
|
alert_svc.window_fields(name="x", start_at=NOW, end_at=NOW, scope_source=None,
|
|
scope_resource=None)
|
|
with pytest.raises(ValueError, match="'nope' is not one of"):
|
|
alert_svc.window_fields(name="x", start_at=NOW, end_at=NOW + timedelta(minutes=1),
|
|
scope_source="nope", scope_resource=None)
|
|
|
|
|
|
async def test_maintenance_window_needs_exactly_one_end():
|
|
with pytest.raises(ValueError, match="exactly one of duration_minutes or end_at"):
|
|
await alerts.create_maintenance_window("x")
|
|
with pytest.raises(ValueError, match="exactly one of duration_minutes or end_at"):
|
|
await alerts.create_maintenance_window("x", duration_minutes=30, end_at="2026-10-06T13:00:00Z")
|
|
with pytest.raises(ValueError, match="not an ISO-8601 time"):
|
|
await alerts.create_maintenance_window("x", end_at="tonight")
|
|
|
|
|
|
def test_when_reads_a_zoneless_time_as_utc():
|
|
assert alerts._when("2026-10-06T22:00:00", "end_at") == datetime(2026, 10, 6, 22, tzinfo=timezone.utc)
|
|
assert alerts._when("2026-10-06T22:00:00Z", "end_at").utcoffset() == timedelta(0)
|
|
|
|
|
|
def test_write_tools_are_registered_with_their_roles(app):
|
|
mcp = app.mcp_server
|
|
for name in OPERATOR_TOOLS:
|
|
assert mcp.tool_roles[name] is UserRole.operator, name
|
|
assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is False, name
|
|
for name in ADMIN_TOOLS:
|
|
assert mcp.tool_roles[name] is UserRole.admin, name
|
|
assert mcp.tool_roles["get_alert_rule_options"] is UserRole.viewer
|
|
for name in OPERATOR_TOOLS | ADMIN_TOOLS | {"get_alert_rule_options"}:
|
|
assert (mcp._tool_manager.get_tool(name).description or "").strip(), name
|