1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
# -*- coding: utf-8 -*-
import asyncio
import sys
try:
from aiosmtpd.controller import Controller
except ImportError:
raise ImportError("Please install 'aiosmtpd' library to use mail_debug command.")
from logging import getLogger
from typing import List
from django.core.management.base import BaseCommand, CommandError
from django_extensions.management.utils import setup_logger, signalcommand
logger = getLogger(__name__)
class CustomHandler:
async def handle_DATA(self, server, session, envelope):
"""Output will be sent to the module logger at INFO level."""
peer = session.peer
inheaders = 1
lines = envelope.content.decode("utf8", errors="replace").splitlines()
logger.info("---------- MESSAGE FOLLOWS ----------")
for line in lines:
# headers first
if inheaders and not line:
logger.info("X-Peer: %s" % peer[0])
inheaders = 0
logger.info(line)
logger.info("------------ END MESSAGE ------------")
return "250 OK"
class Command(BaseCommand):
help = "Starts a test mail server for development."
args = "[optional port number or ippaddr:port]"
requires_system_checks: List[str] = []
def add_arguments(self, parser):
super().add_arguments(parser)
parser.add_argument("addrport", nargs="?")
parser.add_argument(
"--output",
dest="output_file",
default=None,
help="Specifies an output file to send a copy of all messages "
"(not flushed immediately).",
)
parser.add_argument(
"--use-settings",
dest="use_settings",
action="store_true",
default=False,
help="Uses EMAIL_HOST and HOST_PORT from Django settings.",
)
@signalcommand
def handle(self, addrport="", *args, **options):
if not addrport:
if options["use_settings"]:
from django.conf import settings
addr = getattr(settings, "EMAIL_HOST", "")
port = str(getattr(settings, "EMAIL_PORT", "1025"))
else:
addr = ""
port = "1025"
else:
try:
addr, port = addrport.split(":")
except ValueError:
addr, port = "", addrport
if not addr:
addr = "127.0.0.1"
if not port.isdigit():
raise CommandError("%r is not a valid port number." % port)
else:
port = int(port)
# Add console handler
setup_logger(logger, stream=self.stdout, filename=options["output_file"])
def inner_run():
quit_command = (sys.platform == "win32") and "CTRL-BREAK" or "CONTROL-C"
print(
"Now accepting mail at %s:%s -- use %s to quit"
% (addr, port, quit_command)
)
handler = CustomHandler()
controller = Controller(handler, hostname=addr, port=port)
controller.start()
loop = asyncio.get_event_loop()
loop.run_forever()
try:
inner_run()
except KeyboardInterrupt:
pass
|
