"""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()