1"""Testing helpers.""" 2 3import sys 4import socket 5import struct 6import asyncio 7import logging 8import warnings 9 10from contextlib import ExitStack 11from unittest.mock import patch 12 13 14def reset_connection(client): 15 # Close the connection with a TCP RST instead of a TCP FIN. client must 16 # be a smtplib.SMTP instance. 17 # 18 # https://stackoverflow.com/a/6440364/1570972 19 # 20 # socket(7) SO_LINGER option. 21 # 22 # struct linger { 23 # int l_onoff; /* linger active */ 24 # int l_linger; /* how many seconds to linger for */ 25 # }; 26 # 27 # Is this correct for Windows/Cygwin and macOS? 28 struct_format = 'hh' if sys.platform == 'win32' else 'ii' 29 l_onoff = 1 30 l_linger = 0 31 client.sock.setsockopt( 32 socket.SOL_SOCKET, 33 socket.SO_LINGER, 34 struct.pack(struct_format, l_onoff, l_linger)) 35 client.close() 36 37 38# For integration with flufl.testing. 39 40def setup(testobj): 41 testobj.globs['resources'] = ExitStack() 42 43 44def teardown(testobj): 45 testobj.globs['resources'].close() 46 47 48def make_debug_loop(): 49 loop = asyncio.get_event_loop() 50 loop.set_debug(True) 51 return loop 52 53 54def start(plugin): 55 if plugin.stderr: 56 # Turn on lots of debugging. 57 patch('aiosmtpd.smtp.make_loop', make_debug_loop).start() 58 logging.getLogger('asyncio').setLevel(logging.DEBUG) 59 logging.getLogger('mail.log').setLevel(logging.DEBUG) 60 warnings.filterwarnings('always', category=ResourceWarning) 61