diff --git a/pyproject.toml b/pyproject.toml index 3b4e655..81e7820 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,3 +31,8 @@ packages = ["src/logwatcher"] [tool.pytest.ini_options] testpaths = ["tests"] + +[tool.mypy] +exclude = [ + "^tests/" +] diff --git a/src/logwatcher/mail_reader.py b/src/logwatcher/mail_reader.py index 42b68bf..d70127a 100644 --- a/src/logwatcher/mail_reader.py +++ b/src/logwatcher/mail_reader.py @@ -13,7 +13,6 @@ from exchangelib import ( Version, ) from exchangelib.errors import UnauthorizedError -from exchangelib.queryset import QuerySet logger = logging.getLogger(__name__) @@ -129,7 +128,7 @@ def connect_to_mailbox() -> Account: return account -def fetch_log_messages(account: Account) -> QuerySet: +def fetch_log_messages(account: Account) -> list[Message]: """ Retrieve log messages from the log folder. @@ -140,7 +139,7 @@ def fetch_log_messages(account: Account) -> QuerySet: queryset of log-related mails """ logs_folder = account.msg_folder_root / MAILBOX_DIR - return logs_folder.all() + return list(logs_folder.all()) def extract_log_lines(content: str) -> list[str]: @@ -157,18 +156,15 @@ def extract_log_lines(content: str) -> list[str]: return [line.strip("\r\n") for line in content.splitlines() if line.strip()] -def extract_logs_from_mails(messages: QuerySet) -> list[str]: +def extract_logs_from_mails(messages: list[Message]) -> list[str]: """ - Connect to the mailbox, retrieves emails that are N2 log-related, + Retrieves emails that are N2 log-related, clean them and return a list of logs in string format. Returns: list of logs in string format """ logs = [] - account = connect_to_mailbox() - messages = fetch_log_messages(account) - for message in messages: if _has_attachment_marker(message.text_body): attachment = _find_log_attachment(message) diff --git a/tests/test_mail_reader.py b/tests/test_mail_reader.py new file mode 100644 index 0000000..465e6d2 --- /dev/null +++ b/tests/test_mail_reader.py @@ -0,0 +1,521 @@ +import logging +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from exchangelib import FileAttachment, ItemAttachment, Message +from exchangelib.errors import UnauthorizedError +from pytest import MonkeyPatch + +# noinspection protected-member +from logwatcher.mail_reader import ( + LOG_IN_ATTACHMENT_PATTERN, + _find_log_attachment, + _get_attachment_content, + _get_mail_body, + _has_attachment_marker, + connect_to_mailbox, + extract_log_lines, + extract_logs_from_mails, + fetch_log_messages, +) + + +@pytest.fixture(name="empty_mail") +def get_empty_mail_fixture() -> Message: + """A fake Message""" + return Message(subject="test without attachment", text_body="test") + + +@pytest.fixture(name="log_mail_with_attachment") +def get_log_mail_with_attachment(original_log_dir: Path) -> Message: + """A fake Message with a .txt file attachment.""" + file = original_log_dir / "CR_20260727110008.txt" + + return Message( + subject="test with logs in attachment", + text_body="inline attachment here", + attachments=[ + FileAttachment( + name="image.png", + content=b"...", + is_inline=True, + ), + FileAttachment(name=file.name, content=file.read_bytes(), is_inline=False), + ], + ) + + +@pytest.fixture(name="log_mail_with_body") +def get_log_mail_with_body() -> Message: + """A fake Message with logs in text body.""" + return Message( + subject="test with logs in text body", + text_body="Compte-rendu du suivi des imports NOSYMAG du 02/09/2026 à 15:00:10\n\n\n" + "Répertoire scanné : \\192.168.60.40\\e$\\MDC_1110\\Logs\r\n" + "\\192.168.60.41\\e$\\MDC_2110\\Logs\26\08\31\20260831234232.txt [31/08/2026 23:42:39] ExecuteProcess -> GetMasterDataFiles -> UnZipFile -> Erreur : le fichier [E:\\MDC_2110\\Master\\ATELIER_v20260806t233448.zip] n'existe pas !\r\n" + "Répertoire scanné : \\192.168.60.42\\e$\\MDC_3110\\Logs\r\n" + "\\192.168.60.42\\e$\\MDC_3110\\Logs\26\08\31\20260831235503.txt [31/08/2026 23:55:10] ExecuteProcess -> GetMasterDataFiles -> UnZipFile -> Erreur : le fichier [E:\\MDC_3110\\Master\\ATELIER_v20260806t233448.zip] n'existe pas !\r\n" + "Répertoire scanné : \\192.168.60.43\\e$\\MDC_4110\\Logs\r\n" + "\\192.168.60.43\\e$\\MDC_4110\\Logs\26\08\31\20260831233518.txt [31/08/2026 23:35:25] ExecuteProcess -> GetMasterDataFiles -> UnZipFile -> Erreur : le fichier [E:\\MDC_4110\\Master\\ATELIER_v20260806t233448.zip] n'existe pas !\r\n" + "Répertoire scanné : \\192.168.60.44\\e$\\MDC_5110\\Logs\r\n" + "\\192.168.60.44\\e$\\MDC_5110\\Logs\26\08\31\20260831233628.txt [31/08/2026 23:36:37] ExecuteProcess -> GetMasterDataFiles -> UnZipFile -> Erreur : le fichier [E:\\MDC_5110\\Master\\ATELIER_v20260806t233448.zip] n'existe pas !\r\n", + attachments=[ + FileAttachment( + name="image.png", + content=b"...", + is_inline=True, + ), + ], + ) + + +@pytest.fixture +def make_mock_account(): + """A fake Account with a msg_folder_root returning a fake Logs folder.""" + + def _make_mock_account(messages: list[Message] | None = None): + account = MagicMock() + logs_folder = account.msg_folder_root.__truediv__.return_value + logs_folder.all.return_value = messages or [] + return account + + return _make_mock_account + + +@pytest.fixture +def set_env(monkeypatch): + """Set the mailbox environment variables.""" + env = { + "EMAIL": "test@client.com", + "PASSWORD": "passtest", + "EWS_URL": "https://mail.client.com/EWS/Exchange.asmx", + } + + for key, value in env.items(): + monkeypatch.setenv(key, value) + + +class FakeProtocol: + """Protocol whose .version access raises UnauthorizedError.""" + + @property + def version(self): + raise UnauthorizedError("invalid credentials") + + +class FakeAccount: + """Account with a protocol that raises on version access.""" + + def __init__(self): + self.protocol = FakeProtocol() + + +class OkProtocol: + """Protocol whose .version succeeds.""" + + @property + def version(self): + return MagicMock() + + +class OkAccount: + """Account with a protocol that returns a version successfully.""" + + def __init__(self): + self.protocol = OkProtocol() + + +######################################### +# _find_log_attachment # +######################################### +def test_find_log_attachment_returns_log_file( + empty_mail: Message, +): + """Returns the CR_*.txt attachment, not the inline image.""" + image_att = FileAttachment(name="image.png", content=b"...", is_inline=True) + log_att = FileAttachment(name="CR_123.txt", content=b"...", is_inline=False) + + empty_mail.attach((image_att, log_att)) + attachment = _find_log_attachment(empty_mail) + assert attachment is not None + assert attachment.name == log_att.name + + +def test_find_log_attachment_ignores_inline_image(empty_mail: Message): + """When the log attachment is inline, returns None.""" + empty_mail.attach(FileAttachment(name="image.png", content=b"", is_inline=True)) + attachment = _find_log_attachment(empty_mail) + assert attachment is None + + +def test_find_log_attachment_no_log_file(empty_mail): + """Returns None when no CR_*.txt attachment exists.""" + attachment = _find_log_attachment(empty_mail) + assert attachment is None + + +def test_find_log_attachment_accepts_no_extension(empty_mail: Message): + """Accepts a CR_* attachment without extension.""" + empty_mail.attach(FileAttachment(name="CR_123", content=b"...", is_inline=False)) + attachment = _find_log_attachment(empty_mail) + assert attachment is not None + assert "." not in attachment.name + + +def test_find_log_attachment_rejects_other_names( + empty_mail: Message, +): + """ + Rejects attachments not starting with CR_ even with .txt + or .log extension or extensionless. + """ + attachments = ( + FileAttachment( + name="attachment.pdf", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="attachment.xml", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="attachment.test", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="attachment.conf", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="attachment.txt", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="attachment.log", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="attachment", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="CR_attachment.pdf", + content=b"...", + is_inline=False, + ), + FileAttachment( + name="CR_attachment.xml", + content=b"...", + is_inline=False, + ), + ) + + empty_mail.attach(attachments) + assert len(empty_mail.attachments) == len(attachments) + attachment = _find_log_attachment(empty_mail) + assert attachment is None + + +def test_find_log_attachment_rejects_non_file_attachment(empty_mail: Message): + """Skips ItemAttachment (not a FileAttachment).""" + empty_mail.attach( + ItemAttachment( + item=Message(subject="a sub message", text_body="this is a sub message") + ) + ) + assert len(empty_mail.attachments) == 1 + attachment = _find_log_attachment(empty_mail) + assert attachment is None + + +############################ +# _get_attachment_content # +############################ +def test_get_attachment_content_decodes_windows_1252(original_log_dir: Path): + """Decodes content with accents correctly.""" + file_with_accents = original_log_dir / "CR_20260727110008.txt" + assert file_with_accents.is_file() + + file_content = file_with_accents.read_bytes() + attachment = FileAttachment(content=file_content) + content = _get_attachment_content(attachment) + assert "�" not in content + + +def test_get_attachment_content_raises_when_none(): + """Raises ValueError when attachment.content is None.""" + attachment = FileAttachment(content=None) + with pytest.raises(ValueError): + _get_attachment_content(attachment) + + +def test_get_attachment_content_empty_bytes(): + """Decodes empty bytes to empty string without error.""" + attachment = FileAttachment(content=b"") + content = _get_attachment_content(attachment) + assert content == "" + + +######################## +# _get_mail_body(mail) # +######################## +def test_get_mail_body_from_marker_to_end(empty_mail: Message): + """Returns body starting at 'Répertoire scanné'.""" + empty_mail.text_body += "Répertoire scanné : \\192.168.60.40\\e$\\MDC_1110\\Logs" + + body = _get_mail_body(empty_mail) + assert "test" not in body + assert "Répertoire scanné : \\192.168.60.40\\e$\\MDC_1110\\Logs" == body + + +def test_get_mail_body_no_marker_returns_empty(empty_mail: Message): + """Returns empty string and logs warning when marker absent.""" + body = _get_mail_body(empty_mail) + assert body == "" + + +def test_get_mail_body_marker_at_start(empty_mail: Message): + """Returns whole body when marker is at position 0.""" + initial_text_body = empty_mail.text_body + empty_mail.text_body = "Répertoire scanné : " + initial_text_body + content = _get_mail_body(empty_mail) + assert initial_text_body in content + assert content == "Répertoire scanné : " + initial_text_body + + +def test_get_mail_body_multiple_markers(empty_mail: Message): + """Returns from the FIRST 'Répertoire scanné' occurrence.""" + initial_text_body = empty_mail.text_body + empty_mail.text_body = ( + initial_text_body + + "Répertoire\nLorem Ipsum\nRépertoire scanné\nRépertoire scanné\nRépertoire scanné" + ) + + content = _get_mail_body(empty_mail) + assert content != "" + assert len(content) < len(empty_mail.text_body) + assert initial_text_body not in content + assert "Répertoire scanné\nRépertoire scanné\nRépertoire scanné" == content + + +################################ +# _has_attachment_marker(body) # +################################ +def test_has_attachment_marker_true(): + """Returns True when the exact pattern is present.""" + body_with_pattern = "Compte-rendu du suivi des imports NOSYMAG du 02/09/2026 à 15:00:10\n\nLe compte-rendu contient plus de 100 lignes.\nSe reporter à la pièce jointe." + assert _has_attachment_marker(body_with_pattern) + + +def test_has_attachment_marker_false(): + """Returns False when pattern is absent.""" + body_without_pattern = ( + "Compte-rendu du suivi des imports NOSYMAG du 01/09/2026 à 03:00:10\n\nRépertoire scanné : " + "\\192.168.60.40\\e\\$\\MDC_1110\\Logs\n\nRépertoire scanné : \\192.168.60.41\\e$\\MDC_2110" + "\\Logs\\192.168.60.41\\e$\\MDC_2110\\Logs\\26\\08\\31\\20260831234232.txt [31/08/2026 " + "23:42:39] ExecuteProcess -> GetMasterDataFiles -> UnZipFile -> Erreur : le fichier " + "[E:\\MDC_2110\\Master\\ATELIER_v20260806t233448.zip] n'existe pas !" + ) + assert not _has_attachment_marker(body_without_pattern) + + +def test_has_attachment_marker_empty_string(): + """Returns False on empty body.""" + body_empty = "" + assert not _has_attachment_marker(body_empty) + + +######################## +# connect_to_mailbox() # +######################## +def test_connect_to_mailbox_success(monkeypatch: MonkeyPatch, caplog, set_env): + """Returns an Account when credentials are valid.""" + with monkeypatch.context() as m, caplog.at_level(logging.INFO, "logwatcher"): + m.setattr("logwatcher.mail_reader.Account", lambda *a, **k: OkAccount()) + account = connect_to_mailbox() + assert account is not None + assert "Connection failed: invalid credentials." not in caplog.messages + assert "Connection established successfully." in caplog.messages + + +@pytest.mark.parametrize("missing", ["EMAIL", "PASSWORD", "EWS_URL"]) +def test_connect_to_mailbox_missing_env( + monkeypatch: MonkeyPatch, missing: str, set_env +): + """Raises KeyError when an env var is missing.""" + monkeypatch.delenv(missing, raising=False) + + with pytest.raises(KeyError): + connect_to_mailbox() + + +def test_connect_to_mailbox_invalid_credentials( + monkeypatch: MonkeyPatch, caplog, set_env +): + """Logs error and re-raises on UnauthorizedError.""" + with caplog.at_level(logging.ERROR, "logwatcher"): + with monkeypatch.context() as m: + m.setattr("logwatcher.mail_reader.Account", lambda *a, **k: FakeAccount()) + with pytest.raises(UnauthorizedError): + connect_to_mailbox() + assert "Connection failed: invalid credentials." in caplog.messages + 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") + mail_not_inline = Message(subject="mail2", text_body="text2") + account = make_mock_account([mail_inline, mail_not_inline]) + + results = fetch_log_messages(account) + + assert results == [mail_inline, mail_not_inline] + + +def test_fetch_log_messages_folder_path(make_mock_account): + """Accesses msg_folder_root / 'Logs'.""" + account = make_mock_account() + fetch_log_messages(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): + """Splits content, strips \r\n, removes empty lines.""" + file = original_log_dir / "CR_20260727110008.txt" + content = file.read_text(encoding="windows-1252") + lines = extract_log_lines(content) + + assert len(lines) > 1 + for line in lines: + assert "\n" not in line + assert "\r" not in line + assert line != "" + + +def test_extract_log_lines_empty_content(): + """Returns [] on empty string.""" + content = "" + lines = extract_log_lines(content) + + assert lines == [] + + +def test_extract_log_lines_no_trailing_newline(): + """Handles content not ending with a newline.""" + content = ( + "Répertoire scanné : \\192.168.60.40\\e$\\MDC_1110\\Logs\n" + "\\192.168.60.40\\e$\\MDC_1110\\Logs\\26\\07\\27\\20260727083117.txt [27/07/2026 08:31:26] DOSSIER EN COURS : FLAMMIER [27/07/2026 08:31:53] Erreur FTP commande : Requested action not taken" + ) + lines = extract_log_lines(content) + + assert len(lines) > 1 + + +def test_extract_log_lines_crlf(): + """Handles Windows \r\n line endings.""" + content = ( + "Répertoire scanné : \\192.168.60.40\\e$\\MDC_1110\\Logs\n" + "\\192.168.60.40\\e$\\MDC_1110\\Logs\\26\\07\\27\\20260727083117.txt [27/07/2026 08:31:26] DOSSIER EN COURS : FLAMMIER [27/07/2026 08:31:53] Erreur FTP commande : Requested action not taken\r\n" + ) + lines = extract_log_lines(content) + assert len(lines) > 1 + for line in lines: + assert line != "" + assert "\n" not in line + 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 + messages = [] + + for index, file in enumerate(original_log_dir.iterdir()): + file_content = file.read_bytes() + attachment = FileAttachment( + name=file.name, content=file_content, is_inline=False + ) + message = Message( + subject=f"Compte-rendu du suivi des imports NOSYMAG du {index}", + text_body=f"Compte-rendu du suivi des imports NOSYMAG du {index}\n" + f"{LOG_IN_ATTACHMENT_PATTERN}\n" + f"Se reporter à la pièce jointe.", + attachments=[attachment], + ) + messages.append(message) + + content_list = extract_logs_from_mails(messages) + + assert len(content_list) > 1 + + for content in content_list: + assert content != "" + + +def test_extract_logs_from_mails_inline(log_mail_with_body: Message): + """Accumulates log lines from body-only mails.""" + messages = [log_mail_with_body] + + content_list = extract_logs_from_mails(messages) + + assert len(content_list) > 1 + for content in content_list: + assert content != "" + + +def test_extract_logs_from_mails_missing_attachment_logs_warning(caplog): + """Warns and skips when marker says attachment but none found.""" + subject = "Compte-rendu du suivi des imports NOSYMAG" + messages = [ + Message( + subject=subject, + text_body=LOG_IN_ATTACHMENT_PATTERN, + attachments=[ + FileAttachment(name="image.jpg", content=b"...", is_inline=True) + ], + ), + ] + + with caplog.at_level(logging.WARNING, logger="logwatcher"): + extract_logs_from_mails(messages) + + assert ( + f"Mail '{subject}' says attachment but none were found." in caplog.messages + ) + + +def test_extract_logs_from_mails_combines_multiple( + log_mail_with_attachment: Message, log_mail_with_body: Message +): + """Combines logs from several mails (attachment + inline).""" + messages = [log_mail_with_attachment, log_mail_with_body] + + content_list = extract_logs_from_mails(messages) + + assert len(content_list) > 1 + for content in content_list: + assert content != "" + + +def test_extract_logs_from_mails_no_mails(): + """Returns [] when no mails in folder.""" + content_list = extract_logs_from_mails([]) + assert content_list == []