feat(syslog): a source's lines to read and search, on its page, its host's and its SNMP device's, and over MCP (#6000)
CI / lint (push) Successful in 2s
CI / unit (push) Successful in 1m29s
CI / integration (push) Successful in 1m36s
CI / publish (push) Successful in 1m6s

- /plugins/syslog/sources/<id>: the lines newest first, loading as you
  scroll, filtered by severity and worse, app, text and time, with day
  dividers and times to the second in the reader's zone.
- The Syslog page links each source and shows its lines in the last hour.
- A Syslog panel on a host's page and an SNMP device's: the newest lines of
  the sources bound to it or at its address.
- MCP: list_syslog_sources and search_syslog (source, host, text, severity
  and worse, app, range, paging by next_before).
- Shared: escape_like moves to steward/core/like.py and cut_middle to the
  MCP shaping module, replacing four inline copies; |when takes seconds=.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-19 06:05:52 -04:00
co-authored by Claude Opus 5.5
parent bf6019a2af
commit dadec4e355
27 changed files with 720 additions and 40 deletions
+38
View File
@@ -0,0 +1,38 @@
"""Reading the kept syslog lines (#6000): the filter read from a page's
arguments, severity by name or number, and which sources a host's or a
device's page shows."""
from types import SimpleNamespace as NS
import pytest
from werkzeug.datastructures import MultiDict
from plugins.syslog.service import DEFAULT_RANGE, line_filter, severity_level, sources_for
def test_a_severity_reads_by_name_alias_or_number():
assert severity_level("err") == 3 and severity_level("error") == 3 and severity_level("3") == 3
assert severity_level("Warning") == 4 and severity_level("warn") == 4
assert severity_level("") is None and severity_level(None) is None
with pytest.raises(ValueError, match="0-7"):
severity_level("loud")
def test_a_page_filter_drops_what_does_not_read():
f = line_filter(MultiDict({"severity": "loud", "range": "9y", "q": " nfsd ", "app": "kernel"}))
assert (f.severity, f.range, f.text, f.app) == (None, DEFAULT_RANGE, "nfsd", "kernel")
assert f.args() == {"q": "nfsd", "app": "kernel"}
assert line_filter(MultiDict({"severity": "4", "range": "7d"})).args() == {"severity": "4", "range": "7d"}
def _src(sid, state="accepted", host_id=None, address="10.0.0.5"):
return NS(id=sid, state=state, host_id=host_id, address=address)
def test_a_page_shows_sources_bound_to_its_host_or_unbound_at_its_address():
sources = [_src("bound", host_id="h1", address="10.9.9.9"), _src("by-address"),
_src("elsewhere", host_id="h2"), _src("offered", state="offered"),
_src("other", address="10.0.0.6")]
assert [s.id for s in sources_for(sources, "h1", "10.0.0.5")] == ["bound", "by-address"]
# An SNMP device bound to no host matches by its address alone.
assert [s.id for s in sources_for(sources, None, "10.0.0.5")] == ["by-address"]
assert sources_for(sources, None, None) == []