From 467ab1f64c38c0d8fa022fd912e2535597b0af9b Mon Sep 17 00:00:00 2001 From: Gulliby-development <20404543+Gulliby@users.noreply.github.com> Date: Tue, 13 Oct 2026 03:01:19 +0200 Subject: [PATCH] Add -v / --invert-match to keep records whose message does not match --grep --- README.md | 14 +++ src/logslice/cli.py | 8 ++ src/logslice/filters.py | 9 +- tests/test_invert_match.py | 227 +++++++++++++++++++++++++++++++++++++ 4 files changed, 256 insertions(+), 2 deletions(-) create mode 100644 tests/test_invert_match.py diff --git a/README.md b/README.md index 273227e..d5a5f22 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,19 @@ logslice app.log --grep Disk The pattern is a Python regular expression searched in the message only, not in the timestamp or level. +Drop records by message instead: add `-v` (`--invert-match`) to `--grep` to keep only the records whose message does not match. + +```sh +logslice app.log --grep Disk -v +``` + +```text +2026-10-10 12:00:00 INFO Started +2026-10-10T12:00:00.123 warn Slow response +``` + +`-v` needs `--grep` (otherwise it is an error, exit status `2`) and works with `-i`, the time bounds, `--level` and `--summary`; a record with an empty message is kept unless the pattern matches the empty string. + Read stdin with no file arguments or with `-`. On macOS/Linux: ```sh @@ -177,6 +190,7 @@ have been written when a later file fails. | `--level LEVEL` | Minimum level: `DEBUG`, `INFO`, `WARNING` (or `WARN`), `ERROR`, `CRITICAL`; case-insensitive (default `DEBUG`). | | `--grep PATTERN` | Keep only records whose message matches the Python regular expression `PATTERN` (searched in the message only). | | `-i`, `--ignore-case` | Make `--grep` case-insensitive. | +| `-v`, `--invert-match` | With `--grep`, keep only records whose message does not match; an error without `--grep`. | | `--summary` | Print level counts, the top 5 ERROR/CRITICAL messages and the skipped line count instead of records. | | `--summary-format {text,json}` | Output format of `--summary`, default `text`; `json` prints the level counts, the top 5 error messages and the skipped line count as a JSON object. | | `-h`, `--help` | Show help. | diff --git a/src/logslice/cli.py b/src/logslice/cli.py index 9c8642e..24ddecb 100644 --- a/src/logslice/cli.py +++ b/src/logslice/cli.py @@ -47,6 +47,10 @@ def build_parser() -> argparse.ArgumentParser: "-i", "--ignore-case", action="store_true", help="make --grep case-insensitive", ) + parser.add_argument( + "-v", "--invert-match", action="store_true", + help="keep only records whose message does not match --grep", + ) parser.add_argument( "--summary", action="store_true", help="print level counts, top 5 ERROR/CRITICAL messages, and skipped lines", @@ -73,6 +77,9 @@ def main(argv: list[str] | None = None) -> int: if args.summary_format == "json" and not args.summary: print("logslice: --summary-format requires --summary", file=sys.stderr) return 2 + if args.invert_match and args.grep is None: + print("logslice: --invert-match requires --grep", file=sys.stderr) + return 2 pattern = None if args.grep is not None: try: @@ -102,6 +109,7 @@ def main(argv: list[str] | None = None) -> int: until=args.until, level=args.level, pattern=pattern, + invert_match=args.invert_match, ): continue if args.summary: diff --git a/src/logslice/filters.py b/src/logslice/filters.py index 2f27670..bccff73 100644 --- a/src/logslice/filters.py +++ b/src/logslice/filters.py @@ -47,11 +47,13 @@ def matches( until: datetime | None = None, level: Level = Level.DEBUG, pattern: re.Pattern[str] | None = None, + invert_match: bool = False, ) -> bool: """Return whether a record meets both inclusive bounds and minimum severity. If pattern is given, it must also be found (re.search) in record.message - only; the timestamp and level are not searched. None means no message filter. + only; the timestamp and level are not searched. invert_match reverses this + pattern check and has no effect when pattern is None. Omitted bounds are unrestricted. A reversed range matches no records. Timestamps and bounds are timezone-naive, as produced by the parsers. @@ -60,5 +62,8 @@ def matches( (since is None or record.timestamp >= since) and (until is None or record.timestamp <= until) and record.level >= level - and (pattern is None or pattern.search(record.message) is not None) + and ( + pattern is None + or (pattern.search(record.message) is not None) != invert_match + ) ) diff --git a/tests/test_invert_match.py b/tests/test_invert_match.py new file mode 100644 index 0000000..2a89688 --- /dev/null +++ b/tests/test_invert_match.py @@ -0,0 +1,227 @@ +"""Tests for inverted message matching through the filter and CLI.""" + +import json +import re +from datetime import datetime + +import pytest + +from logslice.cli import main +from logslice.filters import matches +from logslice.parser import Level, Record + +LOG_CONTENT = ( + "2026-10-10 12:00:00 INFO Started\n" + "2026-10-10T12:00:00.123 warn Slow response\n" + "2026-10-10 12:01:00 ERROR Disk full\n" + "2026-10-10 12:02:00 ERROR\n" + "not a log line\n" + "2026-10-10 12:03:00 error disk quota\n" +) + + +def write_log(tmp_path): + path = tmp_path / "app.log" + path.write_text(LOG_CONTENT, encoding="utf-8") + return path + + +def run_cli(capsys, *args): + exit_code = main([str(arg) for arg in args]) + captured = capsys.readouterr() + return exit_code, captured.out, captured.err + + +def test_matches_inverts_pattern_result_and_ignores_inversion_without_pattern(): + pattern = re.compile("Disk") + matching = Record(datetime(2026, 10, 10), Level.INFO, "Disk full", "raw") + nonmatching = Record(datetime(2026, 10, 10), Level.INFO, "Started", "raw") + + assert matches(matching, pattern=pattern, invert_match=True) is False + assert matches(nonmatching, pattern=pattern, invert_match=True) is True + assert matches(nonmatching, invert_match=True) is True + + +@pytest.mark.parametrize( + "option", + ["-v", "--invert-match"], +) +def test_invert_match_keeps_records_without_pattern_match(tmp_path, capsys, option): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "--grep", "Disk", option) + + assert exit_code == 0 + assert stderr == "" + assert stdout == ( + "2026-10-10 12:00:00 INFO Started\n" + "2026-10-10T12:00:00.123 warn Slow response\n" + "2026-10-10 12:02:00 ERROR\n" + "2026-10-10 12:03:00 error disk quota\n" + ) + + +@pytest.mark.parametrize( + "args", + [ + ("--grep", "disk", "-i", "-v"), + ("-vi", "--grep", "disk"), + ], +) +def test_case_insensitive_inversion_supports_combined_short_flags( + tmp_path, capsys, args +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, *args) + + assert exit_code == 0 + assert stderr == "" + assert stdout == ( + "2026-10-10 12:00:00 INFO Started\n" + "2026-10-10T12:00:00.123 warn Slow response\n" + "2026-10-10 12:02:00 ERROR\n" + ) + + +@pytest.mark.parametrize( + ("pattern", "expected"), + [ + (".", "2026-10-10 12:02:00 ERROR\n"), + ( + "^$", + ( + "2026-10-10 12:00:00 INFO Started\n" + "2026-10-10T12:00:00.123 warn Slow response\n" + "2026-10-10 12:01:00 ERROR Disk full\n" + "2026-10-10 12:03:00 error disk quota\n" + ), + ), + ], +) +def test_inversion_uses_regex_semantics_for_empty_messages( + tmp_path, capsys, pattern, expected +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "--grep", pattern, "-v") + + assert exit_code == 0 + assert stderr == "" + assert stdout == expected + + +@pytest.mark.parametrize( + "filters", + [ + ("--level", "error", "--grep", "Disk"), + ("--grep", "Disk", "--since", "2026-10-10T12:00:01"), + ], +) +def test_inversion_remains_conjunctive_with_other_filters( + tmp_path, capsys, filters +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "-v", *filters) + + assert exit_code == 0 + assert stderr == "" + assert stdout == ( + "2026-10-10 12:02:00 ERROR\n" + "2026-10-10 12:03:00 error disk quota\n" + ) + + +def test_text_summary_counts_only_retained_records_and_skipped_lines(tmp_path, capsys): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli( + capsys, path, "-v", "--grep", "Disk", "--summary" + ) + + assert exit_code == 0 + assert stderr == "" + assert stdout == ( + "Level counts:\n" + "DEBUG: 0\n" + "INFO: 1\n" + "WARNING: 1\n" + "ERROR: 2\n" + "CRITICAL: 0\n" + "Top 5 error messages:\n" + " 1 \n" + " 1 disk quota\n" + "Skipped lines: 1\n" + ) + + +def test_json_summary_counts_only_retained_records_and_skipped_lines( + tmp_path, capsys +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli( + capsys, path, "-v", "--grep", "Disk", "--summary", "--summary-format", "json" + ) + + assert exit_code == 0 + assert stderr == "" + summary = json.loads(stdout) + assert summary["levels"]["ERROR"] == 2 + assert summary["top_errors"] == [ + {"count": 1, "message": ""}, + {"count": 1, "message": "disk quota"}, + ] + assert summary["skipped_lines"] == 1 + + +@pytest.mark.parametrize("missing", [False, True]) +def test_invert_match_without_grep_fails_before_reading_input( + tmp_path, capsys, missing +): + path = tmp_path / "missing.log" if missing else write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "-v") + + assert exit_code == 2 + assert stdout == "" + assert stderr == "logslice: --invert-match requires --grep\n" + + +def test_summary_format_validation_precedes_invert_match_validation( + tmp_path, capsys +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli( + capsys, path, "-v", "--summary-format", "json" + ) + + assert exit_code == 2 + assert stdout == "" + assert stderr == "logslice: --summary-format requires --summary\n" + + +def test_invalid_grep_pattern_is_checked_after_invert_validation(tmp_path, capsys): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "-v", "--grep", "(") + + assert exit_code == 2 + assert stdout == "" + assert stderr.startswith("logslice: invalid --grep pattern '('") + assert stderr.count("\n") == 1 + + +def test_grep_without_value_keeps_argparse_missing_value_error(tmp_path, capsys): + path = write_log(tmp_path) + + with pytest.raises(SystemExit) as error: + main([str(path), "--grep", "-v"]) + + captured = capsys.readouterr() + assert error.value.code == 2 + assert captured.out == "" + assert "usage:" in captured.err + assert "argument --grep: expected one argument" in captured.err