inbound_email: reorganize tests to enable sharing logic across importers

This will be useful to introduce a web-based view that handles the
message delivery, rather than a management command.

swh/web/inbound_email/tests/common_tests.py

+# Copyright (C) 2024  The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU Affero General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+
+from contextlib import contextmanager
+from email.message import EmailMessage
+import re
+from typing import Iterator
+from unittest.mock import MagicMock
+
+import pytest
+
+from django.dispatch import Signal
+
+from ..signals import EmailProcessingStatus, email_received
+
+
+@contextmanager
+def mock_signal_receiver(
+    signal: Signal, name: str = "receiver_name"
+) -> Iterator[MagicMock]:
+    receiver = MagicMock()
+    receiver.configure_mock(__name__=name, __qualname__=name)
+
+    try:
+        signal.connect(receiver)
+        yield receiver
+    finally:
+        signal.disconnect(receiver)
+
+
+@pytest.fixture
+def inbound_message(request) -> bytes:
+    marker = request.node.get_closest_marker("inbound_message")
+    if marker is None:
+        raise ValueError("Missing inbound_message data")
+    return marker.args[0]
+
+
+@pytest.mark.inbound_message(b"")
+def test_empty_stdin(caplog, receive_inbound_message):
+    receive_inbound_message()
+
+    assert len(caplog.records) == 1
+    log = caplog.records[0]
+    assert log.levelname == "ERROR"
+    assert "Unhandled message" in log.getMessage()
+
+
+def test_message() -> bytes:
+    message = EmailMessage()
+    message["to"] = "test@example.com"
+    message["subject"] = "Test Subject"
+    message.set_content("This is a test message.\n")
+
+    return bytes(message)
+
+
+TEST_MESSAGE = test_message()
+
+
+@pytest.mark.inbound_message(TEST_MESSAGE)
+@pytest.mark.parametrize(
+    "return_value,err_contents",
+    [
+        # When the email gets processed by one of the receivers, the management command
+        # should not emit any output.
+        (EmailProcessingStatus.PROCESSED, ""),
+        # When a receiver fails, the management command outputs a message to this effect
+        (EmailProcessingStatus.FAILED, "Failed receiver.*receiver_name"),
+        # When all receivers ignore a message, this fact is printed too
+        (EmailProcessingStatus.IGNORED, "Unhandled message"),
+    ],
+)
+def test_signal_receiver(return_value, err_contents, caplog, receive_inbound_message):
+    """Check that signal receivers are properly called when running the management command.
+
+    Check for output depending on its return value"""
+
+    with mock_signal_receiver(email_received) as receiver:
+        receiver.return_value = return_value
+
+        message = receive_inbound_message()
+
+        output = "\n".join(record.getMessage() for record in caplog.records)
+        if err_contents:
+            assert re.match(err_contents, output)
+        else:
+            assert output == ""
+
+        calls = receiver.call_args_list
+
+        assert len(calls) == 1
+        assert bytes(calls[0][1]["message"]) == message
+
+
+@pytest.mark.inbound_message(TEST_MESSAGE)
+def test_multiple_receivers(caplog, receive_inbound_message):
+    with mock_signal_receiver(
+        email_received, name="ignored"
+    ) as ignored, mock_signal_receiver(email_received, name="processed") as processed:
+        ignored.return_value = EmailProcessingStatus.IGNORED
+        processed.return_value = EmailProcessingStatus.PROCESSED
+
+        message = receive_inbound_message()
+
+        assert not caplog.records
+
+        for receiver in [ignored, processed]:
+            calls = receiver.call_args_list
+
+            assert len(calls) == 1
+            assert bytes(calls[0][1]["message"]) == message
+
+
+@pytest.mark.inbound_message(TEST_MESSAGE)
+def test_signal_receiver_exception(caplog, receive_inbound_message):
+    with mock_signal_receiver(email_received, name="exception_raised") as receiver:
+        receiver.side_effect = ValueError("I'm broken!")
+
+        receive_inbound_message()
+
+        output = "\n".join(
+            record.getMessage() + ("\n" + record.exc_text if record.exc_text else "")
+            for record in caplog.records
+        )
+
+        assert re.match("Failed receiver.*exception_raised", output)
+        assert "following exception" in output
+        assert "ValueError" in output
+        assert "I'm broken" in output

swh/web/inbound_email/tests/test_management_command.py

-# Copyright (C) 2022  The Software Heritage developers
+# Copyright (C) 2022-2024  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU Affero General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
-from contextlib import contextmanager
 from dataclasses import dataclass
-from email.message import EmailMessage
 from io import BytesIO, StringIO
-import re
 import sys
-from typing import Callable, Iterator
-from unittest.mock import MagicMock
 
 import pytest
 
 from django.core.management import call_command
-from django.dispatch import Signal
 
-from swh.web.inbound_email.signals import EmailProcessingStatus, email_received
+from .common_tests import *  # noqa
 
 
 class MockedStdin:
@@ -31,131 +25,28 @@ class CommandReturn:
     err: str
 
 
-@contextmanager
-def signal_receiver(signal: Signal, name: str = "receiver_name") -> Iterator[Callable]:
-    receiver = MagicMock()
-    receiver.configure_mock(__name__=name, __qualname__=name)
-
-    try:
-        signal.connect(receiver)
-        yield receiver
-    finally:
-        signal.disconnect(receiver)
+@pytest.fixture
+def receive_inbound_message(inbound_message):
+    def do_it():
+        orig_stdin = sys.stdin
+        try:
+            sys.stdin = MockedStdin()
+            sys.stdin.buffer.write(inbound_message)
+            sys.stdin.buffer.seek(0)
 
+            out = StringIO()
+            err = StringIO()
 
-def call_process_inbound_email(stdin_data: bytes) -> CommandReturn:
-    orig_stdin = sys.stdin
-    try:
-        sys.stdin = MockedStdin()  # type: ignore
-        sys.stdin.buffer.write(stdin_data)
-        sys.stdin.buffer.seek(0)
+            call_command("process_inbound_email", stdout=out, stderr=err)
 
-        out = StringIO()
-        err = StringIO()
-
-        call_command("process_inbound_email", stdout=out, stderr=err)
+            out.seek(0)
+            err.seek(0)
 
-        out.seek(0)
-        err.seek(0)
-        return CommandReturn(out=out.read(), err=err.read())
-    finally:
-        sys.stdin = orig_stdin
-
-
-def test_empty_stdin(caplog):
-    ret = call_process_inbound_email(b"")
-    assert ret.out == ""
-    assert ret.err == ""
-
-    assert len(caplog.records) == 1
-    log = caplog.records[0]
-    assert log.levelname == "ERROR"
-    assert "Unhandled message" in log.getMessage()
-
-
-@pytest.mark.parametrize(
-    "return_value,err_contents",
-    [
-        # When the email gets processed by one of the receivers, the management command
-        # should not emit any output.
-        (EmailProcessingStatus.PROCESSED, ""),
-        # When a receiver fails, the management command outputs a message to this effect
-        (EmailProcessingStatus.FAILED, "Failed receiver.*receiver_name"),
-        # When all receivers ignore a message, this fact is printed too
-        (EmailProcessingStatus.IGNORED, "Unhandled message"),
-    ],
-)
-def test_signal_receiver(return_value, err_contents, caplog):
-    """Check that signal receivers are properly called when running the management command.
-
-    Check for output depending on its return value"""
-    message = EmailMessage()
-    message["to"] = "test@example.com"
-    message["subject"] = "Test Subject"
-    message.set_content("This is a test message.\n")
-
-    with signal_receiver(email_received) as receiver:
-        receiver.return_value = return_value
-
-        ret = call_process_inbound_email(bytes(message))
-        assert ret.out == ""
-        assert ret.err == ""
-        output = "\n".join(record.getMessage() for record in caplog.records)
-        if err_contents:
-            assert re.match(err_contents, output)
-        else:
-            assert output == ""
-
-        calls = receiver.call_args_list
-
-        assert len(calls) == 1
-        assert bytes(calls[0][1]["message"]) == bytes(message)
-
-
-def test_multiple_receivers(caplog):
-    message = EmailMessage()
-    message["to"] = "test@example.com"
-    message["subject"] = "Test Subject"
-    message.set_content("This is a test message.\n")
-
-    with signal_receiver(email_received, name="ignored") as ignored, signal_receiver(
-        email_received, name="processed"
-    ) as processed:
-        ignored.return_value = EmailProcessingStatus.IGNORED
-        processed.return_value = EmailProcessingStatus.PROCESSED
-
-        ret = call_process_inbound_email(bytes(message))
-        assert ret.out == ""
-        assert ret.err == ""
-
-        assert not caplog.records
-
-        for receiver in [ignored, processed]:
-            calls = receiver.call_args_list
-
-            assert len(calls) == 1
-            assert bytes(calls[0][1]["message"]) == bytes(message)
-
-
-def test_signal_receiver_exception(caplog):
-    message = EmailMessage()
-    message["to"] = "test@example.com"
-    message["subject"] = "Test Subject"
-    message.set_content("This is a test message.\n")
-
-    with signal_receiver(email_received, name="exception_raised") as receiver:
-        receiver.side_effect = ValueError("I'm broken!")
-
-        ret = call_process_inbound_email(bytes(message))
-        assert ret.out == ""
-        assert ret.err == ""
-
-        output = "\n".join(
-            record.getMessage() + ("\n" + record.exc_text if record.exc_text else "")
-            for record in caplog.records
-        )
-
-        assert re.match("Failed receiver.*exception_raised", output)
-        assert "following exception" in output
-        assert "ValueError" in output
-        assert "I'm broken" in output
+            assert out.read() == ""
+            assert err.read() == ""
+        finally:
+            sys.stdin = orig_stdin
+
+        return inbound_message
+
+    return do_it