diff --git a/pyproject.toml b/pyproject.toml index a7752c1..6d09525 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/src/logwatcher/reporter.py b/src/logwatcher/reporter.py index dc04a07..b23c29b 100644 --- a/src/logwatcher/reporter.py +++ b/src/logwatcher/reporter.py @@ -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) diff --git a/tests/fixtures/valid_logs/no_relevant_error.txt b/tests/fixtures/valid_logs/no_relevant_logs.txt similarity index 100% rename from tests/fixtures/valid_logs/no_relevant_error.txt rename to tests/fixtures/valid_logs/no_relevant_logs.txt diff --git a/tests/test_classifier.py b/tests/test_classifier.py index ec4047e..da501c7 100644 --- a/tests/test_classifier.py +++ b/tests/test_classifier.py @@ -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 diff --git a/tests/test_reporter.py b/tests/test_reporter.py index e69de29..26969b6 100644 --- a/tests/test_reporter.py +++ b/tests/test_reporter.py @@ -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") diff --git a/uv.lock b/uv.lock index a7984e2..6fb4f3a 100644 --- a/uv.lock +++ b/uv.lock @@ -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" }, ]