diff --git a/tests/test_cleaner.py b/tests/test_cleaner.py new file mode 100644 index 0000000..2415c9b --- /dev/null +++ b/tests/test_cleaner.py @@ -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 / ` 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() diff --git a/tests/test_mail_reader.py b/tests/test_mail_reader.py index b99ba83..3d12685 100644 --- a/tests/test_mail_reader.py +++ b/tests/test_mail_reader.py @@ -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