0.4.0: Add match clustering

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