test(cleaner): ✅ add tests for cleaner module

remove also comments in `test_mail_reader.py` that didn't follow package conventions
2026-09-25 17:18:09 +02:00
parent 02e85ddbfd
commit 48503cdd42
2 changed files with 272 additions and 9 deletions
@@ -0,0 +1,272 @@
+import datetime
+import logging
+import os
+from pathlib import Path
+from unittest.mock import MagicMock
+from zipfile import ZipFile
+
+import pytest
+from exchangelib import Account, Message
+
+from logwatcher.cleaner import (
+    _get_retention_days,
+    archive_logs,
+    get_logs_to_archive,
+    prune_logs,
+    prune_mails,
+)
+from logwatcher.config import DATE_FILE_FORMAT, FRENCH_TIMEZONE
+from logwatcher.utils import ANALYZED_FOLDER, SENT_FOLDER
+
+# fixed date so retention cut-offs do not depend on the machine clock
+TODAY = datetime.datetime(2026, 9, 25, 12, 0, 0, tzinfo=FRENCH_TIMEZONE)
+
+
+@pytest.fixture(name="log_dir")
+def log_dir_fixture(tmp_path: Path, monkeypatch) -> Path:
+    """Redirect `LOGGING_PATH` to a temporary log directory.
+
+    Args:
+        tmp_path: Temporary root path for testing results
+        monkeypatch: MonkeyPatch used to fake paths
+
+    Returns:
+        the temporary log directory, created on disk
+
+    """
+    logs_dir = tmp_path / "logs"
+    logs_dir.mkdir()
+    monkeypatch.setattr("logwatcher.cleaner.LOGGING_PATH", logs_dir)
+    return logs_dir
+
+
+@pytest.fixture(name="archive_dir")
+def archive_dir_fixture(tmp_path: Path, monkeypatch) -> Path:
+    """Redirect `ARCHIVE_DIR` to a temporary directory.
+
+    The directory is left uncreated so tests can also check the missing directory case.
+
+    Args:
+        tmp_path: Temporary root path for testing results
+        monkeypatch: MonkeyPatch used to fake paths
+
+    Returns:
+        the temporary archive directory path, not created on disk
+
+    """
+    archives_dir = tmp_path / "archives"
+    monkeypatch.setattr("logwatcher.cleaner.ARCHIVE_DIR", archives_dir)
+    return archives_dir
+
+
+def _create_file(path: Path, age_in_days: int | None = None) -> Path:
+    """Create a file and optionally backdate its `st_mtime`.
+
+    Args:
+        path: path of the file to create
+        age_in_days: number of days between `TODAY` and the file modification time.
+            When None, the modification time is left untouched.
+
+    Returns:
+        path of the created file
+
+    """
+    path.write_text("log content", encoding="utf-8")
+    if age_in_days is not None:
+        timestamp = (TODAY - datetime.timedelta(days=age_in_days)).timestamp()
+        os.utime(path, (timestamp, timestamp))
+    return path
+
+
+def test_get_retention_days_returns_integer(monkeypatch):
+    """Return the number of days found in the environment variable."""
+    monkeypatch.setenv("RETENTION_DAYS_TEST", "42")
+
+    assert _get_retention_days("RETENTION_DAYS_TEST") == 42
+
+
+def test_get_retention_days_missing_variable(monkeypatch, caplog):
+    """Raise KeyError and log an error when the variable is not defined."""
+    monkeypatch.delenv("RETENTION_DAYS_TEST", raising=False)
+
+    with caplog.at_level(logging.ERROR, logger="logwatcher"):
+        with pytest.raises(KeyError):
+            _get_retention_days("RETENTION_DAYS_TEST")
+
+    assert (
+        "'RETENTION_DAYS_TEST' is not defined in environment variables."
+        in caplog.messages
+    )
+
+
+def test_get_retention_days_not_an_integer(monkeypatch, caplog):
+    """Raise ValueError and log an error when the variable is not a number."""
+    monkeypatch.setenv("RETENTION_DAYS_TEST", "thirty")
+
+    with caplog.at_level(logging.ERROR, logger="logwatcher"):
+        with pytest.raises(ValueError):
+            _get_retention_days("RETENTION_DAYS_TEST")
+
+    assert "thirty must be an integer." in caplog.messages
+
+
+def test_get_logs_to_archive_selects_old_logs_only(log_dir: Path, monkeypatch):
+    """Return only the log files older than `LOG_RETENTION_DAYS`."""
+    monkeypatch.setenv("LOG_RETENTION_DAYS", "30")
+    old_log = _create_file(log_dir / "old.log", age_in_days=40)
+    _create_file(log_dir / "recent.log", age_in_days=1)
+
+    assert get_logs_to_archive(TODAY) == [old_log]
+
+
+def test_get_logs_to_archive_ignores_everything_but_logs(log_dir: Path, monkeypatch):
+    """Ignore archives, foreign extensions and directories."""
+    monkeypatch.setenv("LOG_RETENTION_DAYS", "30")
+    (log_dir / "archives").mkdir()
+    _create_file(log_dir / "old.zip", age_in_days=40)
+    _create_file(log_dir / "old.csv", age_in_days=40)
+    expected_log = _create_file(log_dir / "old.txt", age_in_days=40)
+
+    assert get_logs_to_archive(TODAY) == [expected_log]
+
+
+def test_get_logs_to_archive_warns_when_nothing_is_old_enough(
+    log_dir: Path, monkeypatch, caplog
+):
+    """Warn and return an empty list when no log reached the retention duration."""
+    monkeypatch.setenv("LOG_RETENTION_DAYS", "30")
+    _create_file(log_dir / "recent.log", age_in_days=1)
+
+    with caplog.at_level(logging.WARNING, logger="logwatcher"):
+        logs = get_logs_to_archive(TODAY)
+
+    assert logs == []
+    assert "No log files old enough to be archived." in caplog.messages
+
+
+def test_archive_logs_creates_zip_and_deletes_originals(
+    log_dir: Path, archive_dir: Path
+):
+    """Zip the given logs in `ARCHIVE_DIR` then delete them from the log directory."""
+    first_log = _create_file(log_dir / "first.log")
+    second_log = _create_file(log_dir / "second.log")
+
+    archive_logs(TODAY, [first_log, second_log])
+
+    archived_files = list(archive_dir.iterdir())
+    assert len(archived_files) == 1
+    assert archived_files[0].name == f"archive_{TODAY.strftime(DATE_FILE_FORMAT)}.zip"
+    with ZipFile(archived_files[0]) as archive_file:
+        assert sorted(archive_file.namelist()) == ["first.log", "second.log"]
+    assert not first_log.exists()
+    assert not second_log.exists()
+
+
+def test_archive_logs_does_nothing_without_file(
+    log_dir: Path, archive_dir: Path, caplog
+):
+    """Do not create the archive directory when there is no log to archive."""
+    with caplog.at_level(logging.INFO, logger="logwatcher"):
+        archive_logs(TODAY, [])
+
+    assert not archive_dir.exists()
+    assert "No file to archive. Skip." in caplog.messages
+
+
+def test_prune_logs_deletes_old_archives_only(archive_dir: Path, monkeypatch):
+    """Delete archives older than `ARCHIVED_LOG_RETENTION_DAYS` and keep recent ones."""
+    monkeypatch.setenv("ARCHIVED_LOG_RETENTION_DAYS", "60")
+    archive_dir.mkdir()
+    old_archive = _create_file(archive_dir / "archive_old.zip", age_in_days=90)
+    recent_archive = _create_file(archive_dir / "archive_recent.zip", age_in_days=1)
+
+    prune_logs(TODAY)
+
+    assert not old_archive.exists()
+    assert recent_archive.exists()
+
+
+def test_prune_logs_creates_missing_directory(archive_dir: Path, monkeypatch, caplog):
+    """Create the archive directory then warn when it holds no archive."""
+    monkeypatch.setenv("ARCHIVED_LOG_RETENTION_DAYS", "60")
+
+    with caplog.at_level(logging.WARNING, logger="logwatcher"):
+        prune_logs(TODAY)
+
+    assert archive_dir.is_dir()
+    assert "No archive found." in caplog.messages
+
+
+def test_prune_logs_warns_on_empty_directory(archive_dir: Path, monkeypatch, caplog):
+    """Warn when the archive directory exists but holds no archive."""
+    monkeypatch.setenv("ARCHIVED_LOG_RETENTION_DAYS", "60")
+    archive_dir.mkdir()
+
+    with caplog.at_level(logging.WARNING, logger="logwatcher"):
+        prune_logs(TODAY)
+
+    assert "No archive found." in caplog.messages
+
+
+@pytest.fixture(name="mails_account")
+def mails_account_fixture() -> tuple[MagicMock, MagicMock, MagicMock]:
+    """Return a fake account exposing distinct 'Analyzed' and 'Sent' folders.
+
+    `account.msg_folder_root / LOG_FOLDER / <folder>` resolves to one mock per
+    folder, so each folder can be asserted independently.
+
+    Returns:
+        the fake account, the 'Analyzed' folder mock and the 'Sent' folder mock
+
+    """
+    account = MagicMock(spec=Account)
+    folders = {ANALYZED_FOLDER: MagicMock(), SENT_FOLDER: MagicMock()}
+    logs_folder = account.msg_folder_root.__truediv__.return_value
+    logs_folder.__truediv__.side_effect = lambda folder_name: folders[folder_name]
+    return account, folders[ANALYZED_FOLDER], folders[SENT_FOLDER]
+
+
+def test_prune_mails_filters_with_retention_cutoff(mails_account, monkeypatch):
+    """Filter both mail folders with `MAIL_RETENTION_DAYS` applied to the current day."""
+    account, analyzed_folder, sent_folder = mails_account
+    monkeypatch.setenv("MAIL_RETENTION_DAYS", "30")
+    analyzed_folder.filter.return_value = []
+    sent_folder.filter.return_value = []
+    expected_cutoff = TODAY - datetime.timedelta(days=30)
+
+    prune_mails(TODAY, account)
+
+    analyzed_folder.filter.assert_called_once_with(datetime_sent__lt=expected_cutoff)
+    sent_folder.filter.assert_called_once_with(datetime_sent__lt=expected_cutoff)
+
+
+def test_prune_mails_deletes_old_mails(mails_account, monkeypatch):
+    """Delete every old mail of both folders with a single hard delete."""
+    account, analyzed_folder, sent_folder = mails_account
+    monkeypatch.setenv("MAIL_RETENTION_DAYS", "30")
+    analyzed_mail = Message(subject="analyzed", text_body="analyzed")
+    analyzed_mail.id, analyzed_mail.changekey = "id1", "ck1"
+    sent_mail = Message(subject="sent", text_body="sent")
+    sent_mail.id, sent_mail.changekey = "id2", "ck2"
+    analyzed_folder.filter.return_value = [analyzed_mail]
+    sent_folder.filter.return_value = [sent_mail]
+
+    prune_mails(TODAY, account)
+
+    account.bulk_delete.assert_called_once_with(ids=[("id1", "ck1"), ("id2", "ck2")])
+    # `bulk_delete` must stay a hard delete: no `delete_type` nor `to_folder` argument
+    assert set(account.bulk_delete.call_args.kwargs) == {"ids"}
+
+
+def test_prune_mails_warns_when_no_old_mail(mails_account, monkeypatch, caplog):
+    """Warn and delete nothing when both folders only hold recent mails."""
+    account, analyzed_folder, sent_folder = mails_account
+    monkeypatch.setenv("MAIL_RETENTION_DAYS", "30")
+    analyzed_folder.filter.return_value = []
+    sent_folder.filter.return_value = []
+
+    with caplog.at_level(logging.WARNING, logger="logwatcher"):
+        prune_mails(TODAY, account)
+
+    assert "No mail older than '30' days found." in caplog.messages
+    account.bulk_delete.assert_not_called()
@@ -355,9 +355,6 @@ def test_connect_to_mailbox_invalid_credentials(
        assert "Connection established successfully." not in caplog.messages


-###############################
-# fetch_log_messages(account) #
-###############################
 def test_fetch_log_messages_returns_messages(make_mock_account):
    """Returns the .all() of the Logs folder."""
    mail_inline = Message(subject="mail1", text_body="text1")
@@ -376,9 +373,6 @@ def test_fetch_log_messages_folder_path(make_mock_account):
    account.msg_folder_root.__truediv__.assert_called_once_with("Logs")


-##############################
-# extract_log_lines(content) #
-##############################
 def test_extract_log_lines_split_and_clean(original_log_dir: Path):
    r"""Split content, strips '\r\n', removes empty lines."""
    file = original_log_dir / "CR_20260727110008.txt"
@@ -425,9 +419,6 @@ def test_extract_log_lines_crlf():
        assert "\r" not in line


-#############################
-# extract_logs_from_mails() #
-#############################
 def test_extract_logs_from_mails_with_attachment(original_log_dir: Path):
    """Accumulates log lines from attachment mails."""
    # create mail with attachments