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():
    """
    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():
    # 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):
    # 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):
    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):
    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):
    """
    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):
    """
    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):
    """
    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