logwatcher/tests/test_parser.py

from pathlib import Path

import pytest

from logwatcher.models import LogEntry
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.

    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.
    """
    return invalid_log_dir / "bad_log_file.txt"


@pytest.fixture(name="specific_logs")
def specific_logs_fixture(valid_log_dir: Path) -> Path:
    """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().
    """
    return valid_log_dir / "specific_logs.txt"


def test_parse_line(specific_logs: Path):
    """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.
    """
    with open(specific_logs, "r") as file:
        log_lines = file.readlines()

    for log_line in log_lines:
        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
        assert log_entry.start_time is not None
        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 != ""


def test_parse_empty_line():
    r"""An empty or structurally invalid line returns None.

    Covered cases:
    - Empty string.
    - Directory header (\"Répertoire scanné : \\\\...\").
    """
    log_entry = parse_line("")
    assert not log_entry

    log_entry = parse_line("Répertoire scanné : \\\\192.168.60.48\\e$\\MDC_9110\\Logs")
    assert not log_entry


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",
        r"\\192.168.13.22\e\MDC_240\\Logs\26\07\27\20260727160040.txt [27/07/2026 16:04:59] DOSSIER EN COURS : TAPIE [27/07/2026 16:05:08] Erreur FTP SDUpdatePrixISF : Requested action not taken",
        r"\\192.168.60.53\e$\MDC_13110\Logs\26\07\31\20260731102208.txt [31/07/2026 10:26:50] DOSSIER EN COURS : TOSI-NOSYMAG [31/07/2026 10:26:55] Erreur : Exception : Code Fedas non trouvé : 799999",
        (
            r"\\192.168.60.46\e$\MDC_7110\Logs\26\07\30\20260730115342.txt [30/07/2026 11:59:11] DOSSIER EN COURS : "
            r"ROLLAND [30/07/2026 12:00:16] Erreur : GRP_112 / 04507/000 2759_M00054781 WARNING MB153 01058Q26C / "
            r"Art_id : 282646256 - Fusart_id : 197203919 / Impossible de créer un lien en automtique, la grille de "
            r"taille est différente entre l'article d'origne et l'article fusionné"
        ),
        r"\\192.168.13.22\e\MDC_240\Logs\26\07\27\20260727180040.txt [27/07/2026 18:03:44] DOSSIER EN COURS : TAPIE [27/07/2026 18:03:50] Erreur FTP SDUpdatePrixISF : Requested action not taken",
    ]

    log_entries = parse_lines(original_lines)
    assert len(log_entries) == 4


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()
    log_entries = parse_lines(log_lines)
    assert len(log_entries) == len(log_lines)
    for index in range(len(log_lines)):
        assert log_entries[index].error_message in log_lines[index]


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.
    """
    for directory in (original_log_dir, valid_log_dir):
        for log_file in directory.glob("CR_*"):
            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_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.
    """
    log_list = parse_file(empty_log_file)
    assert len(log_list) == 0


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)
    must be silently ignored.
    """
    log_list = parse_file(bad_log_file)
    assert len(log_list) == 0