bvandeusenandClaude Opus 5.5
d20d78498e
Monitors, schedules and maintenance plans already pause with a Pause/Resume button and say "Paused" where they're listed. Prometheus targets and syslog patterns only had a tick box inside the edit form, so a target refused by its endpoint couldn't be stopped from its page. - Both get a Pause/Resume button (target page and list; patterns list) posting to /toggle, recorded in the audit log, and a Paused badge. - The forms lose the tick box. A new one starts running; an edit keeps whether it is paused. - A target's "off" state is now "paused", in its pages and MCP tools. A resumed target reads "not scraped yet" until its next scrape, and its counter rates start over. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
58 lines
2.7 KiB
Python
58 lines
2.7 KiB
Python
"""Syslog pattern alerts (#6001): a pattern read from its form, the line
|
|
test it makes, its metric key, and the per-minute tally its count keeps."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
import pytest
|
|
from werkzeug.datastructures import MultiDict
|
|
|
|
from plugins.syslog.patterns import PRESETS, PatternError, Tally, key_for, matches, parse_form
|
|
|
|
|
|
def test_a_form_reads_into_a_pattern():
|
|
v = parse_form(MultiDict({"name": " NFS ", "phrases": "nfsd:\n\n lockd: \n", "max_severity": "3",
|
|
"source_id": "", "window_minutes": "60"}))
|
|
assert v == {"name": "NFS", "phrases": ["nfsd:", "lockd:"], "max_severity": 3, "source_id": None,
|
|
"window_minutes": 60} # paused or not is the toggle's, not the form's (#6419)
|
|
# A window or severity not offered falls back rather than failing.
|
|
v = parse_form(MultiDict({"name": "x", "phrases": "a", "max_severity": "9", "window_minutes": "7"}))
|
|
assert (v["max_severity"], v["window_minutes"]) == (None, 15)
|
|
|
|
|
|
@pytest.mark.parametrize("form, says", [
|
|
({"phrases": "a"}, "name"),
|
|
({"name": "x", "phrases": " \n "}, "at least one phrase"),
|
|
({"name": "x", "phrases": "\n".join(str(i) for i in range(21))}, "At most 20"),
|
|
({"name": "x", "phrases": "a" * 201}, "200 characters"),
|
|
])
|
|
def test_a_form_that_does_not_read_says_what_to_fix(form, says):
|
|
with pytest.raises(PatternError, match=says):
|
|
parse_form(MultiDict(form))
|
|
|
|
|
|
def test_a_line_matches_any_phrase_in_any_case_at_the_severity():
|
|
p = NS(phrases=["I/O error", "medium error"], max_severity=None)
|
|
assert matches(p, 6, "blk_update_request: i/o ERROR, dev sda")
|
|
assert not matches(p, 6, "all well")
|
|
sev = NS(phrases=[], max_severity=3)
|
|
assert matches(sev, 3, "anything") and matches(sev, 0, "") and not matches(sev, 4, "anything")
|
|
both = NS(phrases=["nfsd"], max_severity=3)
|
|
assert matches(both, 2, "nfsd: too many") and not matches(both, 5, "nfsd: too many")
|
|
|
|
|
|
def test_a_key_is_the_name_in_metric_form():
|
|
assert key_for("Disk I/O errors") == "disk_i_o_errors"
|
|
assert key_for(" --- ") == "pattern" and len(key_for("x" * 90)) == 48
|
|
assert len({p.key for p in PRESETS}) == len(PRESETS)
|
|
assert all(p.phrases or p.max_severity is not None for p in PRESETS)
|
|
|
|
|
|
def test_a_tally_counts_its_window_by_the_minute_and_forgets_older():
|
|
now = datetime(2026, 10, 19, 12, 30, 40, tzinfo=timezone.utc)
|
|
t = Tally(sig=())
|
|
for minutes in (0, 5, 14, 16, 120):
|
|
t.add("s1", now - timedelta(minutes=minutes))
|
|
assert t.count("s1", now - timedelta(minutes=15)) == 3
|
|
assert t.count("s1", now - timedelta(hours=3)) == 3 # what was dropped stays dropped
|
|
assert t.count("s2", now) == 0
|