228 lines
7.4 KiB
Python
228 lines
7.4 KiB
Python
import re
|
|
import warnings
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from logwatcher.classifier import N2_PATTERNS, classify_log_entries
|
|
from logwatcher.config import FIXTURE_PATH
|
|
from logwatcher.parser import parse_file, parse_lines
|
|
|
|
|
|
def _pattern_name_to_filename(name: str) -> str:
|
|
"""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."""
|
|
try:
|
|
content = path.read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
content = path.read_text(encoding="windows-1252")
|
|
if "\ufffd" not in content:
|
|
return content
|
|
return path.read_text(encoding="windows-1252") # TODO: improve coverage score
|
|
|
|
|
|
@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."""
|
|
fixture = valid_log_dir / _pattern_name_to_filename(name)
|
|
if not fixture.exists():
|
|
msg = f"No fixture named '{fixture}'"
|
|
warnings.warn(msg)
|
|
pytest.skip(msg)
|
|
|
|
lignes = [l for l in _read_fixture(fixture).splitlines() if l.strip()]
|
|
|
|
assert lignes, f"empty fixture file : {fixture}"
|
|
|
|
match_count = 0
|
|
|
|
for ligne in lignes:
|
|
match = re.search(pattern, ligne)
|
|
if match:
|
|
match_count += 1
|
|
|
|
assert match_count > 0, (
|
|
f"{name} : no match in file {fixture.name}\npattern: {pattern}"
|
|
)
|
|
|
|
|
|
def test_patterns_do_not_overlap():
|
|
"""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 = [
|
|
line for line in _read_fixture(fixture).splitlines() if line.strip()
|
|
] # TODO: improve coverage score
|
|
for line in line_list: # TODO: improve coverage score
|
|
matching = [ # TODO: improve coverage score
|
|
name
|
|
for name, raw in N2_PATTERNS.items()
|
|
if re.compile(raw).search(line)
|
|
]
|
|
assert len(matching) <= 1, ( # TODO: improve coverage score
|
|
f"Chevauchement détecté sur {fixture.name} : "
|
|
f"{matching} match line : {line[:80]}"
|
|
)
|
|
|
|
|
|
def test_no_orphan_fixtures():
|
|
"""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
|
|
assert not orphan, f"Fixtures orphelines (sans pattern) : {sorted(orphan)}"
|
|
|
|
|
|
def test_classify_log_entries_all_relevant(valid_log_dir: Path):
|
|
"""Must return an empty irrelevant log entry list."""
|
|
log_file = valid_log_dir / "only_relevant_logs.txt"
|
|
log_entries = parse_file(log_file)
|
|
|
|
# all entries are relevant
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert relevant and not irrelevant
|
|
|
|
|
|
def test_classify_log_entries_none_relevant(valid_log_dir: Path):
|
|
"""Must return an empty relevant log entry list."""
|
|
log_file = valid_log_dir / "no_relevant_logs.txt"
|
|
log_entries = parse_file(log_file)
|
|
|
|
# all entries are relevant
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert not relevant and irrelevant
|
|
|
|
|
|
def test_classify_log_entries_mixed(valid_log_dir: Path):
|
|
"""Test classification on relevant and irrelevant log entry list."""
|
|
log_file = valid_log_dir / "mixed_logs.txt"
|
|
log_entries = parse_file(log_file)
|
|
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert len(relevant) and len(irrelevant)
|
|
assert len(relevant) + len(irrelevant) == len(log_entries)
|
|
|
|
|
|
def test_classify_log_entries_empty_log_entries(invalid_log_dir: Path):
|
|
"""Test classification on an empty log entry list."""
|
|
empty_file = invalid_log_dir / "empty_file.txt"
|
|
log_entries = parse_file(empty_file)
|
|
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert not relevant and not irrelevant
|
|
|
|
|
|
N2_ERROR_MESSAGE = "Erreur : GetTvaId -> TVA non trouvée : 20,00"
|
|
OTHER_ERROR_MESSAGE = "Erreur FTP SDUpdatePrixISF : Requested action not taken"
|
|
|
|
|
|
def _build_log_line(
|
|
error_time: str,
|
|
error_message: str,
|
|
store_name: str = "GABRIEL",
|
|
start_time: str = "22/09/2026 09:28:58",
|
|
mdc_server_name: str = "MDC_220",
|
|
server_ip: str = "192.168.13.22",
|
|
) -> str:
|
|
"""Build a raw log line as written by a MDC scan."""
|
|
return (
|
|
f"\\\\{server_ip}\\e\\{mdc_server_name}\\Logs\\26\\09\\22\\20260922092847.txt "
|
|
f"[{start_time}] DOSSIER EN COURS : {store_name} "
|
|
f"[{error_time}] {error_message}"
|
|
)
|
|
|
|
|
|
def test_classify_log_entries_merges_consecutive_duplicates():
|
|
"""Consecutive identical errors become a single log entry."""
|
|
log_entries = parse_lines(
|
|
[
|
|
_build_log_line("22/09/2026 09:32:19", N2_ERROR_MESSAGE),
|
|
_build_log_line("22/09/2026 09:32:20", N2_ERROR_MESSAGE),
|
|
_build_log_line("22/09/2026 09:32:25", N2_ERROR_MESSAGE),
|
|
]
|
|
)
|
|
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert len(relevant) == 1
|
|
assert relevant[0].get_error_time() == "22/09/2026 09:32:19"
|
|
assert not irrelevant
|
|
|
|
|
|
def test_classify_log_entries_merges_same_error_name():
|
|
"""Same error code with a different amount is still a single log entry."""
|
|
log_entries = parse_lines(
|
|
[
|
|
_build_log_line(
|
|
"22/09/2026 09:32:19", "Erreur : GetTvaId -> TVA non trouvée : 8,50"
|
|
),
|
|
_build_log_line(
|
|
"22/09/2026 09:32:20", "Erreur : GetTvaId -> TVA non trouvée : 20,00"
|
|
),
|
|
]
|
|
)
|
|
|
|
relevant, _ = classify_log_entries(log_entries)
|
|
|
|
assert len(relevant) == 1
|
|
|
|
|
|
def test_classify_log_entries_keeps_duplicates_from_other_store_or_scan():
|
|
"""Same error from another scan or another store is not a duplicate."""
|
|
log_entries = parse_lines(
|
|
[
|
|
_build_log_line("22/09/2026 09:32:19", N2_ERROR_MESSAGE),
|
|
_build_log_line(
|
|
"22/09/2026 10:29:10",
|
|
N2_ERROR_MESSAGE,
|
|
start_time="22/09/2026 10:28:58",
|
|
),
|
|
_build_log_line(
|
|
"22/09/2026 09:32:19", N2_ERROR_MESSAGE, store_name="DUTOUR"
|
|
),
|
|
]
|
|
)
|
|
|
|
relevant, _ = classify_log_entries(log_entries)
|
|
|
|
assert len(relevant) == 3
|
|
|
|
|
|
def test_classify_log_entries_merges_consecutive_irrelevant_duplicates():
|
|
"""Deduplication also applies to errors that do not require N2 support."""
|
|
log_entries = parse_lines(
|
|
[
|
|
_build_log_line("22/09/2026 09:32:19", OTHER_ERROR_MESSAGE),
|
|
_build_log_line("22/09/2026 09:32:24", OTHER_ERROR_MESSAGE),
|
|
]
|
|
)
|
|
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert not relevant
|
|
assert len(irrelevant) == 1
|
|
|
|
|
|
def test_classify_log_entries_keeps_same_error_separated_by_another_error():
|
|
"""An interleaved error breaks the run, so both errors are kept."""
|
|
log_entries = parse_lines(
|
|
[
|
|
_build_log_line("22/09/2026 09:32:19", N2_ERROR_MESSAGE),
|
|
_build_log_line("22/09/2026 09:32:20", OTHER_ERROR_MESSAGE),
|
|
_build_log_line("22/09/2026 09:32:21", N2_ERROR_MESSAGE),
|
|
]
|
|
)
|
|
|
|
relevant, irrelevant = classify_log_entries(log_entries)
|
|
|
|
assert len(relevant) == 2
|
|
assert len(irrelevant) == 1
|