bvandeusenandClaude Opus 5.5
dadec4e355
- /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>
39 lines
1.8 KiB
Python
39 lines
1.8 KiB
Python
"""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) == []
|