Add unittests. Allow to return multiple mails as reply. Move configur…

…ations to pyproject.toml

.gitignore

@@ -1,6 +1,7 @@
 *.egg-info
 *.log
-.coverage
+.coverage*
+!.coveragerc
 .mypy_cache
 .nox/
 .pytest_cache

.pre-commit-config.yaml

@@ -6,12 +6,12 @@ repos:
       - id: ruff-format
 
   - repo: https://github.com/pylint-dev/pylint
-    rev: v2.17.4
+    rev: v3.3.1
     hooks:
       - id: pylint
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.3.0
+    rev: v1.14.1
     hooks:
       - id: mypy
         # args: [--disable-error-code, union-attr]

pyproject.toml

@@ -2,6 +2,47 @@
 requires = ["setuptools >= 35.0.2", "wheel >= 0.29.0"]
 build-backend = "setuptools.build_meta"
 
+[tool.coverage]
+source = "*/mail_devel/*"
+concurrency = "multiprocessing"
+parallel = true
+
+[tool.isort]
+multi_line_output = 3
+include_trailing_comma = true
+force_grid_wrap = 0
+use_parentheses = true
+line_length = 88
+
+[tool.pylint.IMPORTS]
+ignored-modules = "aiohttp,nox,pysasl,pymap,aiosmtpd,pytest,mail_devel"
+
+[tool.pylint."MESSAGES CONTROL"]
+disable="""
+  bad-inline-option,
+  bad-mcs-classmethod-argument,
+  broad-except,
+  deprecated-pragma,
+  file-ignored,
+  invalid-name,
+  locally-disabled,
+  logging-fstring-interpolation,
+  missing-class-docstring,
+  missing-function-docstring,
+  missing-module-docstring,
+  protected-access,
+  raw-checker-failed,
+  redefined-outer-name,
+  suppressed-message,
+  too-few-public-methods,
+  too-many-arguments,
+  too-many-instance-attributes,
+  too-many-return-statements,
+  too-many-statements,
+  use-symbolic-message-instead,
+  useless-suppression,
+"""
+
 [tool.ruff]
 line-length = 88
 indent-width = 4
@@ -19,3 +60,8 @@ line-ending = "auto"
 
 [tool.mypy]
 strict = true
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
+timeout = 5

setup.py

@@ -1,3 +1,3 @@
-from setuptools import setup
+from setuptools import setup  # pylint: disable=E0401
 
 setup()

src/mail_devel/auth.py

@@ -38,7 +38,7 @@ def __init__(self, user: str, password: str, multi_user: bool = False) -> None:
         self.user, self.password = map(ensure_bytes, (user, password))
         self.multi_user = multi_user
 
-    def __call__(
+    def __call__(  # pylint: disable=R0917
         self,
         server: SMTP,
         session: Session,

src/mail_devel/automation/reply_always.py

@@ -1,6 +1,6 @@
 from mail_devel.builder import Builder
-from mail_devel.smtp import Logger, Message, Reply
+from mail_devel.smtp import Logger, Message, Reply, Response
 
 
-def reply(message: Message, flags: set[str], _logger: Logger) -> Reply | None:
-    return Reply(Builder.reply_mail(message), flags - {"Seen"})
+def reply(message: Message, flags: set[str], _logger: Logger) -> Response:
+    yield Reply(Builder.reply_mail(message), flags - {"Seen"})

src/mail_devel/automation/reply_once.py

@@ -1,9 +1,7 @@
 from mail_devel.builder import Builder
-from mail_devel.smtp import Logger, Message, Reply
+from mail_devel.smtp import Logger, Message, Reply, Response
 
 
-def reply(message: Message, flags: set[str], _logger: Logger) -> Reply | None:
+def reply(message: Message, flags: set[str], _logger: Logger) -> Response:
     if "@mail-devel" not in message.get("References", ""):
-        return Reply(Builder.reply_mail(message), flags - {"Seen"})
-
-    return None
+        yield Reply(Builder.reply_mail(message), flags - {"Seen"})

src/mail_devel/builder.py

@@ -4,6 +4,8 @@
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 
+from . import utils
+
 
 class Builder:
     """Helper class to generate mails and randomized values"""
@@ -22,9 +24,9 @@ def reply_mail(message: Message) -> Message:
         if message.is_multipart():
             for part in message.walk():
                 if part.get_content_type() == "text/plain":
-                    body = part.get_payload(decode=True).decode()
+                    body = utils.extract_payload(part, True)
         elif message.get_content_type() == "text/plain":
-            body = message.get_payload(decode=True).decode()
+            body = utils.extract_payload(message, True)
 
         reply = MIMEMultipart()
         if body:

src/mail_devel/http.py

@@ -23,7 +23,7 @@
 
 from .builder import Builder
 from .mailbox import TestMailboxDict
-from .utils import VERSION
+from .utils import VERSION, extract_payload
 
 _logger = logging.getLogger(__name__)
 
@@ -77,6 +77,7 @@ async def run_app(
     return app
 
 
+# pylint: disable=R0917
 class Frontend:
     def __init__(
         self,
@@ -159,8 +160,8 @@ async def _websocket(self, request: Request) -> WebSocketResponse:
                 continue
 
             func = getattr(self, f"on_{command}", None)
-            if callable(func):
-                await func(ws, **data)
+            if func and callable(func):
+                await func(ws, **data)  # pylint: disable=E1102
 
         _logger.info(f"Disconnected websocket: {request.remote}")
         return ws
@@ -373,7 +374,7 @@ async def on_list_mails(
             async for msg in mbox.messages()
         ]
 
-        result.sort(key=lambda x: x["date"], reverse=True)  # type: ignore
+        result.sort(key=lambda x: x["date"], reverse=True)
 
         await ws.send_json(
             {
@@ -600,7 +601,7 @@ async def _download_attachment(self, request: Request) -> Response:
                 and part.get_filename() == attachment
             ):
                 cte = part.get("Content-Transfer-Encoding")
-                body = part.get_payload(decode=bool(cte))
+                body = extract_payload(part, bool(cte))
                 return web.Response(
                     body=body,
                     headers={

src/mail_devel/smtp.py

@@ -6,7 +6,7 @@
 from email.message import Message
 from logging import Logger
 from types import ModuleType
-from typing import Callable, Iterable, Type
+from typing import Callable, Iterable, Iterator, Type
 
 from aiosmtpd.handlers import AsyncMessage
 from aiosmtpd.smtp import Envelope, Session
@@ -25,13 +25,13 @@ def __init__(self, message: Message, flags: set[str] | None = None) -> None:
         self.flags = set(flags or [])
 
 
-def dummy(_message: Message, _flags: set[str], _logger: logging.Logger) -> Reply | None:
-    """No auto respond mail"""
-    return None
+__all__ = ["Flag", "Logger", "Message", "Reply"]
+Response = Iterator[Reply]
+Responder = Callable[[Message, set[str], logging.Logger], Response]
 
 
-__all__ = ["Flag", "Logger", "Message", "Reply"]
-Responder = Callable[[Message, set[str], logging.Logger], Reply]
+def dummy(_message: Message, _flags: set[str], _logger: logging.Logger) -> Response:  # type: ignore
+    """No auto respond mail"""
 
 
 class MemoryHandler(AsyncMessage):
@@ -132,16 +132,16 @@ async def auto_respond(self, message: Message) -> None:
         if not self.responder or not callable(self.responder):
             return
 
-        reply = self.responder(
+        for reply in self.responder(
             message,
             self._default_flags(),
             _reply_logger,
-        )
-        if isinstance(reply, Reply) and reply.message:
-            await self.mailboxes.append(
-                reply.message,
-                flags=self._convert_flags(reply.flags),
-            )
+        ):
+            if isinstance(reply, Reply) and reply.message:
+                await self.mailboxes.append(
+                    reply.message,
+                    flags=self._convert_flags(reply.flags),
+                )
 
     def prepare_message(self, session: Session, envelope: Envelope) -> Message:
         if envelope.smtp_utf8 and isinstance(envelope.content, (bytes, bytearray)):

src/mail_devel/utils.py

@@ -2,8 +2,9 @@
 import os
 import ssl
 import sys
+from email.message import Message
 
-VERSION = "0.14.2"
+VERSION = "0.15.0"
 
 DEFAULT_LOG_LEVEL = "info"
 LOG_FORMAT = "{asctime} [{levelname:^8}] {message}"
@@ -100,6 +101,15 @@ def convert_size(x: str) -> int:
     return int(float(x))
 
 
+def extract_payload(message: Message, decode: bool) -> str:
+    if decode:
+        content = message.get_payload(decode=True)
+        return content.decode() if isinstance(content, bytes) else ""
+
+    content = message.get_payload(decode=False)
+    return content if isinstance(content, str) else ""
+
+
 def valid_file(path: str, is_directory: bool = False) -> str:
     """Check if a file exists and return the absolute path otherwise raise an
     error. This function is used for the argument parsing"""

tests/test_mail.py

@@ -46,7 +46,7 @@
 
 
 class ThreadResult(Thread):
-    def __init__(
+    def __init__(  # pylint: disable=R0917
         self,
         group: None = None,
         target: Callable[[int, str, str, int | None], bool] | None = None,
@@ -513,6 +513,38 @@ async def test_http_attachment() -> None:
                 assert response.status == 404
 
 
+@pytest.mark.asyncio
+async def test_responder() -> None:
+    pw = token_hex(10)
+    service = await build_test_service(pw, no_http=None)
+    assert service.handler
+
+    assert service.handler._load_responder_from_module("invalid-reply") is None
+    assert service.handler._load_responder_from_module("invalid_reply") is None
+
+    assert not service.handler.responder
+    service.handler.load_responder("reply_once")
+    assert service.handler.responder
+
+    service.handler.responder = None
+    service.handler.load_responder("reply_always")
+    assert service.handler.responder
+
+
+@pytest.mark.asyncio
+async def test_memory_handler() -> None:
+    pw = token_hex(10)
+    service = await build_test_service(pw, no_http=None)
+    assert service.handler
+
+    flag = Flag("\\Seen")
+    assert service.handler._convert_flags(None) == frozenset()
+    assert service.handler._convert_flags([]) == frozenset()
+    assert service.handler._convert_flags(["seen"]) == frozenset([flag])
+    assert service.handler._convert_flags([b"seen"]) == frozenset([flag])
+    assert service.handler._convert_flags([flag]) == frozenset([flag])
+
+
 @pytest.mark.asyncio
 async def test_smtp_auth() -> None:
     (port,) = unused_ports()