✅test(parser): update and add tests

update existing tests and add tests for `parse_lines`
This commit is contained in:
2026-09-10 11:34:20 +02:00
parent c5466a4171
commit 4a81b71eac
+47 -12
View File
@@ -3,7 +3,7 @@ from pathlib import Path
import pytest
from logwatcher.models import LogEntry
from logwatcher.parser import parse_file, parse_line
from logwatcher.parser import parse_file, parse_line, parse_lines
@pytest.fixture(name="empty_log_file")
@@ -28,24 +28,25 @@ def bad_log_file_fixture(invalid_log_dir: Path) -> Path:
@pytest.fixture(name="specific_logs")
def specific_logs_fixture(valid_log_dir: Path) -> list[str]:
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().
"""
log_file = valid_log_dir / "specific_logs.txt"
with open(log_file, "r") as file:
return file.readlines()
return valid_log_dir / "specific_logs.txt"
def test_parse_line(specific_logs: list[str]):
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.
"""
for log_line in specific_logs:
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
@@ -72,21 +73,55 @@ def test_parse_empty_line():
def test_parse_lines():
pass
# 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_empty_lines():
pass
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_valid_file(original_log_dir: Path):
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 log_file in original_log_dir.glob("CR_*"):
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