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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
"""
Tests that cover asyncio usage.
"""
import asyncio
import ssl
from collections.abc import Awaitable
from typing import Any
import pytest
from aiosmtplib import SMTP
from aiosmtplib.response import SMTPResponse
from .smtpd import mock_response_expn
RECIPIENTS = [
"recipient1@example.com",
"recipient2@example.com",
"recipient3@example.com",
]
async def test_sendmail_multiple_times_in_sequence(
smtp_client: SMTP,
smtpd_server: asyncio.AbstractServer,
sender_str: str,
message_str: str,
) -> None:
async with smtp_client:
for recipient in RECIPIENTS:
errors, response = await smtp_client.sendmail(
sender_str, [recipient], message_str
)
assert not errors
assert isinstance(errors, dict)
assert response != ""
async def test_sendmail_multiple_times_with_gather(
smtp_client: SMTP,
smtpd_server: asyncio.AbstractServer,
sender_str: str,
message_str: str,
) -> None:
async with smtp_client:
tasks = [
smtp_client.sendmail(sender_str, [recipient], message_str)
for recipient in RECIPIENTS
]
results = await asyncio.gather(*tasks)
for errors, message in results:
assert not errors
assert isinstance(errors, dict)
assert message != ""
async def test_connect_and_sendmail_multiple_times_with_gather(
hostname: str,
smtpd_server_port: int,
client_tls_context: ssl.SSLContext,
sender_str: str,
message_str: str,
) -> None:
async def connect_and_send(
*args: Any, **kwargs: Any
) -> tuple[dict[str, SMTPResponse], str]:
async with SMTP(
hostname=hostname, port=smtpd_server_port, tls_context=client_tls_context
) as client:
response = await client.sendmail(*args, **kwargs)
return response
tasks = [
connect_and_send(sender_str, [recipient], message_str)
for recipient in RECIPIENTS
]
results = await asyncio.gather(*tasks)
for errors, message in results:
assert not errors
assert isinstance(errors, dict)
assert message != ""
async def test_multiple_clients_with_gather(
hostname: str,
smtpd_server: asyncio.AbstractServer,
smtpd_server_port: int,
client_tls_context: ssl.SSLContext,
sender_str: str,
message_str: str,
) -> None:
async def connect_and_send(
*args: Any, **kwargs: Any
) -> tuple[dict[str, SMTPResponse], str]:
client = SMTP(
hostname=hostname, port=smtpd_server_port, tls_context=client_tls_context
)
async with client:
response = await client.sendmail(*args, **kwargs)
return response
tasks = [
connect_and_send(sender_str, [recipient], message_str)
for recipient in RECIPIENTS
]
results = await asyncio.gather(*tasks)
for errors, message in results:
assert not errors
assert isinstance(errors, dict)
assert message != ""
async def test_multiple_actions_in_context_manager_with_gather(
hostname: str,
smtpd_server_port: int,
client_tls_context: ssl.SSLContext,
sender_str: str,
message_str: str,
) -> None:
async def connect_and_run_commands(*args: Any, **kwargs: Any) -> SMTPResponse:
async with SMTP(
hostname=hostname, port=smtpd_server_port, tls_context=client_tls_context
) as client:
await client.ehlo()
await client.help()
response = await client.noop()
return response
tasks = [
connect_and_run_commands(sender_str, [recipient], message_str)
for recipient in RECIPIENTS
]
responses = await asyncio.gather(*tasks)
for response in responses:
assert 200 <= response.code < 300
@pytest.mark.smtpd_mocks(smtp_EXPN=mock_response_expn)
async def test_many_commands_with_gather(smtp_client: SMTP) -> None:
"""
Tests that appropriate locks are in place to prevent commands confusing each other.
"""
async with smtp_client:
tasks: list[Awaitable] = [
smtp_client.ehlo(),
smtp_client.helo(),
smtp_client.noop(),
smtp_client.vrfy("foo@bar.com"),
smtp_client.expn("users@example.com"),
smtp_client.mail("alice@example.com"),
smtp_client.help(),
]
results = await asyncio.gather(*tasks)
for result in results[:-1]:
assert 200 <= result.code < 300
# Help text is returned as a string, not a result tuple
assert "Supported commands" in results[-1]
async def test_close_works_on_stopped_loop(
hostname: str,
smtpd_server_port: int,
client_tls_context: ssl.SSLContext,
) -> None:
event_loop = asyncio.get_running_loop()
client = SMTP(
hostname=hostname, port=smtpd_server_port, tls_context=client_tls_context
)
await client.connect()
assert client.is_connected
assert client.transport is not None
event_loop.stop()
client.close()
assert not client.is_connected
async def test_context_manager_entry_multiple_times_with_gather(
smtp_client: SMTP, sender_str: str, message_str: str
) -> None:
async def connect_and_send(
*args: Any, **kwargs: Any
) -> tuple[dict[str, SMTPResponse], str]:
async with smtp_client:
response = await smtp_client.sendmail(*args, **kwargs)
return response
tasks = [
asyncio.wait_for(connect_and_send(sender_str, [recipient], message_str), 2.0)
for recipient in RECIPIENTS
]
results = await asyncio.gather(*tasks)
for errors, message in results:
assert not errors
assert isinstance(errors, dict)
assert message != ""
|
