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
|
"""Testing helpers."""
import sys
import socket
import struct
import asyncio
import logging
import warnings
from contextlib import ExitStack
from unittest import TestCase
from unittest.mock import patch
def reset_connection(client):
# Close the connection with a TCP RST instead of a TCP FIN. client must
# be a smtplib.SMTP instance.
#
# https://stackoverflow.com/a/6440364/1570972
#
# socket(7) SO_LINGER option.
#
# struct linger {
# int l_onoff; /* linger active */
# int l_linger; /* how many seconds to linger for */
# };
#
# Is this correct for Windows/Cygwin and macOS?
struct_format = 'hh' if sys.platform == 'win32' else 'ii'
l_onoff = 1
l_linger = 0
client.sock.setsockopt(
socket.SOL_SOCKET,
socket.SO_LINGER,
struct.pack(struct_format, l_onoff, l_linger))
client.close()
# For integration with flufl.testing.
def setup(testobj):
testobj.globs['resources'] = ExitStack()
def teardown(testobj):
testobj.globs['resources'].close()
def make_debug_loop():
loop = asyncio.get_event_loop()
loop.set_debug(True)
return loop
def start(plugin):
if plugin.stderr:
# Turn on lots of debugging.
patch('aiosmtpd.smtp.make_loop', make_debug_loop).start()
logging.getLogger('asyncio').setLevel(logging.DEBUG)
logging.getLogger('mail.log').setLevel(logging.DEBUG)
warnings.filterwarnings('always', category=ResourceWarning)
def assert_auth_success(testcase: TestCase, code, response):
testcase.assertEqual(code, 235)
testcase.assertEqual(response, b"2.7.0 Authentication successful")
def assert_auth_invalid(testcase: TestCase, code, response):
testcase.assertEqual(code, 535)
testcase.assertEqual(response, b'5.7.8 Authentication credentials invalid')
def assert_auth_required(testcase: TestCase, code, response):
testcase.assertEqual(code, 530)
testcase.assertEqual(response, b'5.7.0 Authentication required')
SUPPORTED_COMMANDS_TLS: bytes = (
b'Supported commands: AUTH DATA EHLO HELO HELP MAIL '
b'NOOP QUIT RCPT RSET STARTTLS VRFY'
)
SUPPORTED_COMMANDS_NOTLS = SUPPORTED_COMMANDS_TLS.replace(b" STARTTLS", b"")
|
