maurane/logwatcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✅test(project): add mail_reader tests

Browse Source 
fix unprecised typings replacing QuerySet type by list[Message] type, exclude test directory from mypy analysis and add tests for mail_reader.
This commit is contained in:
maurane
2026-09-09 19:09:38 +02:00
parent 150eb5145d
commit 32df98e64b
3 changed files with 530 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mail_reader.py
+521
View File
@@ -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 == []