diff --git a/src/logwatcher/__init__.py b/src/logwatcher/__init__.py index 11c9b79..797f1f0 100644 --- a/src/logwatcher/__init__.py +++ b/src/logwatcher/__init__.py @@ -1,3 +1,5 @@ +"""CLI tool to parse, classify and report NOSYMAG LAME MDC error logs.""" + from importlib.metadata import version from dotenv import load_dotenv diff --git a/src/logwatcher/__main__.py b/src/logwatcher/__main__.py index ae781b8..11a0cbe 100644 --- a/src/logwatcher/__main__.py +++ b/src/logwatcher/__main__.py @@ -2,6 +2,7 @@ from logwatcher.cli import app def main(): + """Entrypoint of the package.""" app() diff --git a/src/logwatcher/classifier.py b/src/logwatcher/classifier.py index d8d991d..7b792e2 100644 --- a/src/logwatcher/classifier.py +++ b/src/logwatcher/classifier.py @@ -51,9 +51,7 @@ def _match_n2_pattern(log_entry: LogEntry) -> str | None: def classify_log_entries( log_entries: list[LogEntry], ) -> tuple[list[LogEntry], list[LogEntry]]: - """ - Separate logs that require N2 intervention from those that don't. - """ + """Separate logs that require N2 intervention from those that don't.""" relevant_log_entries = [] irrelevant_log_entries = [] diff --git a/src/logwatcher/config.py b/src/logwatcher/config.py index e64e6da..9dddda0 100644 --- a/src/logwatcher/config.py +++ b/src/logwatcher/config.py @@ -19,5 +19,7 @@ FRENCH_TIMEZONE = timezone(offset=timedelta(hours=2)) # UTC+2 = CEST # file type class SourceType(Enum): + """Designate which type of template to use while building reports.""" + FILE = "fichier" MAIL = "mail" diff --git a/src/logwatcher/logging_config.py b/src/logwatcher/logging_config.py index 7837a7f..a2c6260 100644 --- a/src/logwatcher/logging_config.py +++ b/src/logwatcher/logging_config.py @@ -29,8 +29,8 @@ def setup_logging( file_path: Path = Path("output/logs/logwatcher.log"), verbose: bool = False, ) -> logging.Logger: - """ - Create a new instance of Logger named 'logger' customized for the logwatcher package. + """Create a new instance of Logger named 'logger' customized for the logwatcher package. + The format display in order the time, the level of log and the log message The logs are only redirect in the standard output. """ diff --git a/src/logwatcher/mail_reader.py b/src/logwatcher/mail_reader.py index 64b69d5..2700902 100644 --- a/src/logwatcher/mail_reader.py +++ b/src/logwatcher/mail_reader.py @@ -14,13 +14,10 @@ from exchangelib import ( ) from exchangelib.errors import UnauthorizedError -from logwatcher.utils import get_or_create_folder +from logwatcher.mail_utils import ANALYZED_FOLDER, LOG_IN_ATTACHMENT_PATTERN, get_or_create_folder logger = logging.getLogger(__name__) -ANALYZED_FOLDER = "Analyzed" -LOG_IN_ATTACHMENT_PATTERN = "Le compte-rendu contient plus de 100 lignes." - def _find_log_attachment(mail: Message) -> FileAttachment | None: """Find the right log file attachment among other content such as images from footers. @@ -53,6 +50,7 @@ def _get_attachment_content(attachment: FileAttachment) -> str: Returns: all logs in one string + """ raw = attachment.content if raw is None: diff --git a/src/logwatcher/utils.py b/src/logwatcher/mail_utils.py similarity index 88% rename from src/logwatcher/utils.py rename to src/logwatcher/mail_utils.py index 4a219a6..5ab6601 100644 --- a/src/logwatcher/utils.py +++ b/src/logwatcher/mail_utils.py @@ -4,6 +4,10 @@ from exchangelib import Account, Folder logger = logging.getLogger(__name__) +ANALYZED_FOLDER = "Analyzed" +LOG_IN_ATTACHMENT_PATTERN = "Le compte-rendu contient plus de 100 lignes." + + def get_or_create_folder(account: Account, folder_name: str) -> Folder: """Return the 'Analyzed' folder, creating it if it doesn't exist. diff --git a/src/logwatcher/models.py b/src/logwatcher/models.py index c901611..10e0125 100644 --- a/src/logwatcher/models.py +++ b/src/logwatcher/models.py @@ -6,9 +6,7 @@ from logwatcher.config import DATETIME_FORMAT @dataclass() class LogEntry: - """ - Represents a line in log file. - """ + """Represents a line in log file.""" server_ip: str mdc_server_name: str @@ -19,22 +17,23 @@ class LogEntry: raw_line: str # full initial log error_name: str = "" - def __str__(self): + def __str__(self): # noqa: D105 return self.error_message def get_full_message(self): + """Return raw line of the log entry.""" return self.raw_line - def get_start_time(self) -> str: + def get_start_time(self) -> str: # TODO: remove or adapt mentions of start_time in code base + """Return formatted start time.""" return self.start_time.strftime(DATETIME_FORMAT) - def get_error_time(self) -> str: + def get_error_time(self) -> str: # TODO: remove or adapt mentions of error_time in code base + """Return formatted error time.""" return self.error_time.strftime(DATETIME_FORMAT) def __post_init__(self): - """ - Each property must be validated by specific regex defined in config.py - """ + """Each property must be validated by specific regex defined in config.py.""" # error time cannot be earlier than start time if self.start_time > self.error_time: raise ValueError( diff --git a/src/logwatcher/notifier.py b/src/logwatcher/notifier.py index ebd15d7..6f97b46 100644 --- a/src/logwatcher/notifier.py +++ b/src/logwatcher/notifier.py @@ -4,7 +4,7 @@ from pathlib import Path from exchangelib import Account, FileAttachment, Message -from logwatcher.utils import get_or_create_folder +from logwatcher.mail_utils import get_or_create_folder logger = logging.getLogger(__name__) diff --git a/src/logwatcher/parser.py b/src/logwatcher/parser.py index b93e056..2ec9964 100644 --- a/src/logwatcher/parser.py +++ b/src/logwatcher/parser.py @@ -31,8 +31,7 @@ LOG_PATTERN = re.compile( def parse_line(log_line: str) -> LogEntry | None: - """ - Parse a raw log line into a LogEntry object. + """Parse a raw log line into a LogEntry object. Args: log_line: a line in a log file received by N2 technicians @@ -40,6 +39,7 @@ def parse_line(log_line: str) -> LogEntry | None: Returns: LogEntry if the line is a valid line, None if it is empty or not a valid line. + """ match = LOG_PATTERN.search(log_line) if not match: @@ -62,8 +62,8 @@ def parse_line(log_line: str) -> LogEntry | None: def parse_lines(lines: Iterable[str]) -> list[LogEntry]: - """ - Parse an iterable of log lines into LogEntry objects. + """Parse an iterable of log lines into LogEntry objects. + Unrecognized lines are skipped. Logs an error for lines that raise a ValueError. Args: @@ -71,6 +71,7 @@ def parse_lines(lines: Iterable[str]) -> list[LogEntry]: Returns: List of parsed LogEntry objects. + """ log_entries: list[LogEntry] = [] for index, line in enumerate(lines): @@ -85,14 +86,14 @@ def parse_lines(lines: Iterable[str]) -> list[LogEntry]: def parse_file(log_file_path: Path) -> list[LogEntry]: - """ - Transforms the content of a log file into a list of LogEntry. + """Transform the content of a log file into a list of LogEntry. Args: log_file_path: Path of a log file Returns: A list of LogEntry + """ with open(log_file_path, "r", encoding="windows-1252") as log_file: logger.info(f"\tparsing log file '{log_file_path.name}' started.") diff --git a/src/logwatcher/reporter.py b/src/logwatcher/reporter.py index 04742a6..5012f17 100644 --- a/src/logwatcher/reporter.py +++ b/src/logwatcher/reporter.py @@ -57,6 +57,17 @@ La liste des erreurs se trouvent en pièce jointe `n2.log`. def get_period(entries: list[LogEntry]) -> tuple[str, str]: + """Get start date and end date among dates of log entries. + + Check all log entries dates and find the timestamp of earliest and latest generated logs + + Args: + entries: list of log entries + + Returns: + start date and end date in string format + + """ if not entries: return "", "" times: list[datetime] = [e.error_time for e in entries] @@ -73,9 +84,8 @@ def write_log_report( end_date: str, output_dir: Path = RESULT_PATH, ) -> None: - """ - Write each relevant, irrelevant and general reports in their respective output - file. + """Write each relevant, irrelevant and general reports in their respective output file. + Use the range date of relevant and irrelevant lists to get the period of time the logs were generated. @@ -88,8 +98,7 @@ def write_log_report( irrelevant: List of not n2 log entry nb_files: Number of files scanned output_dir: Location where all reports will be written - Returns: - None + """ logger.info("\twriting reports job started.") reports_dict = build_reports( @@ -116,8 +125,8 @@ def build_reports( end_date: str, nb_files: int, ) -> dict[str, str]: - """ - Build the three output reports: n2, other, and all. + """Build the three output reports: n2, other, and all. + "all" report contains both "n2" and "other" sections. "n2" and "other" are two unique sections. @@ -129,8 +138,10 @@ def build_reports( start_date: Date of the oldest log entry in relevant + irrelevant list end_date: Date of the newest log entry in relevant + irrelevant list nb_files: number of files scanned + Returns: Dictionary of reports in string format + """ logger.info("\t\tbuilding reports job started.") relevant_report = _render_target_report( @@ -197,6 +208,7 @@ def build_mail_summary( Returns: A short summary suitable for an email body. + """ logger.info("generating mail body...") return MAIL_TEMPLATE.substitute( @@ -212,30 +224,34 @@ def build_mail_summary( def _render_target_report( log_entries: list[LogEntry], target_template: Template ) -> str: - """ - Render report using either N2_SUPPORT_TEMPLATE or OTHER_TEMPLATE. + """Render report using either N2_SUPPORT_TEMPLATE or OTHER_TEMPLATE. + Call _render_entries() and include rendered log entries in the new report. Args: log_entries: List of log entries target_template: Template to use for N2_SUPPORT_TEMPLATE + Returns: content of report in string format + """ error_list = _render_entries(log_entries) return target_template.substitute(nb_errors=len(log_entries), error_list=error_list) def _render_entries(log_entries: list[LogEntry]) -> str: - """ - Render a list of log entries using the `ERROR_TEMPLATE`. + """Render a list of log entries using the `ERROR_TEMPLATE`. + Used in generated report files in the `output` directory Args: log_entries: List of log entries + Returns: error report in string format + """ error_list = [] for index, log_entry in enumerate(log_entries, start=1): diff --git a/tests/test_classifier.py b/tests/test_classifier.py index f4e98c0..1a79f37 100644 --- a/tests/test_classifier.py +++ b/tests/test_classifier.py @@ -10,16 +10,12 @@ from logwatcher.parser import parse_file def _pattern_name_to_filename(name: str) -> str: - """ - Returns the fixture corresponding to the error code name - """ + """Return the fixture corresponding to the error code name.""" return f"CR_{name.lower()}.txt" def _read_fixture(path: Path) -> str: - """ - Read a file with fallback Windows-1252. - """ + """Read a file with fallback Windows-1252.""" try: content = path.read_text(encoding="utf-8") except UnicodeDecodeError: @@ -31,10 +27,7 @@ def _read_fixture(path: Path) -> str: @pytest.mark.parametrize("name, pattern", N2_PATTERNS.items()) def test_pattern_matches_fixture(valid_log_dir, name, pattern): - """ - Assert each N2 pattern must match at least one line in its fixture. - """ - + """Assert each N2 pattern must match at least one line in its fixture.""" fixture = valid_log_dir / _pattern_name_to_filename(name) if not fixture.exists(): msg = f"No fixture named '{fixture}'" @@ -58,9 +51,7 @@ def test_pattern_matches_fixture(valid_log_dir, name, pattern): def test_patterns_do_not_overlap(): - """ - Assert that a log line can match only one N2 pattern. - """ + """Assert that a log line can match only one N2 pattern.""" fixtures_dir = FIXTURE_PATH for fixture in fixtures_dir.glob("CR_*.txt"): line_list = [ @@ -79,10 +70,7 @@ def test_patterns_do_not_overlap(): def test_no_orphan_fixtures(): - """ - Assert that a fixture exist only if its associated pattern exists. - Aucun fichier de fixture ne doit exister sans pattern associé. - """ + """Assert that a fixture exist only if its associated pattern exists.""" expected = {f"CR_{name.lower()}.txt" for name in N2_PATTERNS} actual = {f.name for f in FIXTURE_PATH.glob("CR_*.txt")} orphan = actual - expected @@ -90,9 +78,7 @@ def test_no_orphan_fixtures(): def test_classify_log_entries_all_relevant(valid_log_dir: Path): - """ - Must return an empty irrelevant log entry list - """ + """Must return an empty irrelevant log entry list.""" log_file = valid_log_dir / "only_relevant_logs.txt" log_entries = parse_file(log_file) @@ -103,9 +89,7 @@ def test_classify_log_entries_all_relevant(valid_log_dir: Path): def test_classify_log_entries_none_relevant(valid_log_dir: Path): - """ - Must return an empty relevant log entry list - """ + """Must return an empty relevant log entry list.""" log_file = valid_log_dir / "no_relevant_logs.txt" log_entries = parse_file(log_file) @@ -116,9 +100,7 @@ def test_classify_log_entries_none_relevant(valid_log_dir: Path): def test_classify_log_entries_mixed(valid_log_dir: Path): - """ - Test classification on relevant and irrelevant log entry list - """ + """Test classification on relevant and irrelevant log entry list.""" log_file = valid_log_dir / "mixed_logs.txt" log_entries = parse_file(log_file) @@ -129,9 +111,7 @@ def test_classify_log_entries_mixed(valid_log_dir: Path): def test_classify_log_entries_empty_log_entries(invalid_log_dir: Path): - """ - Test classification on an empty log entry list - """ + """Test classification on an empty log entry list.""" empty_file = invalid_log_dir / "empty_file.txt" log_entries = parse_file(empty_file) diff --git a/tests/test_logging.py b/tests/test_logging.py index e5a14fc..a7ff34a 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -1,14 +1,3 @@ -""" -this module contains all tests logging-related : - -- the logger should have at least one StreamHandler and one FileHandler each time -logwatcher is launched. -- the StreamHandler must watch at an error level -- the FileHandler must watch at a debug -level -- -""" - import logging import re from logging import DEBUG, ERROR, INFO, WARNING, FileHandler, StreamHandler, getLogger @@ -20,6 +9,7 @@ from logwatcher.logging_config import _setup_formatter, _setup_handler, setup_lo def test_setup_handler(tmp_log_file): + """Assert `setup_handler` instanciate one StreamHandler and one FileHandler.""" formatter = _setup_formatter() handler = _setup_handler(formatter=formatter, level=DEBUG) assert isinstance(handler, StreamHandler) @@ -29,6 +19,10 @@ def test_setup_handler(tmp_log_file): def test_count_of_handler(tmp_log_file): + """Assert `setup_handler` instanciate exactly one StreamHandler and one FileHandler. + + Must have exactly two handlers in total. + """ logger = getLogger() count_file_handler = 0 count_stream_handler = 0 @@ -46,6 +40,10 @@ def test_count_of_handler(tmp_log_file): def test_type_of_handlers(): + """Assert each handlers capture the right log level and use the right log format. + + FileHandler must capute DEBUG logs while StreamHandler only capture ERROR logs. + """ logger = logging.getLogger("logwatcher") for handler in logger.handlers: if isinstance(handler, FileHandler): @@ -59,8 +57,9 @@ def test_type_of_handlers(): def test_modified_format(): - """ - when starting, the logger is first intialized, thus two handlers are at position + """Assert `setup_logging` allow to use another log format. + + When starting, the logger is first intialized, thus two handlers are at position 1 and 2 in logger.handlers list. """ logger = setup_logging() @@ -82,6 +81,10 @@ def test_modified_format(): def test_logging_in_correct_path(tmp_log_file): + """Assert log file is correctly written in output directory. + + Check content of the file. It must contain logs from CRITICAL to DEBUG level. + """ logger = logging.getLogger("logwatcher") # must be both in file and stdout logger.critical("hi- BYE") @@ -100,7 +103,7 @@ def test_logging_in_correct_path(tmp_log_file): for line in content: assert re.search(regex, line) - # test critical to warning + # test from critical to debug logs assert re.search(r"CRITICAL: hi- BYE", content[0]) assert re.search(r"ERROR: hi... ok, bye", content[1]) assert re.search(r"WARNING: hi, are you alright?", content[2]) diff --git a/tests/test_mail_reader.py b/tests/test_mail_reader.py index 0c2b45f..b99ba83 100644 --- a/tests/test_mail_reader.py +++ b/tests/test_mail_reader.py @@ -110,7 +110,7 @@ class OkProtocol: class OkAccount: """Account with a protocol that returns a version successfully.""" - def __init__(self): # noqa: D102 + def __init__(self): # noqa: D107 self.protocol = OkProtocol() @@ -380,7 +380,7 @@ def test_fetch_log_messages_folder_path(make_mock_account): # extract_log_lines(content) # ############################## def test_extract_log_lines_split_and_clean(original_log_dir: Path): - """Splits content, strips \r\n, removes empty lines.""" + r"""Split content, strips '\r\n', removes empty lines.""" file = original_log_dir / "CR_20260727110008.txt" content = file.read_text(encoding="windows-1252") lines = extract_log_lines(content) @@ -412,7 +412,7 @@ def test_extract_log_lines_no_trailing_newline(): def test_extract_log_lines_crlf(): - """Handles Windows \r\n line endings.""" + r"""Handles Windows \r\n line endings.""" content = ( "Répertoire scanné : \\192.168.60.40\\e$\\MDC_1110\\Logs\n" "\\192.168.60.40\\e$\\MDC_1110\\Logs\\26\\07\\27\\20260727083117.txt [27/07/2026 08:31:26] DOSSIER EN COURS : FLAMMIER [27/07/2026 08:31:53] Erreur FTP commande : Requested action not taken\r\n" diff --git a/tests/test_utils.py b/tests/test_mail_utils.py similarity index 75% rename from tests/test_utils.py rename to tests/test_mail_utils.py index eea16e6..2f4382e 100644 --- a/tests/test_utils.py +++ b/tests/test_mail_utils.py @@ -1,22 +1,22 @@ from unittest.mock import MagicMock -from logwatcher.utils import get_or_create_folder +from logwatcher.mail_utils import get_or_create_folder def test_get_or_create_analyzed_folder_exists(make_mock_account, monkeypatch): - """ - Returns the existing 'Analyzed' folder without creating it. + """Returns the existing 'Analyzed' folder without creating it. Args: - make_mock_account: Fake account owning the Logs folder. + make_mock_account: Fake account owning the Logs folder monkeypatch: MonkeyPatch to generate test context + """ existing_folder = MagicMock() account = make_mock_account(analyzed_folder=existing_folder) fake_folder_cls = MagicMock() with monkeypatch.context() as m: - m.setattr("logwatcher.utils.Folder", fake_folder_cls) + m.setattr("logwatcher.mail_utils.Folder", fake_folder_cls) result = get_or_create_folder(account, "Analyzed") assert result is existing_folder @@ -30,7 +30,7 @@ def test_get_or_create_analyzed_folder_creates(make_mock_account, monkeypatch): fake_folder_instance = MagicMock() fake_folder_cls = MagicMock(return_value=fake_folder_instance) with monkeypatch.context() as m: - m.setattr("logwatcher.utils.Folder", fake_folder_cls) + m.setattr("logwatcher.mail_utils.Folder", fake_folder_cls) result = get_or_create_folder(account, "Analyzed") assert result is fake_folder_instance diff --git a/tests/test_models.py b/tests/test_models.py index 4189d01..aa203b0 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -7,6 +7,7 @@ from logwatcher.models import LogEntry def test_valid_log_entry(): + """Assert a log entry is correctly created when all arguments are correct.""" LogEntry( server_ip="192.168.13.27", mdc_server_name="MDC_720", @@ -18,6 +19,7 @@ def test_valid_log_entry(): ) def test_log_entry_invalid_dates(): + """Assert a log entry cannot be created if start time is not later than error time.""" with pytest.raises(ValueError, match="Error in date-times"): LogEntry( server_ip="192.168.13.27", @@ -30,6 +32,7 @@ def test_log_entry_invalid_dates(): ) def test_equals_models(): + """Assert two models are equals if they have the same value as attributes.""" log_entry_1 = LogEntry( server_ip="192.168.13.27", mdc_server_name="MDC_720", diff --git a/tests/test_parser.py b/tests/test_parser.py index 96db26d..82552b0 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -8,8 +8,8 @@ from logwatcher.parser import parse_file, parse_line, parse_lines @pytest.fixture(name="empty_log_file") def empty_log_file_fixture(invalid_log_dir: Path) -> Path: - """ - Path of an empty log file. + """Path of an empty log file. + Ensures the parser does not raise an exception and returns an empty list. """ @@ -18,8 +18,8 @@ def empty_log_file_fixture(invalid_log_dir: Path) -> Path: @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. + """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. @@ -29,8 +29,8 @@ def bad_log_file_fixture(invalid_log_dir: Path) -> Path: @pytest.fixture(name="specific_logs") def specific_logs_fixture(valid_log_dir: Path) -> Path: - """ - Single log lines chosen to cover specific cases. + """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(). """ @@ -38,8 +38,8 @@ def specific_logs_fixture(valid_log_dir: Path) -> Path: def test_parse_line(specific_logs: Path): - """ - Every structurally valid line produces a complete LogEntry. + """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. """ @@ -59,8 +59,8 @@ def test_parse_line(specific_logs: Path): def test_parse_empty_line(): - """ - An empty or structurally invalid line returns None. + r"""An empty or structurally invalid line returns None. + Covered cases: - Empty string. - Directory header (\"Répertoire scanné : \\\\...\"). @@ -73,6 +73,7 @@ def test_parse_empty_line(): def test_parse_lines(): + """Assert parse_log find the valid logs among a mixed of logs.""" # mix of valid and invalid logs. Have 4 valid logs original_lines = [ r"Répertoire scanné : \\192.168.13.22\e\MDC_240\Logs", @@ -92,6 +93,7 @@ def test_parse_lines(): def test_parse_valid_lines(specific_logs: Path): + """Assert parse_lines works for all valid lines.""" # selected valid logs with open(specific_logs, "r") as file: log_lines = file.readlines() @@ -102,20 +104,24 @@ def test_parse_valid_lines(specific_logs: Path): def test_parse_empty_lines(empty_log_file: Path): + """Assert parsing an empty line does not raise an error and return an empty log entry list.""" with open(empty_log_file, "r") as file: log_entries = parse_lines(file.readlines()) assert len(log_entries) == 0 def test_parse_invalid_lines(bad_log_file: Path): + """Assert parsing an invalid line does not raise an error and return an empty log entry list.""" with open(bad_log_file, "r") as file: log_entries = parse_lines(file.readlines()) assert len(log_entries) == 0 def test_parse_files(original_log_dir: Path, valid_log_dir: Path): - """ + """Assert parse_files works with real life logs. + 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. @@ -130,7 +136,8 @@ def test_parse_files(original_log_dir: Path, valid_log_dir: Path): def test_parse_empty_file(empty_log_file): - """ + """Assert parsing an empty file does not raise an error and return an empty log entrt list instead. + An empty log file can be parsed and must returns an empty LogEntry list. """ @@ -139,7 +146,8 @@ def test_parse_empty_file(empty_log_file): def test_parse_bad_file(bad_log_file): - """ + """Assert parsing a bad file does not raise an error and return an empty log entry list instead. + 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)