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