maurane/logwatcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✅test(reporter): add tests for reporter module

Browse Source 
add tests, reformat code, add an output variable in writing function. All reading and writing functions must read in windows-1252 encoding because original logs are windows-1252 encoded
This commit is contained in:
maurane glaude
2026-08-27 17:34:22 +02:00
parent b83ee451a3
commit 9c06222f37
6 changed files with 379 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2 -3
View File
@@ -4,15 +4,14 @@ version = "0.1.0"
description = "Trier, classifier et extraire automatiquement les entrées pertinentes d'un fichier de logs applicatif."
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"pytest-cov>=7.1.0",
]
dependencies = []
[dependency-groups]
dev = [
"ruff>=0.16.0",
"mypy>=2.3.0",
"pytest>=9.1.1",
"pytest-cov>=7.1.0",
]
[project.scripts]
src/logwatcher/reporter.py
+80 -80
View File
@@ -1,5 +1,5 @@
import json
from datetime import datetime
from pathlib import Path
from string import Template
from logwatcher.config import DATETIME_FORMAT, RESULT_PATH
@@ -9,24 +9,24 @@ from logwatcher.models import LogEntry
BASE_TEMPLATE = Template("""
RAPPORT D'ANALYSE DE LOGS
=========================
Période\t : $start_date → $end_date
Période\t : $start_date -> $end_date
Fichiers lu\t : $nb_files
Nombre total d'erreurs\t: $nb_errors
$sub_content
""")
N2_SUPPORT_TEMPLATE = Template("""
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
=========================
ERREURS SUPPORT N2
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
=========================
Nombre d'erreurs\t : $nb_errors
$error_list
""")
OTHER_TEMPLATE = Template("""
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
=========================
AUTRES ERREURS
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
=========================
Nombre d'erreurs\t : $nb_errors
$error_list
""")
@@ -34,6 +34,8 @@ $error_list
ERROR_TEMPLATE = Template("""
[$index] $error_name
\tMagasin: $store_name
\tMDC: $mdc
\tServeur: $server_ip
\tHeure: $error_time
\tRaison: $error_message
""")
@@ -45,99 +47,97 @@ def _get_period(entries: list[LogEntry]) -> tuple[str, str]:
times: list[datetime] = [e.error_time for e in entries]
return min(times).strftime(DATETIME_FORMAT), max(times).strftime(DATETIME_FORMAT)
def write_log_report(relevant: list[LogEntry], irrelevant: list[LogEntry], nb_files: int):
def write_log_report(
relevant: list[LogEntry],
irrelevant: list[LogEntry],
nb_files: int,
output_dir: Path = RESULT_PATH,
) -> None:
"""
Write each relevant, irrelevant and general reports in their respective output
file.
Use the range date of relevant and irrelevant lists to get the period of time
the logs were generated.
Write each relevant, irrelevant and general reports in their respective output
file.
Use the range date of relevant and irrelevant lists to get the period of time
the logs were generated.
"""
end_date, start_date = _get_period(relevant + irrelevant)
start_date, end_date = _get_period(relevant + irrelevant)
reports_dict = build_reports(relevant, irrelevant, start_date, end_date, nb_files)
for target, content in reports_dict.items():
target_file = RESULT_PATH / str(target + ".log")
target_file.write_text(content, encoding="windows-1252")
for target in reports_dict:
output_dir.mkdir(parents=True, exist_ok=True)
target_file = output_dir / str(target + ".log")
target_file.write_text(reports_dict[target], encoding="windows-1252")
def generate_json_report(relevant: list[LogEntry]) -> str: # TODO: put it in a serializer module
def build_reports(
relevant: list[LogEntry], irrelevant: list[LogEntry], start_date, end_date, nb_files
) -> dict[str, str]:
"""
Serialize from a Python LogEntry object to a JSON object.
Used in the Ivanti API calls
"""
return json.dumps(relevant)
def build_reports(relevant: list[LogEntry], irrelevant: list[LogEntry], start_date, end_date, nb_files) -> dict[str, str]:
"""
Build the three output reports: n2, other, and all.
"all" report contains both "n2" and "other" sections.
"n2" and "other" are two unique sections.
Build the three output reports: n2, other, and all.
"all" report contains both "n2" and "other" sections.
"n2" and "other" are two unique sections.
"""
relevant_report = _render_target_report(
log_entries=relevant,
target_template=N2_SUPPORT_TEMPLATE
log_entries=relevant, target_template=N2_SUPPORT_TEMPLATE
)
irrelevant_report = _render_target_report(
log_entries=irrelevant,
target_template=OTHER_TEMPLATE
log_entries=irrelevant, target_template=OTHER_TEMPLATE
)
return {
"n2": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(relevant),
sub_content=relevant_report,
),
"other": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(irrelevant),
sub_content=irrelevant_report,
),
"all": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(relevant + irrelevant),
sub_content=relevant_report + "\n" + irrelevant_report,
)
}
def _render_entries(log_entries: list[LogEntry]) -> str:
"""
Render a list of log entries using the `ERROR_TEMPLATE`.
Used in generated report files in the `output` directory
"""
error_list = []
for index, log_entry in enumerate(log_entries, start=1):
error_list.append(
ERROR_TEMPLATE.substitute(
index=index,
error_name=log_entry.error_name,
store_name=log_entry.store_name,
error_time=log_entry.get_error_time(),
error_message=log_entry.error_message
)
)
return "".join(error_list)
"n2": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(relevant),
sub_content=relevant_report,
),
"other": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(irrelevant),
sub_content=irrelevant_report,
),
"all": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(relevant + irrelevant),
sub_content=relevant_report + "\n" + irrelevant_report,
),
}
# build templates
def _render_target_report(log_entries: list[LogEntry], target_template) -> str:
"""
Render report using either N2_SUPPORT_TEMPLATE or OTHER_TEMPLATE.
Call _render_entries() and include rendered log entries in the
new report.
Render report using either N2_SUPPORT_TEMPLATE or OTHER_TEMPLATE.
Call _render_entries() and include rendered log entries in the
new report.
"""
error_list = _render_entries(log_entries)
return target_template.substitute(
nb_errors=len(log_entries),
error_list=error_list
)
return target_template.substitute(nb_errors=len(log_entries), error_list=error_list)
def _render_entries(log_entries: list[LogEntry]) -> str:
"""
Render a list of log entries using the `ERROR_TEMPLATE`.
Used in generated report files in the `output` directory
"""
error_list = []
for index, log_entry in enumerate(log_entries, start=1):
error_list.append(
ERROR_TEMPLATE.substitute(
index=index,
error_name=log_entry.error_name,
store_name=log_entry.store_name,
error_time=log_entry.get_error_time(),
error_message=log_entry.error_message,
mdc=log_entry.mdc_server_name,
server_ip=log_entry.server_ip,
)
)
return "".join(error_list)
tests/fixtures/valid_logs/no_relevant_error.txt → tests/fixtures/valid_logs/no_relevant_logs.txt
Vendored
View File
tests/test_classifier.py
+1 -1
View File
@@ -104,7 +104,7 @@ 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_error.txt"
log_file = valid_log_dir / "no_relevant_logs.txt"
log_entries = parse_log_file(log_file)
# all entries are relevant
tests/test_reporter.py
+294
View File
@@ -0,0 +1,294 @@
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
from logwatcher.models import LogEntry
from logwatcher.parser import parse_log_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_log_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_log_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'erreurs\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'erreurs\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(
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"Fichiers lu\t : {nb_files}") != -1
assert (
reports["all"].find(
f"Nombre total d'erreurs\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'erreurs\t: {len(get_all_log_entries[0])}")
!= -1
)
assert (
reports["other"].find(
f"Nombre total d'erreurs\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(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(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'erreurs\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'erreurs\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'erreurs\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(
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(
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")
uv.lock
Generated
+2 -4
View File
@@ -301,24 +301,22 @@ wheels = [
name = "logwatcher"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "pytest-cov" },
]
[package.dev-dependencies]
dev = [
{ name = "mypy" },
{ name = "pytest" },
{ name = "pytest-cov" },
{ name = "ruff" },
]
[package.metadata]
requires-dist = [{ name = "pytest-cov", specifier = ">=7.1.0" }]
[package.metadata.requires-dev]
dev = [
{ name = "mypy", specifier = ">=2.3.0" },
{ name = "pytest", specifier = ">=9.1.1" },
{ name = "pytest-cov", specifier = ">=7.1.0" },
{ name = "ruff", specifier = ">=0.16.0" },
]