test(parser): ✅ refactor initial test, add new tests and fixtures
This commit is contained in:
+106
-24
@@ -1,33 +1,115 @@
|
|||||||
import re
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from logwatcher.config import FIXTURE_PATH
|
from logwatcher.config import FIXTURE_PATH
|
||||||
from logwatcher.parser import parse_log_line
|
from logwatcher.models import LogEntry
|
||||||
|
from logwatcher.parser import parse_log_file, parse_log_line
|
||||||
|
|
||||||
|
ORIGINAL_LOGS_DIR = FIXTURE_PATH / "original_logs"
|
||||||
|
INVALID_LOGS_DIR = FIXTURE_PATH / "invalid_logs"
|
||||||
|
SPECIFIC_LOGS_DIR = FIXTURE_PATH / "valid_logs"
|
||||||
|
|
||||||
|
@pytest.fixture(name="original_log_files")
|
||||||
|
def original_log_files_fixture() -> tuple[Path, ...]:
|
||||||
|
"""
|
||||||
|
All real log files provided by N2 technicians.
|
||||||
|
Used for integration tests: these files contain both valid lines
|
||||||
|
and lines to be ignored (system logs, empty lines, directory headers).
|
||||||
|
"""
|
||||||
|
return tuple(ORIGINAL_LOGS_DIR.iterdir())
|
||||||
|
|
||||||
|
@pytest.fixture(name="empty_log_file")
|
||||||
|
def empty_log_file_fixture() -> Path:
|
||||||
|
"""
|
||||||
|
Path of an empty log file.
|
||||||
|
Ensures the parser does not raise an exception
|
||||||
|
and returns an empty list.
|
||||||
|
"""
|
||||||
|
return Path(INVALID_LOGS_DIR / "empty_file.txt")
|
||||||
|
|
||||||
|
@pytest.fixture(name="bad_log_file")
|
||||||
|
def bad_log_file_fixture() -> 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 Path(INVALID_LOGS_DIR / "bad_log_file.txt")
|
||||||
|
|
||||||
|
@pytest.fixture(name="specific_logs")
|
||||||
|
def specific_logs_fixture() -> list[str]:
|
||||||
|
"""
|
||||||
|
Single log lines chosen to cover specific cases.
|
||||||
|
Each line is structurally valid and must be parsed
|
||||||
|
successfully. Used for unit tests of parse_log_line().
|
||||||
|
"""
|
||||||
|
log_file = Path(SPECIFIC_LOGS_DIR / "specific_logs.txt")
|
||||||
|
with open(log_file, 'r') as file:
|
||||||
|
return file.readlines()
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(name="log_data")
|
def test_parse_log_line(specific_logs: list[str]):
|
||||||
def load_logs_fixture() -> dict[str, list[str]]:
|
"""
|
||||||
file_content_list: dict[str, list[str]] = {}
|
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:
|
||||||
|
log_entry = parse_log_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 != ''
|
||||||
|
|
||||||
for file_path in FIXTURE_PATH.iterdir():
|
|
||||||
with open(file_path, 'r') as file:
|
|
||||||
file_content_list[file_path.name] = []
|
|
||||||
for line in file:
|
|
||||||
if re.match(r"\\\\", line):
|
|
||||||
file_content_list[file_path.name].append(line)
|
|
||||||
|
|
||||||
return file_content_list
|
|
||||||
|
|
||||||
def test_create_log_entry(log_data: dict[str, list[str]]):
|
def test_parse_empty_log_line():
|
||||||
for log_list in log_data.values():
|
"""
|
||||||
for log in log_list:
|
An empty or structurally invalid line returns None.
|
||||||
log_entry = parse_log_line(log)
|
Covered cases:
|
||||||
if log_entry:
|
- Empty string.
|
||||||
assert log_entry.server_ip is not None
|
- Directory header (\"Répertoire scanné : \\\\...\").
|
||||||
assert log_entry.mdc_server_name is not None
|
"""
|
||||||
assert log_entry.start_time is not None
|
log_entry = parse_log_line("")
|
||||||
assert log_entry.store_name is not None
|
assert not log_entry
|
||||||
assert log_entry.error_time is not None
|
|
||||||
assert log_entry.error_message is not None
|
log_entry = parse_log_line("Répertoire scanné : \\\\192.168.60.48\\e$\\MDC_9110\\Logs")
|
||||||
assert log_entry.raw_line != ''
|
assert not log_entry
|
||||||
|
|
||||||
|
def test_parse_valid_log_file(original_log_files: tuple[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_files:
|
||||||
|
log_list = parse_log_file(log_file)
|
||||||
|
assert len(log_list) > 0
|
||||||
|
|
||||||
|
for log_entry in log_list:
|
||||||
|
assert isinstance(log_entry, LogEntry)
|
||||||
|
|
||||||
|
def test_parse_empty_log_file(empty_log_file):
|
||||||
|
"""
|
||||||
|
An empty log file can be parsed and must returns an
|
||||||
|
empty LogEntry list.
|
||||||
|
"""
|
||||||
|
log_list = parse_log_file(empty_log_file)
|
||||||
|
assert len(log_list) == 0
|
||||||
|
|
||||||
|
def test_parse_bad_log_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_log_file(bad_log_file)
|
||||||
|
assert len(log_list) == 0
|
||||||
Reference in New Issue
Block a user