From b36fedb53dfb2ab5a9ca54600b2f070bce854955 Mon Sep 17 00:00:00 2001 From: Gulliby-development <20404543+Gulliby@users.noreply.github.com> Date: Tue, 13 Oct 2026 07:49:58 +0200 Subject: [PATCH] Add --top N to --summary --top N sets how many of the most frequent ERROR and CRITICAL messages the summary lists (1 to 100, default 5). --- README.md | 32 +++++- src/logslice/cli.py | 16 ++- src/logslice/filters.py | 14 +++ src/logslice/summary.py | 20 ++-- tests/test_cli.py | 2 +- tests/test_summary.py | 7 ++ tests/test_top.py | 221 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 298 insertions(+), 14 deletions(-) create mode 100644 tests/test_top.py diff --git a/README.md b/README.md index d5a5f22..b8969f9 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,33 @@ For the three example log lines above, this prints: The object has the keys `levels`, `top_errors` and `skipped_lines`. +Top N: add `--top N` to `--summary` to change how many error messages are listed +(default 5, at most 100). + +For example, with this `errors.log`: + +```text +2026-10-10 12:00:00 ERROR Disk full +2026-10-10 12:00:05 ERROR Timeout +2026-10-10 12:00:09 ERROR Disk full +2026-10-10 12:01:00 CRITICAL Out of memory +``` + +`logslice errors.log --summary --top 2` prints: + +```text +Level counts: +DEBUG: 0 +INFO: 0 +WARNING: 0 +ERROR: 3 +CRITICAL: 1 +Top 2 error messages: + 2 Disk full + 1 Timeout +Skipped lines: 0 +``` + Input is processed line by line as UTF-8, replacing invalid bytes. LF and CRLF line endings are supported; output uses UTF-8. Summary mode keeps per-level and per-distinct-error-message counters rather than loading all records into memory. @@ -191,8 +218,9 @@ have been written when a later file fails. | `--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. | +| `--summary` | Print level counts, up to the top 5 ERROR/CRITICAL messages by default, and the skipped line count instead of records. | +| `--summary-format {text,json}` | Output format of `--summary`, default `text`; `json` prints the level counts, up to the top 5 error messages by default, and the skipped line count as a JSON object. | +| `--top N` | With `--summary`, list the N most frequent ERROR/CRITICAL messages instead of 5; `N` is 1 to 100. An error without `--summary`. | | `-h`, `--help` | Show help. | ## Development diff --git a/src/logslice/cli.py b/src/logslice/cli.py index 24ddecb..ecfb31d 100644 --- a/src/logslice/cli.py +++ b/src/logslice/cli.py @@ -6,7 +6,7 @@ from contextlib import nullcontext from typing import TextIO -from .filters import matches, parse_datetime, parse_level +from .filters import matches, parse_datetime, parse_level, parse_top from .parser import Level, parse_line from .summary import Summary @@ -59,6 +59,10 @@ def build_parser() -> argparse.ArgumentParser: "--summary-format", choices=["text", "json"], default="text", help="output format of --summary: text or json (default: text)", ) + parser.add_argument( + "--top", type=parse_top, default=None, metavar="N", + help="number of ERROR/CRITICAL messages with --summary (1-100; default: 5)", + ) return parser @@ -77,6 +81,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.top is not None and not args.summary: + print("logslice: --top 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 @@ -120,5 +127,10 @@ def main(argv: list[str] | None = None) -> int: print(f"logslice: cannot read {path!r}: {error}", file=sys.stderr) return 2 if args.summary: - print(summary.render_json() if args.summary_format == "json" else summary.render()) + top = 5 if args.top is None else args.top + print( + summary.render_json(top) + if args.summary_format == "json" + else summary.render(top) + ) return 0 diff --git a/src/logslice/filters.py b/src/logslice/filters.py index bccff73..ccc38ec 100644 --- a/src/logslice/filters.py +++ b/src/logslice/filters.py @@ -40,6 +40,20 @@ def parse_level(value: str) -> Level: ) from None +def parse_top(value: str) -> int: + """Parse a top-error limit from one to 100.""" + if re.fullmatch(r"[0-9]{1,3}", value) is None: + raise argparse.ArgumentTypeError( + f"invalid top count {value!r}: expected an integer from 1 through 100" + ) + count = int(value) + if not 1 <= count <= 100: + raise argparse.ArgumentTypeError( + f"invalid top count {value!r}: expected an integer from 1 through 100" + ) + return count + + def matches( record: Record, *, diff --git a/src/logslice/summary.py b/src/logslice/summary.py index 7e5d1fe..852c689 100644 --- a/src/logslice/summary.py +++ b/src/logslice/summary.py @@ -21,26 +21,28 @@ def add(self, record: Record) -> None: if record.level >= Level.ERROR: self.error_messages[record.message] += 1 - def render(self) -> str: - """Return all levels, up to five error messages, and skipped count.""" + def render(self, top: int = 5) -> str: + """Return all levels, up to top error messages, and skipped count.""" lines = ["Level counts:"] lines.extend(f"{level.name}: {self.level_counts[level]}" for level in Level) - lines.append("Top 5 error messages:") - top = self.error_messages.most_common(5) - if top: - lines.extend(f" {count} {message}" for message, count in top) + lines.append(f"Top {top} error messages:") + top_messages = self.error_messages.most_common(top) + if top_messages: + lines.extend( + f" {count} {message}" for message, count in top_messages + ) else: lines.append(" (none)") lines.append(f"Skipped lines: {self.skipped_lines}") return "\n".join(lines) - def render_json(self) -> str: - """Return level counts, up to five error messages, and skipped count as JSON.""" + def render_json(self, top: int = 5) -> str: + """Return level counts, up to top error messages, and skipped count as JSON.""" payload = { "levels": {level.name: self.level_counts[level] for level in Level}, "top_errors": [ {"count": count, "message": message} - for message, count in self.error_messages.most_common(5) + for message, count in self.error_messages.most_common(top) ], "skipped_lines": self.skipped_lines, } diff --git a/tests/test_cli.py b/tests/test_cli.py index 518636d..a6a5573 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -49,7 +49,7 @@ def test_help_documents_inputs_options_and_accepted_forms(): "--since", "--until", "--level", "--summary", "stdin", "'-'", "YYYY-MM-DD", "HH:MM:SS", "fraction", "DEBUG", "INFO", "WARNING", "WARN", "ERROR", "CRITICAL", "case-insensitive", - "--grep", "--ignore-case", "--summary-format", + "--grep", "--ignore-case", "--summary-format", "--top", ): assert text in result.stdout diff --git a/tests/test_summary.py b/tests/test_summary.py index 4b96a93..0df5f18 100644 --- a/tests/test_summary.py +++ b/tests/test_summary.py @@ -104,6 +104,13 @@ def test_top_five_descending_counts_first_occurrence_ties_and_skipped_count(): ) +def test_render_default_matches_explicit_top_five(): + summary = Summary() + summary.add(make_record(Level.ERROR, "failure")) + + assert summary.render() == summary.render(5) + + def test_error_and_critical_messages_share_counts_but_lower_levels_do_not(): summary = Summary() for level in Level: diff --git a/tests/test_top.py b/tests/test_top.py new file mode 100644 index 0000000..8824027 --- /dev/null +++ b/tests/test_top.py @@ -0,0 +1,221 @@ +"""Tests for configurable top-error summaries through the CLI.""" + +import json + +import pytest + +from logslice.cli import main + +LOG_CONTENT = ( + "2026-10-10 12:00:00 ERROR Disk full\n" + "2026-10-10 12:00:05 ERROR Timeout\n" + "2026-10-10 12:00:09 ERROR Disk full\n" + "2026-10-10 12:01:00 CRITICAL Out of memory\n" +) + + +def write_log(tmp_path, content=LOG_CONTENT): + path = tmp_path / "app.log" + path.write_text(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 run_argparse_cli(capsys, *args): + with pytest.raises(SystemExit) as error: + main([str(arg) for arg in args]) + captured = capsys.readouterr() + return error.value.code, captured.out, captured.err + + +def test_top_two_text_summary_matches_worked_example(tmp_path, capsys): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "--summary", "--top", "2") + + assert exit_code == 0 + assert stderr == "" + assert stdout == ( + "Level counts:\n" + "DEBUG: 0\n" + "INFO: 0\n" + "WARNING: 0\n" + "ERROR: 3\n" + "CRITICAL: 1\n" + "Top 2 error messages:\n" + " 2 Disk full\n" + " 1 Timeout\n" + "Skipped lines: 0\n" + ) + + +def test_top_limits_json_summary_without_changing_keys_or_counts(tmp_path, capsys): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli( + capsys, path, "--summary", "--top", "1", "--summary-format", "json" + ) + + assert exit_code == 0 + assert stderr == "" + summary = json.loads(stdout) + assert list(summary) == ["levels", "top_errors", "skipped_lines"] + assert summary["levels"]["ERROR"] == 3 + assert summary["levels"]["CRITICAL"] == 1 + assert summary["top_errors"] == [{"count": 2, "message": "Disk full"}] + assert summary["skipped_lines"] == 0 + + +@pytest.mark.parametrize( + ("top_args", "expected_header", "expected_entries"), + [ + (("--top", "1"), "Top 1 error messages:\n", " 2 Disk full\n"), + ( + ("--top=2",), + "Top 2 error messages:\n", + " 2 Disk full\n 1 Timeout\n", + ), + ( + ("--top", "07"), + "Top 7 error messages:\n", + " 2 Disk full\n 1 Timeout\n 1 Out of memory\n", + ), + ( + ("--top", "100"), + "Top 100 error messages:\n", + " 2 Disk full\n 1 Timeout\n 1 Out of memory\n", + ), + ], +) +def test_top_accepts_valid_values_and_normalizes_the_header( + tmp_path, capsys, top_args, expected_header, expected_entries +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli(capsys, path, "--summary", *top_args) + + assert exit_code == 0 + assert stderr == "" + assert expected_header in stdout + assert expected_entries in stdout + + +@pytest.mark.parametrize( + "value", + ["0", "101", "1000", "-3", "+5", " 5", "5 ", "1e1", "x", "٣", ""], +) +def test_invalid_top_values_are_argparse_usage_errors(capsys, value): + exit_code, stdout, stderr = run_argparse_cli( + capsys, "--summary", "--top", value + ) + + assert exit_code == 2 + assert stdout == "" + assert stderr.rstrip().endswith( + "argument --top: invalid top count " + f"{value!r}: expected an integer from 1 through 100" + ) + assert "Traceback" not in stderr + + +def test_top_without_value_keeps_argparse_missing_value_error(capsys): + exit_code, stdout, stderr = run_argparse_cli(capsys, "--summary", "--top") + + assert exit_code == 2 + assert stdout == "" + assert "argument --top: expected one argument" in stderr + + +def test_top_without_summary_fails_before_reading_input(tmp_path, capsys): + missing_path = tmp_path / "missing.log" + + exit_code, stdout, stderr = run_cli(capsys, missing_path, "--top", "5") + + assert exit_code == 2 + assert stdout == "" + assert stderr == "logslice: --top requires --summary\n" + + +def test_summary_format_validation_precedes_top_validation(capsys): + exit_code, stdout, stderr = run_cli( + capsys, "--top", "5", "--summary-format", "json" + ) + + assert exit_code == 2 + assert stdout == "" + assert stderr == "logslice: --summary-format requires --summary\n" + + +def test_top_validation_precedes_invert_match_validation(capsys): + exit_code, stdout, stderr = run_cli(capsys, "--top", "2", "-v") + + assert exit_code == 2 + assert stdout == "" + assert stderr == "logslice: --top requires --summary\n" + + +def test_repeated_top_uses_last_value(tmp_path, capsys): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli( + capsys, path, "--summary", "--top", "1", "--top", "2" + ) + + assert exit_code == 0 + assert stderr == "" + assert "Top 2 error messages:\n" in stdout + assert " 1 Timeout\n" in stdout + + +def test_top_with_fewer_distinct_errors_lists_only_available_messages(tmp_path, capsys): + path = write_log( + tmp_path, + "2026-10-10 12:00:00 ERROR Disk full\n" + "2026-10-10 12:00:01 ERROR Disk full\n", + ) + + exit_code, stdout, stderr = run_cli(capsys, path, "--summary", "--top", "100") + + assert exit_code == 0 + assert stderr == "" + assert "Top 100 error messages:\n 2 Disk full\n" in stdout + assert " (none)" not in stdout + + +def test_top_with_no_error_messages_keeps_none_marker(tmp_path, capsys): + path = write_log( + tmp_path, "2026-10-10 12:00:00 INFO Started\n" + ) + + exit_code, stdout, stderr = run_cli(capsys, path, "--summary", "--top", "3") + + assert exit_code == 0 + assert stderr == "" + assert "Top 3 error messages:\n (none)\n" in stdout + + +@pytest.mark.parametrize( + ("filters", "expected_entries"), + [ + (("--level", "critical"), " 1 Out of memory\n"), + (("--grep", "Disk"), " 2 Disk full\n"), + (("--grep", "Disk", "-v"), " 1 Timeout\n 1 Out of memory\n"), + ], +) +def test_top_preserves_level_and_grep_filter_interactions( + tmp_path, capsys, filters, expected_entries +): + path = write_log(tmp_path) + + exit_code, stdout, stderr = run_cli( + capsys, path, "--summary", "--top", "3", *filters + ) + + assert exit_code == 0 + assert stderr == "" + assert expected_entries in stdout