bvandeusenandClaude Opus 5.5
eee8469a17
The snmp plugin registers list_snmp_devices and get_snmp_device; the prometheus plugin registers list_prometheus_targets and get_prometheus_target. Both are viewer-level, read what the poller and scraper stored, and exist only while their plugin is enabled. get_overview names a device or target whose last poll or scrape failed; stale, pending and off are left out. A device's community and v3 keys and a target's token are never returned, and a password in a target URL is masked. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
48 lines
2.4 KiB
Python
48 lines
2.4 KiB
Python
"""The prometheus plugin's MCP tools (#5840): which targets the overview
|
|
names, what a target row carries (never its token or a password in its URL),
|
|
and the tools it registers."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
import plugins.prometheus as prometheus_plugin
|
|
from plugins.prometheus import mcp_tools as t
|
|
from steward.models.users import UserRole
|
|
|
|
NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _target(name, *, enabled=True, scraped=timedelta(seconds=30), error=None,
|
|
url="http://archives:8080/metrics"):
|
|
return NS(id=f"id-{name}", name=name, url=url, token="bearer-s3cret", enabled=enabled,
|
|
interval_seconds=60, last_scraped_at=NOW - scraped if scraped else None,
|
|
last_ok_at=NOW - timedelta(hours=1), last_error=error, last_kept=120, last_dropped=30,
|
|
last_capped=0, series_cap=1000, keep=["archives_*"], drop=[])
|
|
|
|
|
|
def test_only_a_target_whose_scrape_failed_is_an_overview_finding():
|
|
views = [t._target_view(x, {}, NOW) | {"_ok_at": x.last_ok_at} for x in (
|
|
_target("archives", error="connection refused"), _target("fine"),
|
|
_target("off", enabled=False, error="connection refused"),
|
|
_target("stale", scraped=timedelta(days=1), error="connection refused"),
|
|
_target("new", scraped=None))]
|
|
[f] = t.target_findings(views)
|
|
assert f["kind"] == "prometheus_target_down" and f["subject"] == "archives"
|
|
assert "connection refused" in f["summary"] and f["next"] == 'get_prometheus_target(target="archives")'
|
|
|
|
|
|
def test_a_target_row_never_carries_its_token_or_url_password():
|
|
row = t._target_view(_target("archives", url="http://scrape:hunter2@archives:8080/metrics"),
|
|
{"scrape_ms": 42.0}, NOW)
|
|
text = repr(row)
|
|
assert "bearer-s3cret" not in text and "hunter2" not in text
|
|
assert row["url"] == "http://scrape:***@archives:8080/metrics"
|
|
assert row["scrape_ms"] == 42.0 and row["series_kept"] == 120 and row["reach"] == "up"
|
|
|
|
|
|
def test_the_plugin_registers_its_tools_for_viewers():
|
|
got = {}
|
|
prometheus_plugin.register_mcp_tools(lambda fn, role, read_only: got.update({fn.__name__: (role, read_only)}))
|
|
assert got == {name: (UserRole.viewer, True)
|
|
for name in ("list_prometheus_targets", "get_prometheus_target")}
|
|
assert callable(prometheus_plugin.mcp_overview)
|