From cf83a06cfe5878568cbae7d1a57b66c2472c8c92 Mon Sep 17 00:00:00 2001 From: maurane Date: Thu, 10 Sep 2026 09:46:16 +0200 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8Frefactor(parser):=20refactor?= =?UTF-8?q?=20parsing=20methods?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/logwatcher/cli.py | 40 +++++++------------- src/logwatcher/parser.py | 53 ++++++++++++++++---------- tests/test_parser.py | 82 +++++++++++++++++++++------------------- 3 files changed, 91 insertions(+), 84 deletions(-) diff --git a/src/logwatcher/cli.py b/src/logwatcher/cli.py index 55a4523..fd64243 100644 --- a/src/logwatcher/cli.py +++ b/src/logwatcher/cli.py @@ -14,7 +14,7 @@ from logwatcher.mail_reader import ( fetch_log_messages, ) from logwatcher.models import LogEntry -from logwatcher.parser import parse_log_file, parse_log_line +from logwatcher.parser import parse_file, parse_lines from logwatcher.reporter import write_log_report app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.") @@ -23,19 +23,6 @@ app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.") logger = logging.getLogger("logwatcher") -def version_callback(value: bool): - """ - Print the version of package. - Stop the CLI after printing version. - - Args: - value: True if '--version' is provided, False otherwise. - """ - if value: - print(f"logwatcher version: {__version__}") - raise typer.Exit() - - def _is_log_file(path: Path) -> bool: """ Return True if file is a log file (.txt, .log, or extensionless). @@ -60,26 +47,27 @@ def _get_files_from_dir(input_dir: Path) -> list[Path]: def _files_to_log_entries(input_files: list[Path]) -> list[LogEntry]: + """Parse each input file, raising if any path is not a file.""" log_entries = [] for file in input_files: if not file.is_file(): logger.error(f"Unrecognized input file '{file}'.") raise ValueError(f"'{file}' is not a file") - log_entries.extend(parse_log_file(file)) + log_entries.extend(parse_file(file)) return log_entries -def _lines_to_log_entries(logs: list[str]) -> list[LogEntry]: - log_entries = [] - for log in logs: - log_entry = parse_log_line(log) - if log_entry: - log_entries.append(log_entry) - return log_entries +def version_callback(value: bool): + """ + Print the version of package. + Stop the CLI after printing version. - -def analyze_logs(): - pass + Args: + value: True if '--version' is provided, False otherwise. + """ + if value: + print(f"logwatcher version: {__version__}") + raise typer.Exit() @app.command() @@ -95,7 +83,7 @@ def from_mail( account = connect_to_mailbox() messages = fetch_log_messages(account) logs = extract_logs_from_mails(messages) - log_entries = _lines_to_log_entries(logs) + log_entries = parse_lines(logs) nb_files = len(messages) relevant, irrelevant = classify_log_entries(log_entries) diff --git a/src/logwatcher/parser.py b/src/logwatcher/parser.py index 97a0bc5..abd84df 100644 --- a/src/logwatcher/parser.py +++ b/src/logwatcher/parser.py @@ -1,5 +1,6 @@ import logging import re +from collections.abc import Iterable from datetime import datetime from pathlib import Path @@ -35,8 +36,9 @@ misc_file_path = Path.cwd() / "output/misc.log" effective_file_path = Path.cwd() / "output/effective.log" -def parse_log_line(log_line: str) -> LogEntry | None: - """Parse a raw log line into a LogEntry object. +def parse_line(log_line: str) -> LogEntry | None: + """ + Parse a raw log line into a LogEntry object. Args: log_line: a line in a log file received by N2 technicians @@ -65,10 +67,32 @@ def parse_log_line(log_line: str) -> LogEntry | None: ) -def parse_log_file(log_file_path: Path) -> list[LogEntry]: - """Transforms the content of a log file into a list of LogEntry. - Uses `parse_log_line` to parse each line of the choosen log file. - Raise exception if a log line encounters an error. +def parse_lines(lines: Iterable[str]) -> list[LogEntry]: + """ + Parse an iterable of log lines into LogEntry objects. + Unrecognized lines are skipped. Logs an error for lines that raise a ValueError. + + Args: + lines: Iterable of raw log lines. + + Returns: + List of parsed LogEntry objects. + """ + log_entries: list[LogEntry] = [] + for index, line in enumerate(lines): + try: + log_entry = parse_line(line) + except ValueError as error: + logger.error(f"\t\tline {index}: error while parsing the log: {error}") + else: + if log_entry: + log_entries.append(log_entry) + return log_entries + + +def parse_file(log_file_path: Path) -> list[LogEntry]: + """ + Transforms the content of a log file into a list of LogEntry. Args: log_file_path: Path of a log file @@ -76,20 +100,9 @@ def parse_log_file(log_file_path: Path) -> list[LogEntry]: Returns: A list of LogEntry """ - log_entry_list: list[LogEntry] = [] - with open(log_file_path, "r", encoding="windows-1252") as log_file: logger.info(f"\tparsing log file '{log_file_path.name}' started.") - for index, log_line in enumerate(log_file): - try: - log_entry = parse_log_line(log_line) - except ValueError as error: - logger.error( - f"\t\tline {index}: error while parsing the log file '{log_file.name}': {error}" - ) - else: - if log_entry: - log_entry_list.append(log_entry) - logger.info(f"\t\t{len(log_entry_list)} lines were parsed.") + log_entries = parse_lines(log_file) + logger.info(f"\t\t{len(log_entries)} lines were parsed.") logger.info("\tparsing completed.") - return log_entry_list + return log_entries diff --git a/tests/test_parser.py b/tests/test_parser.py index 36254a8..a993760 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -3,48 +3,50 @@ from pathlib import Path import pytest from logwatcher.models import LogEntry -from logwatcher.parser import parse_log_file, parse_log_line +from logwatcher.parser import parse_file, parse_line @pytest.fixture(name="empty_log_file") def empty_log_file_fixture(invalid_log_dir: Path) -> Path: """ - Path of an empty log file. - Ensures the parser does not raise an exception - and returns an empty list. + Path of an empty log file. + Ensures the parser does not raise an exception + and returns an empty list. """ return invalid_log_dir / "empty_file.txt" + @pytest.fixture(name="bad_log_file") def bad_log_file_fixture(invalid_log_dir: Path) -> Path: """ - Path of a file containing only incorrectly formatted lines. - Each line fails at a different point in LOG_PATTERN - (invalid IP, missing timestamp, missing DOSSIER EN COURS, etc.). - No LogEntry should be produced. + Path of a file containing only incorrectly formatted lines. + Each line fails at a different point in LOG_PATTERN + (invalid IP, missing timestamp, missing DOSSIER EN COURS, etc.). + No LogEntry should be produced. """ return invalid_log_dir / "bad_log_file.txt" + @pytest.fixture(name="specific_logs") def specific_logs_fixture(valid_log_dir: Path) -> list[str]: """ - Single log lines chosen to cover specific cases. - Each line is structurally valid and must be parsed - successfully. Used for unit tests of parse_log_line(). + Single log lines chosen to cover specific cases. + Each line is structurally valid and must be parsed + successfully. Used for unit tests of parse_line(). """ log_file = valid_log_dir / "specific_logs.txt" - with open(log_file, 'r') as file: + with open(log_file, "r") as file: return file.readlines() -def test_parse_log_line(specific_logs: list[str]): +def test_parse_line(specific_logs: list[str]): """ - Every structurally valid line produces a complete LogEntry. - Ensures no required field is empty or None after parsing - a line conforming to the LAME MDC format. + Every structurally valid line produces a complete LogEntry. + Ensures no required field is empty or None after parsing + a line conforming to the LAME MDC format. """ for log_line in specific_logs: - log_entry = parse_log_line(log_line) + log_entry = parse_line(log_line) assert log_entry is not None assert log_entry.server_ip is not None assert log_entry.mdc_server_name is not None @@ -52,49 +54,53 @@ def test_parse_log_line(specific_logs: list[str]): assert log_entry.store_name is not None assert log_entry.error_time is not None assert log_entry.error_message is not None - assert log_entry.raw_line != '' + assert log_entry.raw_line != "" + def test_parse_empty_log_line(): """ - An empty or structurally invalid line returns None. - Covered cases: - - Empty string. - - Directory header (\"Répertoire scanné : \\\\...\"). + An empty or structurally invalid line returns None. + Covered cases: + - Empty string. + - Directory header (\"Répertoire scanné : \\\\...\"). """ - log_entry = parse_log_line("") + log_entry = parse_line("") assert not log_entry - log_entry = parse_log_line("Répertoire scanné : \\\\192.168.60.48\\e$\\MDC_9110\\Logs") + log_entry = parse_line("Répertoire scanné : \\\\192.168.60.48\\e$\\MDC_9110\\Logs") assert not log_entry + def test_parse_valid_log_file(original_log_dir: Path): """ - Each real log file produces at least one valid LogEntry. - Integration test: files provided by technicians contain - a mix of valid lines and lines to be ignored. - Ensures the parser extracts at least one entry per file. + Each real log file produces at least one valid LogEntry. + Integration test: files provided by technicians contain + a mix of valid lines and lines to be ignored. + Ensures the parser extracts at least one entry per file. """ for log_file in original_log_dir.glob("CR_*"): - log_list = parse_log_file(log_file) + log_list = parse_file(log_file) assert len(log_list) > 0 for log_entry in log_list: assert isinstance(log_entry, LogEntry) + def test_parse_empty_log_file(empty_log_file): """ - An empty log file can be parsed and must returns an - empty LogEntry list. + An empty log file can be parsed and must returns an + empty LogEntry list. """ - log_list = parse_log_file(empty_log_file) + log_list = parse_file(empty_log_file) assert len(log_list) == 0 + def test_parse_bad_log_file(bad_log_file): """ - A file containing only invalid lines produces nothing. - Any line that does not match the LAME MDC structure - (IP, timestamps, DOSSIER EN COURS, error message) - must be silently ignored. + A file containing only invalid lines produces nothing. + Any line that does not match the LAME MDC structure + (IP, timestamps, DOSSIER EN COURS, error message) + must be silently ignored. """ - log_list = parse_log_file(bad_log_file) - assert len(log_list) == 0 \ No newline at end of file + log_list = parse_file(bad_log_file) + assert len(log_list) == 0