diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_generate_matches.py | 226 |
1 files changed, 226 insertions, 0 deletions
diff --git a/tests/test_generate_matches.py b/tests/test_generate_matches.py new file mode 100644 index 0000000..f9dd828 --- /dev/null +++ b/tests/test_generate_matches.py @@ -0,0 +1,226 @@ +"""Tests for generate_matches clustering logic.""" + +from __future__ import annotations + +import re +import unittest + +from pylogsentinel.core import ( + MatchEvent, + Rule, + generate_matches, +) + + +def _make_rule( + rule_id: str = "test", + pattern: str = "error", + flags: int = re.IGNORECASE, + base_paths: list[str] | None = None, +) -> Rule: + """Helper to build a Rule for testing.""" + if base_paths is None: + base_paths = ["/var/log"] + return Rule( + rule_id=rule_id, + pattern=f"/{pattern}/i", + description="test rule", + action_id="default", + compiled=re.compile(pattern, flags), + log_set_ids=["default"], + log_base_paths=base_paths, + ) + + +FILE_PATH = "/var/log/app.log" + + +class TestSingleMatch(unittest.TestCase): + """A single match produces one event with surrounding context.""" + + def test_single_match_mid_file(self): + lines = [f"line {i}\n" for i in range(20)] + lines[10] = "ERROR happened here\n" + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + ev = events[0] + self.assertEqual(ev.line_number, 11) # 1-based + self.assertEqual(ev.match_count, 1) + self.assertEqual(ev.matched_lines, "11") + # Context: lines 7-13 (idx 7 to 13 inclusive) + expected_ctx = "".join(lines[7:14]) + self.assertEqual(ev.context, expected_ctx) + + def test_single_match_near_start(self): + lines = [f"line {i}\n" for i in range(10)] + lines[1] = "ERROR at line 1\n" + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + ev = events[0] + # Context start clamped to 0 + expected_ctx = "".join(lines[0:5]) + self.assertEqual(ev.context, expected_ctx) + + def test_single_match_near_end(self): + lines = [f"line {i}\n" for i in range(10)] + lines[9] = "ERROR at the end\n" + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + ev = events[0] + # Context end clamped to last line + expected_ctx = "".join(lines[6:10]) + self.assertEqual(ev.context, expected_ctx) + + +class TestClusteringMerge(unittest.TestCase): + """Matches within context_radius lines of each other merge.""" + + def test_two_matches_within_radius(self): + """Matches 2 lines apart (gap=1) with radius=3 → one cluster.""" + lines = [f"line {i}\n" for i in range(20)] + lines[8] = "ERROR first\n" + lines[10] = "ERROR second\n" + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + ev = events[0] + self.assertEqual(ev.match_count, 2) + self.assertEqual(ev.matched_lines, "9,11") + self.assertEqual(ev.line_number, 9) # first match + # Context: idx 5 to 13 + expected_ctx = "".join(lines[5:14]) + self.assertEqual(ev.context, expected_ctx) + + def test_three_matches_chained(self): + """Three matches each within radius of the previous → one cluster.""" + lines = [f"line {i}\n" for i in range(30)] + lines[5] = "ERROR one\n" + lines[8] = "ERROR two\n" + lines[11] = "ERROR three\n" + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + ev = events[0] + self.assertEqual(ev.match_count, 3) + self.assertEqual(ev.matched_lines, "6,9,12") + # Context: idx 2 to 14 + expected_ctx = "".join(lines[2:15]) + self.assertEqual(ev.context, expected_ctx) + + +class TestClusteringSplit(unittest.TestCase): + """Matches far apart produce separate events.""" + + def test_two_matches_beyond_radius(self): + """Gap of 5 with radius=3 → two clusters.""" + lines = [f"line {i}\n" for i in range(30)] + lines[5] = "ERROR first\n" + lines[11] = "ERROR second\n" # gap = 5 lines (idx 6-10) + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 2) + self.assertEqual(events[0].match_count, 1) + self.assertEqual(events[0].line_number, 6) + self.assertEqual(events[1].match_count, 1) + self.assertEqual(events[1].line_number, 12) + + def test_gap_exactly_at_radius(self): + """Gap == radius → merged (gap <= context_radius).""" + lines = [f"line {i}\n" for i in range(20)] + lines[5] = "ERROR first\n" + lines[9] = "ERROR second\n" # gap = 3 (idx 6,7,8) + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].match_count, 2) + + def test_gap_one_beyond_radius(self): + """Gap == radius + 1 → split.""" + lines = [f"line {i}\n" for i in range(20)] + lines[5] = "ERROR first\n" + lines[10] = "ERROR second\n" # gap = 4 (idx 6,7,8,9) + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 2) + + +class TestMultipleRules(unittest.TestCase): + """Different rules produce independent clusters even on same lines.""" + + def test_two_rules_same_line(self): + lines = [f"line {i}\n" for i in range(10)] + lines[5] = "ERROR WARN something\n" + rule_error = _make_rule(rule_id="error", pattern="ERROR") + rule_warn = _make_rule(rule_id="warn", pattern="WARN") + rules = {"error": rule_error, "warn": rule_warn} + events = list( + generate_matches(FILE_PATH, lines, 0, rules, context_radius=2) + ) + self.assertEqual(len(events), 2) + rule_ids = {ev.rule.rule_id for ev in events} + self.assertEqual(rule_ids, {"error", "warn"}) + + +class TestRuleNotApplicable(unittest.TestCase): + """Rule with non-matching base paths produces no events.""" + + def test_rule_does_not_apply(self): + lines = ["ERROR something\n"] + rule = _make_rule(base_paths=["/other/path"]) + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 0) + + +class TestInitialLineNumber(unittest.TestCase): + """Verify absolute line numbers use initial_line_number offset.""" + + def test_offset_line_numbers(self): + lines = ["ERROR line\n"] + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 100, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + # initial_line_number=100 means 100 lines already processed + # idx=0 → absolute = 100 + 0 + 1 = 101 + self.assertEqual(events[0].line_number, 101) + + +class TestMatchedText(unittest.TestCase): + """matched_text field contains the first match's text.""" + + def test_matched_text_first_in_cluster(self): + lines = [f"line {i}\n" for i in range(10)] + lines[3] = "first ERROR here\n" + lines[5] = "second ERROR there\n" + rule = _make_rule() + events = list( + generate_matches(FILE_PATH, lines, 0, {"test": rule}, context_radius=3) + ) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].matched_text, "ERROR") + + +if __name__ == "__main__": + unittest.main() |