maurane
c5466a4171
adapt to new naming of parsing methods, remove unused paths in parser module, all module import logwatcher.config instead of importing logwatcher, rename test functions
311 lines
11 KiB
Python
311 lines
11 KiB
Python
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from logwatcher.classifier import classify_log_entries
|
|
from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH, SourceType
|
|
from logwatcher.models import LogEntry
|
|
from logwatcher.parser import parse_file
|
|
from logwatcher.reporter import (
|
|
N2_SUPPORT_TEMPLATE,
|
|
OTHER_TEMPLATE,
|
|
_get_period,
|
|
_render_entries,
|
|
_render_target_report,
|
|
build_reports,
|
|
write_log_report,
|
|
)
|
|
|
|
# path of all CR_* files used in `get_all_log_entries_fixture`
|
|
CR_LOG_FILES = list(FIXTURE_PATH.glob("*/CR_*"))
|
|
|
|
|
|
def _get_classified_log_entries(
|
|
log_file: Path,
|
|
) -> tuple[list[LogEntry], list[LogEntry]]:
|
|
"""Parse a log file and split its entries into relevant and irrelevant lists."""
|
|
log_entries = parse_file(log_file)
|
|
return classify_log_entries(log_entries)
|
|
|
|
|
|
@pytest.fixture(name="nb_files")
|
|
def get_nb_log_files_used_fixture():
|
|
"""Return the number of CR_* log files used by the other fixtures."""
|
|
return len(CR_LOG_FILES)
|
|
|
|
|
|
@pytest.fixture(name="get_all_log_entries")
|
|
def get_all_log_entries_fixture() -> tuple[list[LogEntry], list[LogEntry]]:
|
|
"""Parse and classify every CR_* fixture, returning (relevant, irrelevant)."""
|
|
all_entries = []
|
|
for log_file in CR_LOG_FILES:
|
|
all_entries.extend(parse_file(log_file))
|
|
relevant_list, irrelevant_list = classify_log_entries(all_entries)
|
|
return relevant_list, irrelevant_list
|
|
|
|
|
|
@pytest.fixture(name="get_only_relevant_log_entries")
|
|
def get_only_relevant_log_entries_fixture(valid_log_dir: Path):
|
|
"""Return (relevant, irrelevant) for a file containing only N2-relevant errors."""
|
|
log_file = valid_log_dir / "only_relevant_logs.txt"
|
|
return _get_classified_log_entries(log_file)
|
|
|
|
|
|
@pytest.fixture(name="get_only_irrelevant_log_entries")
|
|
def get_only_irrelevant_log_entries_fixture(valid_log_dir: Path):
|
|
"""Return (relevant, irrelevant) for a file containing only non-N2 errors."""
|
|
log_file = valid_log_dir / "no_relevant_logs.txt"
|
|
return _get_classified_log_entries(log_file)
|
|
|
|
|
|
def test_get_period_empty_list():
|
|
"""An empty entry list yields an empty (start, end) period."""
|
|
assert _get_period([]) == ("", "")
|
|
|
|
|
|
def test_get_period_format(
|
|
get_all_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
):
|
|
"""Period bounds are formatted according to the configured datetime format."""
|
|
period = _get_period(get_all_log_entries[0] + get_all_log_entries[1])
|
|
for period_bound in period:
|
|
try:
|
|
assert datetime.strptime(period_bound, DATETIME_FORMAT) # noqa: DTZ007
|
|
except ValueError:
|
|
pytest.fail(f"Wrong date format for {period_bound}")
|
|
|
|
|
|
def test_render_entries_all(get_all_log_entries: tuple[list[LogEntry], list[LogEntry]]):
|
|
"""Rendered entries expose every useful field but never the raw line."""
|
|
for log_entries in get_all_log_entries:
|
|
error_report = _render_entries(log_entries)
|
|
if log_entries:
|
|
for index, log_entry in enumerate(log_entries, start=1):
|
|
assert f"[{index}]" in error_report
|
|
assert log_entry.error_name in error_report
|
|
assert log_entry.store_name in error_report
|
|
assert log_entry.get_error_time() in error_report
|
|
try:
|
|
assert datetime.strptime( # noqa: DTZ007
|
|
log_entry.get_error_time(), DATETIME_FORMAT
|
|
)
|
|
except ValueError:
|
|
pytest.fail(f"Wrong date format for {log_entry}")
|
|
assert log_entry.error_message in error_report
|
|
assert log_entry.server_ip in error_report
|
|
assert log_entry.mdc_server_name in error_report
|
|
# must not be in report file
|
|
assert log_entry.raw_line not in error_report
|
|
elif not log_entries:
|
|
assert error_report == ""
|
|
else:
|
|
pytest.fail("`log_entries` must be defined")
|
|
|
|
|
|
def test_render_entries_relevant(
|
|
get_only_relevant_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
):
|
|
"""Only-relevant input renders a non-empty relevant block and an empty other block."""
|
|
error_report_relevant = _render_entries(get_only_relevant_log_entries[0])
|
|
error_report_irrelevant = _render_entries(get_only_relevant_log_entries[1])
|
|
assert error_report_relevant != "", error_report_irrelevant == ""
|
|
|
|
|
|
def test_render_entries_irrelevant(
|
|
get_only_irrelevant_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
):
|
|
"""Only-irrelevant input renders an empty relevant block and a non-empty other block."""
|
|
error_report_relevant = _render_entries(get_only_irrelevant_log_entries[0])
|
|
error_report_irrelevant = _render_entries(get_only_irrelevant_log_entries[1])
|
|
assert error_report_relevant == "", error_report_irrelevant != ""
|
|
|
|
|
|
def test_render_target_report_relevant(
|
|
get_only_relevant_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
):
|
|
"""The N2 section header, count, and rendered entries are all present."""
|
|
relevant = get_only_relevant_log_entries[0]
|
|
target_report = _render_target_report(relevant, N2_SUPPORT_TEMPLATE)
|
|
assert "ERREURS SUPPORT N2" in target_report
|
|
assert target_report.find("Nombre d'erreur(s)\t : " + str(len(relevant))) != -1
|
|
assert _render_entries(relevant) in target_report
|
|
|
|
|
|
def test_render_target_report_irrelevant(
|
|
get_only_irrelevant_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
):
|
|
"""The "other" section header, count, and rendered entries are all present."""
|
|
irrelevant = get_only_irrelevant_log_entries[1]
|
|
target_report = _render_target_report(irrelevant, OTHER_TEMPLATE)
|
|
assert "AUTRES ERREURS" in target_report
|
|
assert target_report.find("Nombre d'erreur(s)\t : " + str(len(irrelevant))) != -1
|
|
assert _render_entries(irrelevant) in target_report
|
|
|
|
|
|
def test_render_target_report_all(
|
|
get_all_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
):
|
|
"""Each target template renders the expected section header."""
|
|
relevant, irrelevant = get_all_log_entries
|
|
relevant_report = _render_target_report(relevant, N2_SUPPORT_TEMPLATE)
|
|
irrelevant_report = _render_target_report(irrelevant, OTHER_TEMPLATE)
|
|
assert "ERREURS SUPPORT N2" in relevant_report
|
|
assert "AUTRES ERREURS" in irrelevant_report
|
|
|
|
|
|
def test_build_reports(
|
|
get_all_log_entries: tuple[list[LogEntry], list[LogEntry]], nb_files
|
|
):
|
|
"""The three reports are built with correct headers, counts, and section separation."""
|
|
start_date, end_date = _get_period(get_all_log_entries[0] + get_all_log_entries[1])
|
|
reports = build_reports(
|
|
SourceType.FILE,
|
|
get_all_log_entries[0],
|
|
get_all_log_entries[1],
|
|
start_date,
|
|
end_date,
|
|
nb_files,
|
|
)
|
|
|
|
assert "all" in reports and "n2" in reports and "other" in reports
|
|
assert reports["all"].find(f"Période\t : {start_date} -> {end_date}") != -1
|
|
assert reports["all"].find(f"Fichier(s) lu(s)\t : {nb_files}") != -1
|
|
assert (
|
|
reports["all"].find(
|
|
f"Nombre total d'erreur(s)\t: {len(get_all_log_entries[0] + get_all_log_entries[1])}"
|
|
)
|
|
!= -1
|
|
)
|
|
assert "ERREURS SUPPORT N2" in reports["all"] or "AUTRES ERREURS" in reports["all"]
|
|
assert (
|
|
"AUTRES ERREURS" not in reports["n2"] and "ERREURS SUPPORT N2" in reports["n2"]
|
|
)
|
|
assert (
|
|
"ERREURS SUPPORT N2" not in reports["other"]
|
|
and "AUTRES ERREURS" in reports["other"]
|
|
)
|
|
assert (
|
|
reports["n2"].find(f"Nombre total d'erreur(s)\t: {len(get_all_log_entries[0])}")
|
|
!= -1
|
|
)
|
|
assert (
|
|
reports["other"].find(
|
|
f"Nombre total d'erreur(s)\t: {len(get_all_log_entries[1])}"
|
|
)
|
|
!= -1
|
|
)
|
|
|
|
|
|
def _test_write_log_report(
|
|
log_entries_tuple: tuple[list[LogEntry], list[LogEntry]],
|
|
nb_files: int,
|
|
target_name_empty: str,
|
|
output_dir: Path,
|
|
):
|
|
"""Write the reports, then verify files exist and their content matches build_reports."""
|
|
results_path = output_dir / "results"
|
|
relevant, irrelevant = log_entries_tuple
|
|
write_log_report(SourceType.FILE, relevant, irrelevant, nb_files, results_path)
|
|
# check if files were created
|
|
all_file, n2_file, other_file = (
|
|
(results_path / "all.log"),
|
|
(results_path / "n2.log"),
|
|
(results_path / "other.log"),
|
|
)
|
|
assert all_file.exists()
|
|
assert n2_file.exists()
|
|
assert other_file.exists()
|
|
|
|
# check content
|
|
start_date, end_date = _get_period(relevant + irrelevant)
|
|
reports = build_reports(
|
|
SourceType.FILE, relevant, irrelevant, start_date, end_date, nb_files
|
|
)
|
|
nb_errors = {
|
|
"all": len(relevant + irrelevant),
|
|
"n2": len(relevant),
|
|
"other": len(irrelevant),
|
|
}
|
|
|
|
assert reports["all"] == all_file.read_text(encoding="windows-1252")
|
|
assert (
|
|
all_file.read_text(encoding="windows-1252").find(
|
|
f"Nombre total d'erreur(s)\t: {nb_errors['all']}"
|
|
)
|
|
!= -1
|
|
)
|
|
assert reports["n2"] == n2_file.read_text(encoding="windows-1252")
|
|
assert (
|
|
n2_file.read_text(encoding="windows-1252").find(
|
|
f"Nombre total d'erreur(s)\t: {nb_errors['n2']}"
|
|
)
|
|
!= -1
|
|
)
|
|
assert reports["other"] == other_file.read_text(encoding="windows-1252")
|
|
assert (
|
|
other_file.read_text(encoding="windows-1252").find(
|
|
f"Nombre total d'erreur(s)\t: {nb_errors['other']}"
|
|
)
|
|
!= -1
|
|
)
|
|
assert nb_errors[target_name_empty] == 0 # the target list must be empty
|
|
|
|
|
|
def test_write_log_report_relevant(
|
|
get_only_relevant_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
nb_files: int,
|
|
tmp_path: Path,
|
|
):
|
|
"""Only-relevant input writes the three files with an empty "other" list."""
|
|
_test_write_log_report(get_only_relevant_log_entries, nb_files, "other", tmp_path)
|
|
|
|
|
|
def test_write_log_report_irrelevant(
|
|
get_only_irrelevant_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
nb_files: int,
|
|
tmp_path: Path,
|
|
):
|
|
"""Only-irrelevant input writes the three files with an empty "n2" list."""
|
|
_test_write_log_report(get_only_irrelevant_log_entries, nb_files, "n2", tmp_path)
|
|
|
|
|
|
def test_write_log_report_all_logs(
|
|
get_all_log_entries: tuple[list[LogEntry], list[LogEntry]],
|
|
nb_files: int,
|
|
tmp_path: Path,
|
|
):
|
|
"""The full fixture writes three files whose content matches build_reports exactly."""
|
|
results_path = tmp_path / "results"
|
|
write_log_report(
|
|
SourceType.FILE,
|
|
get_all_log_entries[0],
|
|
get_all_log_entries[1],
|
|
nb_files,
|
|
results_path,
|
|
)
|
|
|
|
# check if files were created
|
|
all_file, n2_file, other_file = (
|
|
(results_path / "all.log"),
|
|
(results_path / "n2.log"),
|
|
(results_path / "other.log"),
|
|
)
|
|
assert all_file.exists()
|
|
assert n2_file.exists()
|
|
assert other_file.exists()
|
|
|
|
# check content
|
|
start_date, end_date = _get_period(get_all_log_entries[0] + get_all_log_entries[1])
|
|
reports = build_reports(
|
|
SourceType.FILE,
|
|
get_all_log_entries[0],
|
|
get_all_log_entries[1],
|
|
start_date,
|
|
end_date,
|
|
nb_files,
|
|
)
|
|
assert reports["all"] == all_file.read_text(encoding="windows-1252")
|
|
assert reports["n2"] == n2_file.read_text(encoding="windows-1252")
|
|
assert reports["other"] == other_file.read_text(encoding="windows-1252")
|