logwatcher/tests/test_reporter.py

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, FRENCH_TIMEZONE, SourceType
from logwatcher.models import LogEntry
from logwatcher.parser import parse_file
from logwatcher.reporter import (
    N2_SUPPORT_TEMPLATE,
    OTHER_TEMPLATE,
    _render_entries,
    _render_target_report,
    build_reports,
    get_period,
    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],
        datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT),
        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
    today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
    start_date, end_date = get_period(relevant + irrelevant)
    write_log_report(
        SourceType.FILE,
        relevant,
        irrelevant,
        nb_files,
        today,
        start_date,
        end_date,
        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,
        datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT),
        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"
    today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
    start_date, end_date = get_period(get_all_log_entries[0] + get_all_log_entries[1])

    write_log_report(
        source_type=SourceType.FILE,
        relevant=get_all_log_entries[0],
        irrelevant=get_all_log_entries[1],
        nb_files=nb_files,
        today=today,
        start_date=start_date,
        end_date=end_date,
        output_dir=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(
        source_type=SourceType.FILE,
        relevant=get_all_log_entries[0],
        irrelevant=get_all_log_entries[1],
        today=today,
        start_date=start_date,
        end_date=end_date,
        nb_files=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")