test: reduce message type test overlap

Author: cole

tests/conftest.py

@@ -143,6 +143,11 @@ def message_str(recipient_str: str, sender_str: str) -> str:
     )
 
 
+@pytest.fixture(scope="session")
+def message_bytes(message_str: str) -> bytes:
+    return message_str.encode("ascii")
+
+
 @pytest.fixture(scope="session")
 def smtpd_class() -> type[SMTPD]:
     return TestSMTPD
@@ -307,16 +312,39 @@ def mime_message(
     return message
 
 
-@pytest.fixture(scope="function", params=["mime_multipart", "compat32"])
+@pytest.fixture(scope="function")
+def email_message(recipient_str: str, sender_str: str) -> email.message.EmailMessage:
+    message = email.message.EmailMessage()
+    message["To"] = recipient_str
+    message["From"] = sender_str
+    message["Subject"] = "A message"
+    message.set_payload("Hello World")
+
+    return message
+
+
+@pytest.fixture(scope="function")
 def message(
-    request: ParamFixtureRequest,
+    request: pytest.FixtureRequest,
+    email_message: email.message.EmailMessage,
     compat32_message: email.message.Message,
-    mime_message: email.message.EmailMessage,
-) -> Union[email.message.Message, email.message.EmailMessage]:
-    if request.param == "compat32":
+    mime_message: email.mime.multipart.MIMEMultipart,
+    message_str: str,
+    message_bytes: bytes,
+) -> Union[email.message.EmailMessage, email.message.Message, str, bytes]:
+    if not hasattr(request, "param"):
+        return email_message
+
+    if request.param == "compat32_message":
         return compat32_message
-    else:
+    elif request.param == "mime_message":
         return mime_message
+    elif request.param == "str":
+        return message_str
+    elif request.param == "bytes":
+        return message_bytes
+    else:
+        return email_message
 
 
 # Server helpers and factories #

tests/test_api.py

@@ -15,14 +15,17 @@
 from aiosmtplib import send
 
 
+@pytest.mark.parametrize(
+    "message", ["message", "compat32_message", "mime_message"], indirect=True
+)
 async def test_send(
     hostname: str,
     smtpd_server_port: int,
-    message: email.message.Message,
+    message: Union[email.message.EmailMessage, email.message.Message],
     received_messages: list[email.message.EmailMessage],
     client_tls_context: ssl.SSLContext,
 ) -> None:
-    errors, response = await send(
+    errors, _ = await send(
         message,
         hostname=hostname,
         port=smtpd_server_port,
@@ -33,37 +36,17 @@ async def test_send(
     assert len(received_messages) == 1
 
 
-async def test_send_with_str(
+@pytest.mark.parametrize("message", ["message_str", "message_bytes"], indirect=True)
+async def test_send_with_raw_message(
     hostname: str,
     smtpd_server_port: int,
     recipient_str: str,
     sender_str: str,
-    message_str: str,
+    message: Union[str, bytes],
     received_messages: list[email.message.EmailMessage],
 ) -> None:
-    errors, response = await send(
-        message_str,
-        hostname=hostname,
-        port=smtpd_server_port,
-        sender=sender_str,
-        recipients=[recipient_str],
-        start_tls=False,
-    )
-
-    assert not errors
-    assert len(received_messages) == 1
-
-
-async def test_send_with_bytes(
-    hostname: str,
-    smtpd_server_port: int,
-    recipient_str: str,
-    sender_str: str,
-    message_str: str,
-    received_messages: list[email.message.EmailMessage],
-) -> None:
-    errors, response = await send(
-        bytes(message_str, "ascii"),
+    errors, _ = await send(
+        message,
         hostname=hostname,
         port=smtpd_server_port,
         sender=sender_str,
@@ -80,10 +63,9 @@ async def test_send_without_sender(
     smtpd_server_port: int,
     recipient_str: str,
     message_str: str,
-    received_messages: list[email.message.EmailMessage],
 ) -> None:
     with pytest.raises(ValueError):
-        errors, response = await send(
+        await send(
             message_str,
             hostname=hostname,
             port=smtpd_server_port,
@@ -98,10 +80,9 @@ async def test_send_without_recipients(
     smtpd_server_port: int,
     sender_str: str,
     message_str: str,
-    received_messages: list[email.message.EmailMessage],
 ) -> None:
     with pytest.raises(ValueError):
-        errors, response = await send(
+        await send(
             message_str,
             hostname=hostname,
             port=smtpd_server_port,
@@ -115,11 +96,11 @@ async def test_send_with_start_tls(
     hostname: str,
     smtpd_server_port: int,
     client_tls_context: ssl.SSLContext,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
     received_commands: list[tuple[str, tuple[Any, ...]]],
 ) -> None:
-    errors, response = await send(
+    errors, _ = await send(
         message,
         hostname=hostname,
         port=smtpd_server_port,
@@ -135,14 +116,14 @@ async def test_send_with_start_tls(
 async def test_send_with_login(
     hostname: str,
     smtpd_server_port: int,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
     received_commands: list[tuple[str, tuple[Any, ...]]],
     auth_username: str,
     auth_password: str,
     client_tls_context: ssl.SSLContext,
 ) -> None:
-    errors, response = await send(
+    errors, _ = await send(
         message,
         hostname=hostname,
         port=smtpd_server_port,
@@ -160,14 +141,14 @@ async def test_send_with_login(
 async def test_send_via_socket(
     hostname: str,
     smtpd_server_port: int,
-    mime_message: email.message.EmailMessage,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
 ) -> None:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         sock.connect((hostname, smtpd_server_port))
 
         errors, _ = await send(
-            mime_message,
+            message,
             hostname=None,
             port=None,
             sock=sock,
@@ -183,14 +164,14 @@ async def test_send_via_socket_tls_and_hostname(
     hostname: str,
     client_tls_context: ssl.SSLContext,
     smtpd_server_port: int,
-    mime_message: email.message.EmailMessage,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
 ) -> None:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         sock.connect((hostname, smtpd_server_port))
 
         errors, _ = await send(
-            mime_message,
+            message,
             hostname=hostname,
             port=None,
             sock=sock,
@@ -211,11 +192,11 @@ async def test_send_via_socket_path(
     smtpd_server_socket_path: asyncio.AbstractServer,
     socket_path: pathlib.Path,
     socket_path_type: Union[type[str], type[pathlib.Path], type[bytes]],
-    mime_message: email.message.EmailMessage,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
 ) -> None:
     errors, _ = await send(
-        mime_message,
+        message,
         hostname=None,
         port=None,
         socket_path=socket_path_type(socket_path),
@@ -232,11 +213,11 @@ async def test_send_via_socket_path_with_tls(
     socket_path: pathlib.Path,
     hostname: str,
     client_tls_context: ssl.SSLContext,
-    mime_message: email.message.EmailMessage,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
 ) -> None:
     errors, _ = await send(
-        mime_message,
+        message,
         hostname=hostname,
         port=None,
         socket_path=socket_path,

tests/test_sendmail.py

@@ -6,7 +6,7 @@
 import email.generator
 import email.header
 import email.message
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 import pytest
 
@@ -248,7 +248,7 @@ async def test_send_message(smtp_client: SMTP, message: email.message.Message) -
 
 async def test_send_message_with_sender_and_recipient_args(
     smtp_client: SMTP,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
 ) -> None:
     sender = "sender2@example.com"
@@ -270,7 +270,7 @@ async def test_send_message_with_sender_and_recipient_args(
 async def test_send_message_with_cc_recipients(
     smtp_client: SMTP,
     recipient_str: str,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
     received_commands: list[tuple[str, tuple[Any, ...]]],
 ) -> None:
@@ -299,7 +299,7 @@ async def test_send_message_with_cc_recipients(
 async def test_send_message_with_bcc_recipients(
     smtp_client: SMTP,
     recipient_str: str,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
     received_commands: list[tuple[str, tuple[Any, ...]]],
 ) -> None:
@@ -324,7 +324,7 @@ async def test_send_message_with_bcc_recipients(
 async def test_send_message_with_cc_and_bcc_recipients(
     smtp_client: SMTP,
     recipient_str: str,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_messages: list[email.message.EmailMessage],
     received_commands: list[tuple[str, tuple[Any, ...]]],
 ) -> None:
@@ -354,7 +354,7 @@ async def test_send_message_with_cc_and_bcc_recipients(
 
 async def test_send_message_recipient_str(
     smtp_client: SMTP,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
     received_commands: list[tuple[str, tuple[Any, ...]]],
 ) -> None:
     recipient_str = "1234@example.org"
@@ -371,7 +371,7 @@ async def test_send_message_recipient_str(
 
 async def test_send_message_mail_options(
     smtp_client: SMTP,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
 ) -> None:
     async with smtp_client:
         errors, response = await smtp_client.send_message(
@@ -384,7 +384,7 @@ async def test_send_message_mail_options(
 
 
 async def test_send_multiple_messages_in_sequence(
-    smtp_client: SMTP, message: email.message.Message
+    smtp_client: SMTP, message: email.message.EmailMessage
 ) -> None:
     message1 = copy.copy(message)
 
@@ -407,7 +407,7 @@ async def test_send_multiple_messages_in_sequence(
 
 
 async def test_send_message_without_recipients(
-    smtp_client: SMTP, message: email.message.Message
+    smtp_client: SMTP, message: email.message.EmailMessage
 ) -> None:
     del message["To"]
 
@@ -417,7 +417,7 @@ async def test_send_message_without_recipients(
 
 
 async def test_send_message_without_sender(
-    smtp_client: SMTP, message: email.message.Message
+    smtp_client: SMTP, message: email.message.EmailMessage
 ) -> None:
     del message["From"]
 
@@ -426,10 +426,13 @@ async def test_send_message_without_sender(
             await smtp_client.send_message(message)
 
 
+@pytest.mark.parametrize(
+    "message", ["message", "compat32_message", "mime_message"], indirect=True
+)
 @pytest.mark.smtpd_options(smtputf8=True)
 async def test_send_message_smtputf8_sender(
     smtp_client: SMTP,
-    message: email.message.Message,
+    message: Union[email.message.EmailMessage, email.message.Message],
     received_commands: list[tuple[str, tuple[Any, ...]]],
     received_messages: list[email.message.EmailMessage],
 ) -> None:
@@ -510,8 +513,9 @@ async def test_send_compat32_message_smtputf8_recipient(
 
 @pytest.mark.smtpd_options(smtputf8=False)
 async def test_send_message_smtputf8_not_supported(
-    smtp_client: SMTP, message: email.message.Message
+    smtp_client: SMTP, message: email.message.EmailMessage
 ) -> None:
+    del message["To"]
     message["To"] = "reçipïént2@exåmple.com"
 
     async with smtp_client:
@@ -581,10 +585,13 @@ async def test_send_mime_message_utf8_text_without_smtputf8(
     ]
 
 
+@pytest.mark.parametrize(
+    "message", ["message", "compat32_message", "mime_message"], indirect=True
+)
 @pytest.mark.smtpd_options(**{"smtputf8": False, "7bit": True})
 async def test_send_message_7bit(
     smtp_client: SMTP,
-    message: email.message.Message,
+    message: Union[email.message.EmailMessage, email.message.Message],
     received_commands: list[tuple[str, tuple[Any, ...]]],
 ) -> None:
     async with smtp_client:

tests/test_sync.py

@@ -24,7 +24,7 @@ def test_sendmail_sync(
 
 def test_send_message_sync(
     smtp_client_threaded: SMTP,
-    message: email.message.Message,
+    message: email.message.EmailMessage,
 ) -> None:
     errors, response = smtp_client_threaded.send_message_sync(message)