Python code coverage for Lib/test/test_smtplib.py

#	count	content
1	n/a	import asyncore
2	n/a	import base64
3	n/a	import email.mime.text
4	n/a	from email.message import EmailMessage
5	n/a	from email.base64mime import body_encode as encode_base64
6	n/a	import email.utils
7	n/a	import hmac
8	n/a	import socket
9	n/a	import smtpd
10	n/a	import smtplib
11	n/a	import io
12	n/a	import re
13	n/a	import sys
14	n/a	import time
15	n/a	import select
16	n/a	import errno
17	n/a	import textwrap
18	n/a
19	n/a	import unittest
20	n/a	from test import support, mock_socket
21	n/a
22	n/a	try:
23	n/a	import threading
24	n/a	except ImportError:
25	n/a	threading = None
26	n/a
27	n/a	HOST = support.HOST
28	n/a
29	n/a	if sys.platform == 'darwin':
30	n/a	# select.poll returns a select.POLLHUP at the end of the tests
31	n/a	# on darwin, so just ignore it
32	n/a	def handle_expt(self):
33	n/a	pass
34	n/a	smtpd.SMTPChannel.handle_expt = handle_expt
35	n/a
36	n/a
37	n/a	def server(evt, buf, serv):
38	n/a	serv.listen()
39	n/a	evt.set()
40	n/a	try:
41	n/a	conn, addr = serv.accept()
42	n/a	except socket.timeout:
43	n/a	pass
44	n/a	else:
45	n/a	n = 500
46	n/a	while buf and n > 0:
47	n/a	r, w, e = select.select([], [conn], [])
48	n/a	if w:
49	n/a	sent = conn.send(buf)
50	n/a	buf = buf[sent:]
51	n/a
52	n/a	n -= 1
53	n/a
54	n/a	conn.close()
55	n/a	finally:
56	n/a	serv.close()
57	n/a	evt.set()
58	n/a
59	n/a	class GeneralTests(unittest.TestCase):
60	n/a
61	n/a	def setUp(self):
62	n/a	smtplib.socket = mock_socket
63	n/a	self.port = 25
64	n/a
65	n/a	def tearDown(self):
66	n/a	smtplib.socket = socket
67	n/a
68	n/a	# This method is no longer used but is retained for backward compatibility,
69	n/a	# so test to make sure it still works.
70	n/a	def testQuoteData(self):
71	n/a	teststr = "abc\n.jkl\rfoo\r\n..blue"
72	n/a	expected = "abc\r\n..jkl\r\nfoo\r\n...blue"
73	n/a	self.assertEqual(expected, smtplib.quotedata(teststr))
74	n/a
75	n/a	def testBasic1(self):
76	n/a	mock_socket.reply_with(b"220 Hola mundo")
77	n/a	# connects
78	n/a	smtp = smtplib.SMTP(HOST, self.port)
79	n/a	smtp.close()
80	n/a
81	n/a	def testSourceAddress(self):
82	n/a	mock_socket.reply_with(b"220 Hola mundo")
83	n/a	# connects
84	n/a	smtp = smtplib.SMTP(HOST, self.port,
85	n/a	source_address=('127.0.0.1',19876))
86	n/a	self.assertEqual(smtp.source_address, ('127.0.0.1', 19876))
87	n/a	smtp.close()
88	n/a
89	n/a	def testBasic2(self):
90	n/a	mock_socket.reply_with(b"220 Hola mundo")
91	n/a	# connects, include port in host name
92	n/a	smtp = smtplib.SMTP("%s:%s" % (HOST, self.port))
93	n/a	smtp.close()
94	n/a
95	n/a	def testLocalHostName(self):
96	n/a	mock_socket.reply_with(b"220 Hola mundo")
97	n/a	# check that supplied local_hostname is used
98	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname="testhost")
99	n/a	self.assertEqual(smtp.local_hostname, "testhost")
100	n/a	smtp.close()
101	n/a
102	n/a	def testTimeoutDefault(self):
103	n/a	mock_socket.reply_with(b"220 Hola mundo")
104	n/a	self.assertIsNone(mock_socket.getdefaulttimeout())
105	n/a	mock_socket.setdefaulttimeout(30)
106	n/a	self.assertEqual(mock_socket.getdefaulttimeout(), 30)
107	n/a	try:
108	n/a	smtp = smtplib.SMTP(HOST, self.port)
109	n/a	finally:
110	n/a	mock_socket.setdefaulttimeout(None)
111	n/a	self.assertEqual(smtp.sock.gettimeout(), 30)
112	n/a	smtp.close()
113	n/a
114	n/a	def testTimeoutNone(self):
115	n/a	mock_socket.reply_with(b"220 Hola mundo")
116	n/a	self.assertIsNone(socket.getdefaulttimeout())
117	n/a	socket.setdefaulttimeout(30)
118	n/a	try:
119	n/a	smtp = smtplib.SMTP(HOST, self.port, timeout=None)
120	n/a	finally:
121	n/a	socket.setdefaulttimeout(None)
122	n/a	self.assertIsNone(smtp.sock.gettimeout())
123	n/a	smtp.close()
124	n/a
125	n/a	def testTimeoutValue(self):
126	n/a	mock_socket.reply_with(b"220 Hola mundo")
127	n/a	smtp = smtplib.SMTP(HOST, self.port, timeout=30)
128	n/a	self.assertEqual(smtp.sock.gettimeout(), 30)
129	n/a	smtp.close()
130	n/a
131	n/a	def test_debuglevel(self):
132	n/a	mock_socket.reply_with(b"220 Hello world")
133	n/a	smtp = smtplib.SMTP()
134	n/a	smtp.set_debuglevel(1)
135	n/a	with support.captured_stderr() as stderr:
136	n/a	smtp.connect(HOST, self.port)
137	n/a	smtp.close()
138	n/a	expected = re.compile(r"^connect:", re.MULTILINE)
139	n/a	self.assertRegex(stderr.getvalue(), expected)
140	n/a
141	n/a	def test_debuglevel_2(self):
142	n/a	mock_socket.reply_with(b"220 Hello world")
143	n/a	smtp = smtplib.SMTP()
144	n/a	smtp.set_debuglevel(2)
145	n/a	with support.captured_stderr() as stderr:
146	n/a	smtp.connect(HOST, self.port)
147	n/a	smtp.close()
148	n/a	expected = re.compile(r"^\d{2}:\d{2}:\d{2}\.\d{6} connect: ",
149	n/a	re.MULTILINE)
150	n/a	self.assertRegex(stderr.getvalue(), expected)
151	n/a
152	n/a
153	n/a	# Test server thread using the specified SMTP server class
154	n/a	def debugging_server(serv, serv_evt, client_evt):
155	n/a	serv_evt.set()
156	n/a
157	n/a	try:
158	n/a	if hasattr(select, 'poll'):
159	n/a	poll_fun = asyncore.poll2
160	n/a	else:
161	n/a	poll_fun = asyncore.poll
162	n/a
163	n/a	n = 1000
164	n/a	while asyncore.socket_map and n > 0:
165	n/a	poll_fun(0.01, asyncore.socket_map)
166	n/a
167	n/a	# when the client conversation is finished, it will
168	n/a	# set client_evt, and it's then ok to kill the server
169	n/a	if client_evt.is_set():
170	n/a	serv.close()
171	n/a	break
172	n/a
173	n/a	n -= 1
174	n/a
175	n/a	except socket.timeout:
176	n/a	pass
177	n/a	finally:
178	n/a	if not client_evt.is_set():
179	n/a	# allow some time for the client to read the result
180	n/a	time.sleep(0.5)
181	n/a	serv.close()
182	n/a	asyncore.close_all()
183	n/a	serv_evt.set()
184	n/a
185	n/a	MSG_BEGIN = '---------- MESSAGE FOLLOWS ----------\n'
186	n/a	MSG_END = '------------ END MESSAGE ------------\n'
187	n/a
188	n/a	# NOTE: Some SMTP objects in the tests below are created with a non-default
189	n/a	# local_hostname argument to the constructor, since (on some systems) the FQDN
190	n/a	# lookup caused by the default local_hostname sometimes takes so long that the
191	n/a	# test server times out, causing the test to fail.
192	n/a
193	n/a	# Test behavior of smtpd.DebuggingServer
194	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
195	n/a	class DebuggingServerTests(unittest.TestCase):
196	n/a
197	n/a	maxDiff = None
198	n/a
199	n/a	def setUp(self):
200	n/a	self.real_getfqdn = socket.getfqdn
201	n/a	socket.getfqdn = mock_socket.getfqdn
202	n/a	# temporarily replace sys.stdout to capture DebuggingServer output
203	n/a	self.old_stdout = sys.stdout
204	n/a	self.output = io.StringIO()
205	n/a	sys.stdout = self.output
206	n/a
207	n/a	self.serv_evt = threading.Event()
208	n/a	self.client_evt = threading.Event()
209	n/a	# Capture SMTPChannel debug output
210	n/a	self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
211	n/a	smtpd.DEBUGSTREAM = io.StringIO()
212	n/a	# Pick a random unused port by passing 0 for the port number
213	n/a	self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1),
214	n/a	decode_data=True)
215	n/a	# Keep a note of what port was assigned
216	n/a	self.port = self.serv.socket.getsockname()[1]
217	n/a	serv_args = (self.serv, self.serv_evt, self.client_evt)
218	n/a	self.thread = threading.Thread(target=debugging_server, args=serv_args)
219	n/a	self.thread.start()
220	n/a
221	n/a	# wait until server thread has assigned a port number
222	n/a	self.serv_evt.wait()
223	n/a	self.serv_evt.clear()
224	n/a
225	n/a	def tearDown(self):
226	n/a	socket.getfqdn = self.real_getfqdn
227	n/a	# indicate that the client is finished
228	n/a	self.client_evt.set()
229	n/a	# wait for the server thread to terminate
230	n/a	self.serv_evt.wait()
231	n/a	self.thread.join()
232	n/a	# restore sys.stdout
233	n/a	sys.stdout = self.old_stdout
234	n/a	# restore DEBUGSTREAM
235	n/a	smtpd.DEBUGSTREAM.close()
236	n/a	smtpd.DEBUGSTREAM = self.old_DEBUGSTREAM
237	n/a
238	n/a	def testBasic(self):
239	n/a	# connect
240	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
241	n/a	smtp.quit()
242	n/a
243	n/a	def testSourceAddress(self):
244	n/a	# connect
245	n/a	port = support.find_unused_port()
246	n/a	try:
247	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
248	n/a	timeout=3, source_address=('127.0.0.1', port))
249	n/a	self.assertEqual(smtp.source_address, ('127.0.0.1', port))
250	n/a	self.assertEqual(smtp.local_hostname, 'localhost')
251	n/a	smtp.quit()
252	n/a	except OSError as e:
253	n/a	if e.errno == errno.EADDRINUSE:
254	n/a	self.skipTest("couldn't bind to port %d" % port)
255	n/a	raise
256	n/a
257	n/a	def testNOOP(self):
258	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
259	n/a	expected = (250, b'OK')
260	n/a	self.assertEqual(smtp.noop(), expected)
261	n/a	smtp.quit()
262	n/a
263	n/a	def testRSET(self):
264	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
265	n/a	expected = (250, b'OK')
266	n/a	self.assertEqual(smtp.rset(), expected)
267	n/a	smtp.quit()
268	n/a
269	n/a	def testELHO(self):
270	n/a	# EHLO isn't implemented in DebuggingServer
271	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
272	n/a	expected = (250, b'\nSIZE 33554432\nHELP')
273	n/a	self.assertEqual(smtp.ehlo(), expected)
274	n/a	smtp.quit()
275	n/a
276	n/a	def testEXPNNotImplemented(self):
277	n/a	# EXPN isn't implemented in DebuggingServer
278	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
279	n/a	expected = (502, b'EXPN not implemented')
280	n/a	smtp.putcmd('EXPN')
281	n/a	self.assertEqual(smtp.getreply(), expected)
282	n/a	smtp.quit()
283	n/a
284	n/a	def testVRFY(self):
285	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
286	n/a	expected = (252, b'Cannot VRFY user, but will accept message ' + \
287	n/a	b'and attempt delivery')
288	n/a	self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
289	n/a	self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
290	n/a	smtp.quit()
291	n/a
292	n/a	def testSecondHELO(self):
293	n/a	# check that a second HELO returns a message that it's a duplicate
294	n/a	# (this behavior is specific to smtpd.SMTPChannel)
295	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
296	n/a	smtp.helo()
297	n/a	expected = (503, b'Duplicate HELO/EHLO')
298	n/a	self.assertEqual(smtp.helo(), expected)
299	n/a	smtp.quit()
300	n/a
301	n/a	def testHELP(self):
302	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
303	n/a	self.assertEqual(smtp.help(), b'Supported commands: EHLO HELO MAIL ' + \
304	n/a	b'RCPT DATA RSET NOOP QUIT VRFY')
305	n/a	smtp.quit()
306	n/a
307	n/a	def testSend(self):
308	n/a	# connect and send mail
309	n/a	m = 'A test message'
310	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
311	n/a	smtp.sendmail('John', 'Sally', m)
312	n/a	# XXX(nnorwitz): this test is flaky and dies with a bad file descriptor
313	n/a	# in asyncore. This sleep might help, but should really be fixed
314	n/a	# properly by using an Event variable.
315	n/a	time.sleep(0.01)
316	n/a	smtp.quit()
317	n/a
318	n/a	self.client_evt.set()
319	n/a	self.serv_evt.wait()
320	n/a	self.output.flush()
321	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
322	n/a	self.assertEqual(self.output.getvalue(), mexpect)
323	n/a
324	n/a	def testSendBinary(self):
325	n/a	m = b'A test message'
326	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
327	n/a	smtp.sendmail('John', 'Sally', m)
328	n/a	# XXX (see comment in testSend)
329	n/a	time.sleep(0.01)
330	n/a	smtp.quit()
331	n/a
332	n/a	self.client_evt.set()
333	n/a	self.serv_evt.wait()
334	n/a	self.output.flush()
335	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.decode('ascii'), MSG_END)
336	n/a	self.assertEqual(self.output.getvalue(), mexpect)
337	n/a
338	n/a	def testSendNeedingDotQuote(self):
339	n/a	# Issue 12283
340	n/a	m = '.A test\n.mes.sage.'
341	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
342	n/a	smtp.sendmail('John', 'Sally', m)
343	n/a	# XXX (see comment in testSend)
344	n/a	time.sleep(0.01)
345	n/a	smtp.quit()
346	n/a
347	n/a	self.client_evt.set()
348	n/a	self.serv_evt.wait()
349	n/a	self.output.flush()
350	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
351	n/a	self.assertEqual(self.output.getvalue(), mexpect)
352	n/a
353	n/a	def testSendNullSender(self):
354	n/a	m = 'A test message'
355	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
356	n/a	smtp.sendmail('<>', 'Sally', m)
357	n/a	# XXX (see comment in testSend)
358	n/a	time.sleep(0.01)
359	n/a	smtp.quit()
360	n/a
361	n/a	self.client_evt.set()
362	n/a	self.serv_evt.wait()
363	n/a	self.output.flush()
364	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
365	n/a	self.assertEqual(self.output.getvalue(), mexpect)
366	n/a	debugout = smtpd.DEBUGSTREAM.getvalue()
367	n/a	sender = re.compile("^sender: <>$", re.MULTILINE)
368	n/a	self.assertRegex(debugout, sender)
369	n/a
370	n/a	def testSendMessage(self):
371	n/a	m = email.mime.text.MIMEText('A test message')
372	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
373	n/a	smtp.send_message(m, from_addr='John', to_addrs='Sally')
374	n/a	# XXX (see comment in testSend)
375	n/a	time.sleep(0.01)
376	n/a	smtp.quit()
377	n/a
378	n/a	self.client_evt.set()
379	n/a	self.serv_evt.wait()
380	n/a	self.output.flush()
381	n/a	# Add the X-Peer header that DebuggingServer adds
382	n/a	m['X-Peer'] = socket.gethostbyname('localhost')
383	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
384	n/a	self.assertEqual(self.output.getvalue(), mexpect)
385	n/a
386	n/a	def testSendMessageWithAddresses(self):
387	n/a	m = email.mime.text.MIMEText('A test message')
388	n/a	m['From'] = 'foo@bar.com'
389	n/a	m['To'] = 'John'
390	n/a	m['CC'] = 'Sally, Fred'
391	n/a	m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
392	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
393	n/a	smtp.send_message(m)
394	n/a	# XXX (see comment in testSend)
395	n/a	time.sleep(0.01)
396	n/a	smtp.quit()
397	n/a	# make sure the Bcc header is still in the message.
398	n/a	self.assertEqual(m['Bcc'], 'John Root <root@localhost>, "Dinsdale" '
399	n/a	'<warped@silly.walks.com>')
400	n/a
401	n/a	self.client_evt.set()
402	n/a	self.serv_evt.wait()
403	n/a	self.output.flush()
404	n/a	# Add the X-Peer header that DebuggingServer adds
405	n/a	m['X-Peer'] = socket.gethostbyname('localhost')
406	n/a	# The Bcc header should not be transmitted.
407	n/a	del m['Bcc']
408	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
409	n/a	self.assertEqual(self.output.getvalue(), mexpect)
410	n/a	debugout = smtpd.DEBUGSTREAM.getvalue()
411	n/a	sender = re.compile("^sender: foo@bar.com$", re.MULTILINE)
412	n/a	self.assertRegex(debugout, sender)
413	n/a	for addr in ('John', 'Sally', 'Fred', 'root@localhost',
414	n/a	'warped@silly.walks.com'):
415	n/a	to_addr = re.compile(r"^recips: .'{}'.$".format(addr),
416	n/a	re.MULTILINE)
417	n/a	self.assertRegex(debugout, to_addr)
418	n/a
419	n/a	def testSendMessageWithSomeAddresses(self):
420	n/a	# Make sure nothing breaks if not all of the three 'to' headers exist
421	n/a	m = email.mime.text.MIMEText('A test message')
422	n/a	m['From'] = 'foo@bar.com'
423	n/a	m['To'] = 'John, Dinsdale'
424	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
425	n/a	smtp.send_message(m)
426	n/a	# XXX (see comment in testSend)
427	n/a	time.sleep(0.01)
428	n/a	smtp.quit()
429	n/a
430	n/a	self.client_evt.set()
431	n/a	self.serv_evt.wait()
432	n/a	self.output.flush()
433	n/a	# Add the X-Peer header that DebuggingServer adds
434	n/a	m['X-Peer'] = socket.gethostbyname('localhost')
435	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
436	n/a	self.assertEqual(self.output.getvalue(), mexpect)
437	n/a	debugout = smtpd.DEBUGSTREAM.getvalue()
438	n/a	sender = re.compile("^sender: foo@bar.com$", re.MULTILINE)
439	n/a	self.assertRegex(debugout, sender)
440	n/a	for addr in ('John', 'Dinsdale'):
441	n/a	to_addr = re.compile(r"^recips: .'{}'.$".format(addr),
442	n/a	re.MULTILINE)
443	n/a	self.assertRegex(debugout, to_addr)
444	n/a
445	n/a	def testSendMessageWithSpecifiedAddresses(self):
446	n/a	# Make sure addresses specified in call override those in message.
447	n/a	m = email.mime.text.MIMEText('A test message')
448	n/a	m['From'] = 'foo@bar.com'
449	n/a	m['To'] = 'John, Dinsdale'
450	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
451	n/a	smtp.send_message(m, from_addr='joe@example.com', to_addrs='foo@example.net')
452	n/a	# XXX (see comment in testSend)
453	n/a	time.sleep(0.01)
454	n/a	smtp.quit()
455	n/a
456	n/a	self.client_evt.set()
457	n/a	self.serv_evt.wait()
458	n/a	self.output.flush()
459	n/a	# Add the X-Peer header that DebuggingServer adds
460	n/a	m['X-Peer'] = socket.gethostbyname('localhost')
461	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
462	n/a	self.assertEqual(self.output.getvalue(), mexpect)
463	n/a	debugout = smtpd.DEBUGSTREAM.getvalue()
464	n/a	sender = re.compile("^sender: joe@example.com$", re.MULTILINE)
465	n/a	self.assertRegex(debugout, sender)
466	n/a	for addr in ('John', 'Dinsdale'):
467	n/a	to_addr = re.compile(r"^recips: .'{}'.$".format(addr),
468	n/a	re.MULTILINE)
469	n/a	self.assertNotRegex(debugout, to_addr)
470	n/a	recip = re.compile(r"^recips: .'foo@example.net'.$", re.MULTILINE)
471	n/a	self.assertRegex(debugout, recip)
472	n/a
473	n/a	def testSendMessageWithMultipleFrom(self):
474	n/a	# Sender overrides To
475	n/a	m = email.mime.text.MIMEText('A test message')
476	n/a	m['From'] = 'Bernard, Bianca'
477	n/a	m['Sender'] = 'the_rescuers@Rescue-Aid-Society.com'
478	n/a	m['To'] = 'John, Dinsdale'
479	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
480	n/a	smtp.send_message(m)
481	n/a	# XXX (see comment in testSend)
482	n/a	time.sleep(0.01)
483	n/a	smtp.quit()
484	n/a
485	n/a	self.client_evt.set()
486	n/a	self.serv_evt.wait()
487	n/a	self.output.flush()
488	n/a	# Add the X-Peer header that DebuggingServer adds
489	n/a	m['X-Peer'] = socket.gethostbyname('localhost')
490	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
491	n/a	self.assertEqual(self.output.getvalue(), mexpect)
492	n/a	debugout = smtpd.DEBUGSTREAM.getvalue()
493	n/a	sender = re.compile("^sender: the_rescuers@Rescue-Aid-Society.com$", re.MULTILINE)
494	n/a	self.assertRegex(debugout, sender)
495	n/a	for addr in ('John', 'Dinsdale'):
496	n/a	to_addr = re.compile(r"^recips: .'{}'.$".format(addr),
497	n/a	re.MULTILINE)
498	n/a	self.assertRegex(debugout, to_addr)
499	n/a
500	n/a	def testSendMessageResent(self):
501	n/a	m = email.mime.text.MIMEText('A test message')
502	n/a	m['From'] = 'foo@bar.com'
503	n/a	m['To'] = 'John'
504	n/a	m['CC'] = 'Sally, Fred'
505	n/a	m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
506	n/a	m['Resent-Date'] = 'Thu, 1 Jan 1970 17:42:00 +0000'
507	n/a	m['Resent-From'] = 'holy@grail.net'
508	n/a	m['Resent-To'] = 'Martha <my_mom@great.cooker.com>, Jeff'
509	n/a	m['Resent-Bcc'] = 'doe@losthope.net'
510	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
511	n/a	smtp.send_message(m)
512	n/a	# XXX (see comment in testSend)
513	n/a	time.sleep(0.01)
514	n/a	smtp.quit()
515	n/a
516	n/a	self.client_evt.set()
517	n/a	self.serv_evt.wait()
518	n/a	self.output.flush()
519	n/a	# The Resent-Bcc headers are deleted before serialization.
520	n/a	del m['Bcc']
521	n/a	del m['Resent-Bcc']
522	n/a	# Add the X-Peer header that DebuggingServer adds
523	n/a	m['X-Peer'] = socket.gethostbyname('localhost')
524	n/a	mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
525	n/a	self.assertEqual(self.output.getvalue(), mexpect)
526	n/a	debugout = smtpd.DEBUGSTREAM.getvalue()
527	n/a	sender = re.compile("^sender: holy@grail.net$", re.MULTILINE)
528	n/a	self.assertRegex(debugout, sender)
529	n/a	for addr in ('my_mom@great.cooker.com', 'Jeff', 'doe@losthope.net'):
530	n/a	to_addr = re.compile(r"^recips: .'{}'.$".format(addr),
531	n/a	re.MULTILINE)
532	n/a	self.assertRegex(debugout, to_addr)
533	n/a
534	n/a	def testSendMessageMultipleResentRaises(self):
535	n/a	m = email.mime.text.MIMEText('A test message')
536	n/a	m['From'] = 'foo@bar.com'
537	n/a	m['To'] = 'John'
538	n/a	m['CC'] = 'Sally, Fred'
539	n/a	m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
540	n/a	m['Resent-Date'] = 'Thu, 1 Jan 1970 17:42:00 +0000'
541	n/a	m['Resent-From'] = 'holy@grail.net'
542	n/a	m['Resent-To'] = 'Martha <my_mom@great.cooker.com>, Jeff'
543	n/a	m['Resent-Bcc'] = 'doe@losthope.net'
544	n/a	m['Resent-Date'] = 'Thu, 2 Jan 1970 17:42:00 +0000'
545	n/a	m['Resent-To'] = 'holy@grail.net'
546	n/a	m['Resent-From'] = 'Martha <my_mom@great.cooker.com>, Jeff'
547	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
548	n/a	with self.assertRaises(ValueError):
549	n/a	smtp.send_message(m)
550	n/a	smtp.close()
551	n/a
552	n/a	class NonConnectingTests(unittest.TestCase):
553	n/a
554	n/a	def testNotConnected(self):
555	n/a	# Test various operations on an unconnected SMTP object that
556	n/a	# should raise exceptions (at present the attempt in SMTP.send
557	n/a	# to reference the nonexistent 'sock' attribute of the SMTP object
558	n/a	# causes an AttributeError)
559	n/a	smtp = smtplib.SMTP()
560	n/a	self.assertRaises(smtplib.SMTPServerDisconnected, smtp.ehlo)
561	n/a	self.assertRaises(smtplib.SMTPServerDisconnected,
562	n/a	smtp.send, 'test msg')
563	n/a
564	n/a	def testNonnumericPort(self):
565	n/a	# check that non-numeric port raises OSError
566	n/a	self.assertRaises(OSError, smtplib.SMTP,
567	n/a	"localhost", "bogus")
568	n/a	self.assertRaises(OSError, smtplib.SMTP,
569	n/a	"localhost:bogus")
570	n/a
571	n/a
572	n/a	# test response of client to a non-successful HELO message
573	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
574	n/a	class BadHELOServerTests(unittest.TestCase):
575	n/a
576	n/a	def setUp(self):
577	n/a	smtplib.socket = mock_socket
578	n/a	mock_socket.reply_with(b"199 no hello for you!")
579	n/a	self.old_stdout = sys.stdout
580	n/a	self.output = io.StringIO()
581	n/a	sys.stdout = self.output
582	n/a	self.port = 25
583	n/a
584	n/a	def tearDown(self):
585	n/a	smtplib.socket = socket
586	n/a	sys.stdout = self.old_stdout
587	n/a
588	n/a	def testFailingHELO(self):
589	n/a	self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP,
590	n/a	HOST, self.port, 'localhost', 3)
591	n/a
592	n/a
593	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
594	n/a	class TooLongLineTests(unittest.TestCase):
595	n/a	respdata = b'250 OK' + (b'.' * smtplib._MAXLINE * 2) + b'\n'
596	n/a
597	n/a	def setUp(self):
598	n/a	self.old_stdout = sys.stdout
599	n/a	self.output = io.StringIO()
600	n/a	sys.stdout = self.output
601	n/a
602	n/a	self.evt = threading.Event()
603	n/a	self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
604	n/a	self.sock.settimeout(15)
605	n/a	self.port = support.bind_port(self.sock)
606	n/a	servargs = (self.evt, self.respdata, self.sock)
607	n/a	threading.Thread(target=server, args=servargs).start()
608	n/a	self.evt.wait()
609	n/a	self.evt.clear()
610	n/a
611	n/a	def tearDown(self):
612	n/a	self.evt.wait()
613	n/a	sys.stdout = self.old_stdout
614	n/a
615	n/a	def testLineTooLong(self):
616	n/a	self.assertRaises(smtplib.SMTPResponseException, smtplib.SMTP,
617	n/a	HOST, self.port, 'localhost', 3)
618	n/a
619	n/a
620	n/a	sim_users = {'Mr.A@somewhere.com':'John A',
621	n/a	'Ms.B@xn--fo-fka.com':'Sally B',
622	n/a	'Mrs.C@somewhereesle.com':'Ruth C',
623	n/a	}
624	n/a
625	n/a	sim_auth = ('Mr.A@somewhere.com', 'somepassword')
626	n/a	sim_cram_md5_challenge = ('PENCeUxFREJoU0NnbmhNWitOMjNGNn'
627	n/a	'dAZWx3b29kLmlubm9zb2Z0LmNvbT4=')
628	n/a	sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'],
629	n/a	'list-2':['Ms.B@xn--fo-fka.com',],
630	n/a	}
631	n/a
632	n/a	# Simulated SMTP channel & server
633	n/a	class ResponseException(Exception): pass
634	n/a	class SimSMTPChannel(smtpd.SMTPChannel):
635	n/a
636	n/a	quit_response = None
637	n/a	mail_response = None
638	n/a	rcpt_response = None
639	n/a	data_response = None
640	n/a	rcpt_count = 0
641	n/a	rset_count = 0
642	n/a	disconnect = 0
643	n/a	AUTH = 99 # Add protocol state to enable auth testing.
644	n/a	authenticated_user = None
645	n/a
646	n/a	def __init__(self, extra_features, args, *kw):
647	n/a	self._extrafeatures = ''.join(
648	n/a	[ "250-{0}\r\n".format(x) for x in extra_features ])
649	n/a	super(SimSMTPChannel, self).__init__(args, *kw)
650	n/a
651	n/a	# AUTH related stuff. It would be nice if support for this were in smtpd.
652	n/a	def found_terminator(self):
653	n/a	if self.smtp_state == self.AUTH:
654	n/a	line = self._emptystring.join(self.received_lines)
655	n/a	print('Data:', repr(line), file=smtpd.DEBUGSTREAM)
656	n/a	self.received_lines = []
657	n/a	try:
658	n/a	self.auth_object(line)
659	n/a	except ResponseException as e:
660	n/a	self.smtp_state = self.COMMAND
661	n/a	self.push('%s %s' % (e.smtp_code, e.smtp_error))
662	n/a	return
663	n/a	super().found_terminator()
664	n/a
665	n/a
666	n/a	def smtp_AUTH(self, arg):
667	n/a	if not self.seen_greeting:
668	n/a	self.push('503 Error: send EHLO first')
669	n/a	return
670	n/a	if not self.extended_smtp or 'AUTH' not in self._extrafeatures:
671	n/a	self.push('500 Error: command "AUTH" not recognized')
672	n/a	return
673	n/a	if self.authenticated_user is not None:
674	n/a	self.push(
675	n/a	'503 Bad sequence of commands: already authenticated')
676	n/a	return
677	n/a	args = arg.split()
678	n/a	if len(args) not in [1, 2]:
679	n/a	self.push('501 Syntax: AUTH <mechanism> [initial-response]')
680	n/a	return
681	n/a	auth_object_name = '_auth_%s' % args[0].lower().replace('-', '_')
682	n/a	try:
683	n/a	self.auth_object = getattr(self, auth_object_name)
684	n/a	except AttributeError:
685	n/a	self.push('504 Command parameter not implemented: unsupported '
686	n/a	' authentication mechanism {!r}'.format(auth_object_name))
687	n/a	return
688	n/a	self.smtp_state = self.AUTH
689	n/a	self.auth_object(args[1] if len(args) == 2 else None)
690	n/a
691	n/a	def _authenticated(self, user, valid):
692	n/a	if valid:
693	n/a	self.authenticated_user = user
694	n/a	self.push('235 Authentication Succeeded')
695	n/a	else:
696	n/a	self.push('535 Authentication credentials invalid')
697	n/a	self.smtp_state = self.COMMAND
698	n/a
699	n/a	def _decode_base64(self, string):
700	n/a	return base64.decodebytes(string.encode('ascii')).decode('utf-8')
701	n/a
702	n/a	def _auth_plain(self, arg=None):
703	n/a	if arg is None:
704	n/a	self.push('334 ')
705	n/a	else:
706	n/a	logpass = self._decode_base64(arg)
707	n/a	try:
708	n/a	*_, user, password = logpass.split('\0')
709	n/a	except ValueError as e:
710	n/a	self.push('535 Splitting response {!r} into user and password'
711	n/a	' failed: {}'.format(logpass, e))
712	n/a	return
713	n/a	self._authenticated(user, password == sim_auth[1])
714	n/a
715	n/a	def _auth_login(self, arg=None):
716	n/a	if arg is None:
717	n/a	# base64 encoded 'Username:'
718	n/a	self.push('334 VXNlcm5hbWU6')
719	n/a	elif not hasattr(self, '_auth_login_user'):
720	n/a	self._auth_login_user = self._decode_base64(arg)
721	n/a	# base64 encoded 'Password:'
722	n/a	self.push('334 UGFzc3dvcmQ6')
723	n/a	else:
724	n/a	password = self._decode_base64(arg)
725	n/a	self._authenticated(self._auth_login_user, password == sim_auth[1])
726	n/a	del self._auth_login_user
727	n/a
728	n/a	def _auth_cram_md5(self, arg=None):
729	n/a	if arg is None:
730	n/a	self.push('334 {}'.format(sim_cram_md5_challenge))
731	n/a	else:
732	n/a	logpass = self._decode_base64(arg)
733	n/a	try:
734	n/a	user, hashed_pass = logpass.split()
735	n/a	except ValueError as e:
736	n/a	self.push('535 Splitting response {!r} into user and password'
737	n/a	'failed: {}'.format(logpass, e))
738	n/a	return False
739	n/a	valid_hashed_pass = hmac.HMAC(
740	n/a	sim_auth[1].encode('ascii'),
741	n/a	self._decode_base64(sim_cram_md5_challenge).encode('ascii'),
742	n/a	'md5').hexdigest()
743	n/a	self._authenticated(user, hashed_pass == valid_hashed_pass)
744	n/a	# end AUTH related stuff.
745	n/a
746	n/a	def smtp_EHLO(self, arg):
747	n/a	resp = ('250-testhost\r\n'
748	n/a	'250-EXPN\r\n'
749	n/a	'250-SIZE 20000000\r\n'
750	n/a	'250-STARTTLS\r\n'
751	n/a	'250-DELIVERBY\r\n')
752	n/a	resp = resp + self._extrafeatures + '250 HELP'
753	n/a	self.push(resp)
754	n/a	self.seen_greeting = arg
755	n/a	self.extended_smtp = True
756	n/a
757	n/a	def smtp_VRFY(self, arg):
758	n/a	# For max compatibility smtplib should be sending the raw address.
759	n/a	if arg in sim_users:
760	n/a	self.push('250 %s %s' % (sim_users[arg], smtplib.quoteaddr(arg)))
761	n/a	else:
762	n/a	self.push('550 No such user: %s' % arg)
763	n/a
764	n/a	def smtp_EXPN(self, arg):
765	n/a	list_name = arg.lower()
766	n/a	if list_name in sim_lists:
767	n/a	user_list = sim_lists[list_name]
768	n/a	for n, user_email in enumerate(user_list):
769	n/a	quoted_addr = smtplib.quoteaddr(user_email)
770	n/a	if n < len(user_list) - 1:
771	n/a	self.push('250-%s %s' % (sim_users[user_email], quoted_addr))
772	n/a	else:
773	n/a	self.push('250 %s %s' % (sim_users[user_email], quoted_addr))
774	n/a	else:
775	n/a	self.push('550 No access for you!')
776	n/a
777	n/a	def smtp_QUIT(self, arg):
778	n/a	if self.quit_response is None:
779	n/a	super(SimSMTPChannel, self).smtp_QUIT(arg)
780	n/a	else:
781	n/a	self.push(self.quit_response)
782	n/a	self.close_when_done()
783	n/a
784	n/a	def smtp_MAIL(self, arg):
785	n/a	if self.mail_response is None:
786	n/a	super().smtp_MAIL(arg)
787	n/a	else:
788	n/a	self.push(self.mail_response)
789	n/a	if self.disconnect:
790	n/a	self.close_when_done()
791	n/a
792	n/a	def smtp_RCPT(self, arg):
793	n/a	if self.rcpt_response is None:
794	n/a	super().smtp_RCPT(arg)
795	n/a	return
796	n/a	self.rcpt_count += 1
797	n/a	self.push(self.rcpt_response[self.rcpt_count-1])
798	n/a
799	n/a	def smtp_RSET(self, arg):
800	n/a	self.rset_count += 1
801	n/a	super().smtp_RSET(arg)
802	n/a
803	n/a	def smtp_DATA(self, arg):
804	n/a	if self.data_response is None:
805	n/a	super().smtp_DATA(arg)
806	n/a	else:
807	n/a	self.push(self.data_response)
808	n/a
809	n/a	def handle_error(self):
810	n/a	raise
811	n/a
812	n/a
813	n/a	class SimSMTPServer(smtpd.SMTPServer):
814	n/a
815	n/a	channel_class = SimSMTPChannel
816	n/a
817	n/a	def __init__(self, args, *kw):
818	n/a	self._extra_features = []
819	n/a	smtpd.SMTPServer.__init__(self, args, *kw)
820	n/a
821	n/a	def handle_accepted(self, conn, addr):
822	n/a	self._SMTPchannel = self.channel_class(
823	n/a	self._extra_features, self, conn, addr,
824	n/a	decode_data=self._decode_data)
825	n/a
826	n/a	def process_message(self, peer, mailfrom, rcpttos, data):
827	n/a	pass
828	n/a
829	n/a	def add_feature(self, feature):
830	n/a	self._extra_features.append(feature)
831	n/a
832	n/a	def handle_error(self):
833	n/a	raise
834	n/a
835	n/a
836	n/a	# Test various SMTP & ESMTP commands/behaviors that require a simulated server
837	n/a	# (i.e., something with more features than DebuggingServer)
838	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
839	n/a	class SMTPSimTests(unittest.TestCase):
840	n/a
841	n/a	def setUp(self):
842	n/a	self.real_getfqdn = socket.getfqdn
843	n/a	socket.getfqdn = mock_socket.getfqdn
844	n/a	self.serv_evt = threading.Event()
845	n/a	self.client_evt = threading.Event()
846	n/a	# Pick a random unused port by passing 0 for the port number
847	n/a	self.serv = SimSMTPServer((HOST, 0), ('nowhere', -1), decode_data=True)
848	n/a	# Keep a note of what port was assigned
849	n/a	self.port = self.serv.socket.getsockname()[1]
850	n/a	serv_args = (self.serv, self.serv_evt, self.client_evt)
851	n/a	self.thread = threading.Thread(target=debugging_server, args=serv_args)
852	n/a	self.thread.start()
853	n/a
854	n/a	# wait until server thread has assigned a port number
855	n/a	self.serv_evt.wait()
856	n/a	self.serv_evt.clear()
857	n/a
858	n/a	def tearDown(self):
859	n/a	socket.getfqdn = self.real_getfqdn
860	n/a	# indicate that the client is finished
861	n/a	self.client_evt.set()
862	n/a	# wait for the server thread to terminate
863	n/a	self.serv_evt.wait()
864	n/a	self.thread.join()
865	n/a
866	n/a	def testBasic(self):
867	n/a	# smoke test
868	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
869	n/a	smtp.quit()
870	n/a
871	n/a	def testEHLO(self):
872	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
873	n/a
874	n/a	# no features should be present before the EHLO
875	n/a	self.assertEqual(smtp.esmtp_features, {})
876	n/a
877	n/a	# features expected from the test server
878	n/a	expected_features = {'expn':'',
879	n/a	'size': '20000000',
880	n/a	'starttls': '',
881	n/a	'deliverby': '',
882	n/a	'help': '',
883	n/a	}
884	n/a
885	n/a	smtp.ehlo()
886	n/a	self.assertEqual(smtp.esmtp_features, expected_features)
887	n/a	for k in expected_features:
888	n/a	self.assertTrue(smtp.has_extn(k))
889	n/a	self.assertFalse(smtp.has_extn('unsupported-feature'))
890	n/a	smtp.quit()
891	n/a
892	n/a	def testVRFY(self):
893	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
894	n/a
895	n/a	for addr_spec, name in sim_users.items():
896	n/a	expected_known = (250, bytes('%s %s' %
897	n/a	(name, smtplib.quoteaddr(addr_spec)),
898	n/a	"ascii"))
899	n/a	self.assertEqual(smtp.vrfy(addr_spec), expected_known)
900	n/a
901	n/a	u = 'nobody@nowhere.com'
902	n/a	expected_unknown = (550, ('No such user: %s' % u).encode('ascii'))
903	n/a	self.assertEqual(smtp.vrfy(u), expected_unknown)
904	n/a	smtp.quit()
905	n/a
906	n/a	def testEXPN(self):
907	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
908	n/a
909	n/a	for listname, members in sim_lists.items():
910	n/a	users = []
911	n/a	for m in members:
912	n/a	users.append('%s %s' % (sim_users[m], smtplib.quoteaddr(m)))
913	n/a	expected_known = (250, bytes('\n'.join(users), "ascii"))
914	n/a	self.assertEqual(smtp.expn(listname), expected_known)
915	n/a
916	n/a	u = 'PSU-Members-List'
917	n/a	expected_unknown = (550, b'No access for you!')
918	n/a	self.assertEqual(smtp.expn(u), expected_unknown)
919	n/a	smtp.quit()
920	n/a
921	n/a	def testAUTH_PLAIN(self):
922	n/a	self.serv.add_feature("AUTH PLAIN")
923	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
924	n/a	resp = smtp.login(sim_auth[0], sim_auth[1])
925	n/a	self.assertEqual(resp, (235, b'Authentication Succeeded'))
926	n/a	smtp.close()
927	n/a
928	n/a	def testAUTH_LOGIN(self):
929	n/a	self.serv.add_feature("AUTH LOGIN")
930	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
931	n/a	resp = smtp.login(sim_auth[0], sim_auth[1])
932	n/a	self.assertEqual(resp, (235, b'Authentication Succeeded'))
933	n/a	smtp.close()
934	n/a
935	n/a	def testAUTH_CRAM_MD5(self):
936	n/a	self.serv.add_feature("AUTH CRAM-MD5")
937	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
938	n/a	resp = smtp.login(sim_auth[0], sim_auth[1])
939	n/a	self.assertEqual(resp, (235, b'Authentication Succeeded'))
940	n/a	smtp.close()
941	n/a
942	n/a	def testAUTH_multiple(self):
943	n/a	# Test that multiple authentication methods are tried.
944	n/a	self.serv.add_feature("AUTH BOGUS PLAIN LOGIN CRAM-MD5")
945	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
946	n/a	resp = smtp.login(sim_auth[0], sim_auth[1])
947	n/a	self.assertEqual(resp, (235, b'Authentication Succeeded'))
948	n/a	smtp.close()
949	n/a
950	n/a	def test_auth_function(self):
951	n/a	supported = {'CRAM-MD5', 'PLAIN', 'LOGIN'}
952	n/a	for mechanism in supported:
953	n/a	self.serv.add_feature("AUTH {}".format(mechanism))
954	n/a	for mechanism in supported:
955	n/a	with self.subTest(mechanism=mechanism):
956	n/a	smtp = smtplib.SMTP(HOST, self.port,
957	n/a	local_hostname='localhost', timeout=15)
958	n/a	smtp.ehlo('foo')
959	n/a	smtp.user, smtp.password = sim_auth[0], sim_auth[1]
960	n/a	method = 'auth_' + mechanism.lower().replace('-', '_')
961	n/a	resp = smtp.auth(mechanism, getattr(smtp, method))
962	n/a	self.assertEqual(resp, (235, b'Authentication Succeeded'))
963	n/a	smtp.close()
964	n/a
965	n/a	def test_quit_resets_greeting(self):
966	n/a	smtp = smtplib.SMTP(HOST, self.port,
967	n/a	local_hostname='localhost',
968	n/a	timeout=15)
969	n/a	code, message = smtp.ehlo()
970	n/a	self.assertEqual(code, 250)
971	n/a	self.assertIn('size', smtp.esmtp_features)
972	n/a	smtp.quit()
973	n/a	self.assertNotIn('size', smtp.esmtp_features)
974	n/a	smtp.connect(HOST, self.port)
975	n/a	self.assertNotIn('size', smtp.esmtp_features)
976	n/a	smtp.ehlo_or_helo_if_needed()
977	n/a	self.assertIn('size', smtp.esmtp_features)
978	n/a	smtp.quit()
979	n/a
980	n/a	def test_with_statement(self):
981	n/a	with smtplib.SMTP(HOST, self.port) as smtp:
982	n/a	code, message = smtp.noop()
983	n/a	self.assertEqual(code, 250)
984	n/a	self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo')
985	n/a	with smtplib.SMTP(HOST, self.port) as smtp:
986	n/a	smtp.close()
987	n/a	self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo')
988	n/a
989	n/a	def test_with_statement_QUIT_failure(self):
990	n/a	with self.assertRaises(smtplib.SMTPResponseException) as error:
991	n/a	with smtplib.SMTP(HOST, self.port) as smtp:
992	n/a	smtp.noop()
993	n/a	self.serv._SMTPchannel.quit_response = '421 QUIT FAILED'
994	n/a	self.assertEqual(error.exception.smtp_code, 421)
995	n/a	self.assertEqual(error.exception.smtp_error, b'QUIT FAILED')
996	n/a
997	n/a	#TODO: add tests for correct AUTH method fallback now that the
998	n/a	#test infrastructure can support it.
999	n/a
1000	n/a	# Issue 17498: make sure _rset does not raise SMTPServerDisconnected exception
1001	n/a	def test__rest_from_mail_cmd(self):
1002	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
1003	n/a	smtp.noop()
1004	n/a	self.serv._SMTPchannel.mail_response = '451 Requested action aborted'
1005	n/a	self.serv._SMTPchannel.disconnect = True
1006	n/a	with self.assertRaises(smtplib.SMTPSenderRefused):
1007	n/a	smtp.sendmail('John', 'Sally', 'test message')
1008	n/a	self.assertIsNone(smtp.sock)
1009	n/a
1010	n/a	# Issue 5713: make sure close, not rset, is called if we get a 421 error
1011	n/a	def test_421_from_mail_cmd(self):
1012	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
1013	n/a	smtp.noop()
1014	n/a	self.serv._SMTPchannel.mail_response = '421 closing connection'
1015	n/a	with self.assertRaises(smtplib.SMTPSenderRefused):
1016	n/a	smtp.sendmail('John', 'Sally', 'test message')
1017	n/a	self.assertIsNone(smtp.sock)
1018	n/a	self.assertEqual(self.serv._SMTPchannel.rset_count, 0)
1019	n/a
1020	n/a	def test_421_from_rcpt_cmd(self):
1021	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
1022	n/a	smtp.noop()
1023	n/a	self.serv._SMTPchannel.rcpt_response = ['250 accepted', '421 closing']
1024	n/a	with self.assertRaises(smtplib.SMTPRecipientsRefused) as r:
1025	n/a	smtp.sendmail('John', ['Sally', 'Frank', 'George'], 'test message')
1026	n/a	self.assertIsNone(smtp.sock)
1027	n/a	self.assertEqual(self.serv._SMTPchannel.rset_count, 0)
1028	n/a	self.assertDictEqual(r.exception.args[0], {'Frank': (421, b'closing')})
1029	n/a
1030	n/a	def test_421_from_data_cmd(self):
1031	n/a	class MySimSMTPChannel(SimSMTPChannel):
1032	n/a	def found_terminator(self):
1033	n/a	if self.smtp_state == self.DATA:
1034	n/a	self.push('421 closing')
1035	n/a	else:
1036	n/a	super().found_terminator()
1037	n/a	self.serv.channel_class = MySimSMTPChannel
1038	n/a	smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
1039	n/a	smtp.noop()
1040	n/a	with self.assertRaises(smtplib.SMTPDataError):
1041	n/a	smtp.sendmail('John@foo.org', ['Sally@foo.org'], 'test message')
1042	n/a	self.assertIsNone(smtp.sock)
1043	n/a	self.assertEqual(self.serv._SMTPchannel.rcpt_count, 0)
1044	n/a
1045	n/a	def test_smtputf8_NotSupportedError_if_no_server_support(self):
1046	n/a	smtp = smtplib.SMTP(
1047	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1048	n/a	self.addCleanup(smtp.close)
1049	n/a	smtp.ehlo()
1050	n/a	self.assertTrue(smtp.does_esmtp)
1051	n/a	self.assertFalse(smtp.has_extn('smtputf8'))
1052	n/a	self.assertRaises(
1053	n/a	smtplib.SMTPNotSupportedError,
1054	n/a	smtp.sendmail,
1055	n/a	'John', 'Sally', '', mail_options=['BODY=8BITMIME', 'SMTPUTF8'])
1056	n/a	self.assertRaises(
1057	n/a	smtplib.SMTPNotSupportedError,
1058	n/a	smtp.mail, 'John', options=['BODY=8BITMIME', 'SMTPUTF8'])
1059	n/a
1060	n/a	def test_send_unicode_without_SMTPUTF8(self):
1061	n/a	smtp = smtplib.SMTP(
1062	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1063	n/a	self.addCleanup(smtp.close)
1064	n/a	self.assertRaises(UnicodeEncodeError, smtp.sendmail, 'Alice', 'BÃ¶b', '')
1065	n/a	self.assertRaises(UnicodeEncodeError, smtp.mail, 'Ã„lice')
1066	n/a
1067	n/a
1068	n/a	class SimSMTPUTF8Server(SimSMTPServer):
1069	n/a
1070	n/a	def __init__(self, args, *kw):
1071	n/a	# The base SMTP server turns these on automatically, but our test
1072	n/a	# server is set up to munge the EHLO response, so we need to provide
1073	n/a	# them as well. And yes, the call is to SMTPServer not SimSMTPServer.
1074	n/a	self._extra_features = ['SMTPUTF8', '8BITMIME']
1075	n/a	smtpd.SMTPServer.__init__(self, args, *kw)
1076	n/a
1077	n/a	def handle_accepted(self, conn, addr):
1078	n/a	self._SMTPchannel = self.channel_class(
1079	n/a	self._extra_features, self, conn, addr,
1080	n/a	decode_data=self._decode_data,
1081	n/a	enable_SMTPUTF8=self.enable_SMTPUTF8,
1082	n/a	)
1083	n/a
1084	n/a	def process_message(self, peer, mailfrom, rcpttos, data, mail_options=None,
1085	n/a	rcpt_options=None):
1086	n/a	self.last_peer = peer
1087	n/a	self.last_mailfrom = mailfrom
1088	n/a	self.last_rcpttos = rcpttos
1089	n/a	self.last_message = data
1090	n/a	self.last_mail_options = mail_options
1091	n/a	self.last_rcpt_options = rcpt_options
1092	n/a
1093	n/a
1094	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
1095	n/a	class SMTPUTF8SimTests(unittest.TestCase):
1096	n/a
1097	n/a	maxDiff = None
1098	n/a
1099	n/a	def setUp(self):
1100	n/a	self.real_getfqdn = socket.getfqdn
1101	n/a	socket.getfqdn = mock_socket.getfqdn
1102	n/a	self.serv_evt = threading.Event()
1103	n/a	self.client_evt = threading.Event()
1104	n/a	# Pick a random unused port by passing 0 for the port number
1105	n/a	self.serv = SimSMTPUTF8Server((HOST, 0), ('nowhere', -1),
1106	n/a	decode_data=False,
1107	n/a	enable_SMTPUTF8=True)
1108	n/a	# Keep a note of what port was assigned
1109	n/a	self.port = self.serv.socket.getsockname()[1]
1110	n/a	serv_args = (self.serv, self.serv_evt, self.client_evt)
1111	n/a	self.thread = threading.Thread(target=debugging_server, args=serv_args)
1112	n/a	self.thread.start()
1113	n/a
1114	n/a	# wait until server thread has assigned a port number
1115	n/a	self.serv_evt.wait()
1116	n/a	self.serv_evt.clear()
1117	n/a
1118	n/a	def tearDown(self):
1119	n/a	socket.getfqdn = self.real_getfqdn
1120	n/a	# indicate that the client is finished
1121	n/a	self.client_evt.set()
1122	n/a	# wait for the server thread to terminate
1123	n/a	self.serv_evt.wait()
1124	n/a	self.thread.join()
1125	n/a
1126	n/a	def test_test_server_supports_extensions(self):
1127	n/a	smtp = smtplib.SMTP(
1128	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1129	n/a	self.addCleanup(smtp.close)
1130	n/a	smtp.ehlo()
1131	n/a	self.assertTrue(smtp.does_esmtp)
1132	n/a	self.assertTrue(smtp.has_extn('smtputf8'))
1133	n/a
1134	n/a	def test_send_unicode_with_SMTPUTF8_via_sendmail(self):
1135	n/a	m = 'Â¡a test message containing unicode!'.encode('utf-8')
1136	n/a	smtp = smtplib.SMTP(
1137	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1138	n/a	self.addCleanup(smtp.close)
1139	n/a	smtp.sendmail('JÅ‘hn', 'SÃ¡lly', m,
1140	n/a	mail_options=['BODY=8BITMIME', 'SMTPUTF8'])
1141	n/a	self.assertEqual(self.serv.last_mailfrom, 'JÅ‘hn')
1142	n/a	self.assertEqual(self.serv.last_rcpttos, ['SÃ¡lly'])
1143	n/a	self.assertEqual(self.serv.last_message, m)
1144	n/a	self.assertIn('BODY=8BITMIME', self.serv.last_mail_options)
1145	n/a	self.assertIn('SMTPUTF8', self.serv.last_mail_options)
1146	n/a	self.assertEqual(self.serv.last_rcpt_options, [])
1147	n/a
1148	n/a	def test_send_unicode_with_SMTPUTF8_via_low_level_API(self):
1149	n/a	m = 'Â¡a test message containing unicode!'.encode('utf-8')
1150	n/a	smtp = smtplib.SMTP(
1151	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1152	n/a	self.addCleanup(smtp.close)
1153	n/a	smtp.ehlo()
1154	n/a	self.assertEqual(
1155	n/a	smtp.mail('JÅ‘', options=['BODY=8BITMIME', 'SMTPUTF8']),
1156	n/a	(250, b'OK'))
1157	n/a	self.assertEqual(smtp.rcpt('JÃ¡nos'), (250, b'OK'))
1158	n/a	self.assertEqual(smtp.data(m), (250, b'OK'))
1159	n/a	self.assertEqual(self.serv.last_mailfrom, 'JÅ‘')
1160	n/a	self.assertEqual(self.serv.last_rcpttos, ['JÃ¡nos'])
1161	n/a	self.assertEqual(self.serv.last_message, m)
1162	n/a	self.assertIn('BODY=8BITMIME', self.serv.last_mail_options)
1163	n/a	self.assertIn('SMTPUTF8', self.serv.last_mail_options)
1164	n/a	self.assertEqual(self.serv.last_rcpt_options, [])
1165	n/a
1166	n/a	def test_send_message_uses_smtputf8_if_addrs_non_ascii(self):
1167	n/a	msg = EmailMessage()
1168	n/a	msg['From'] = "PÃ¡olo <fÅ‘o@bar.com>"
1169	n/a	msg['To'] = 'Dinsdale'
1170	n/a	msg['Subject'] = 'Nudge nudge, wink, wink \u1F609'
1171	n/a	# XXX I don't know why I need two \n's here, but this is an existing
1172	n/a	# bug (if it is one) and not a problem with the new functionality.
1173	n/a	msg.set_content("oh lÃ lÃ , know what I mean, know what I mean?\n\n")
1174	n/a	# XXX smtpd converts received /r/n to /n, so we can't easily test that
1175	n/a	# we are successfully sending /r/n :(.
1176	n/a	expected = textwrap.dedent("""\
1177	n/a	From: PÃ¡olo <fÅ‘o@bar.com>
1178	n/a	To: Dinsdale
1179	n/a	Subject: Nudge nudge, wink, wink \u1F609
1180	n/a	Content-Type: text/plain; charset="utf-8"
1181	n/a	Content-Transfer-Encoding: 8bit
1182	n/a	MIME-Version: 1.0
1183	n/a
1184	n/a	oh lÃ lÃ , know what I mean, know what I mean?
1185	n/a	""")
1186	n/a	smtp = smtplib.SMTP(
1187	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1188	n/a	self.addCleanup(smtp.close)
1189	n/a	self.assertEqual(smtp.send_message(msg), {})
1190	n/a	self.assertEqual(self.serv.last_mailfrom, 'fÅ‘o@bar.com')
1191	n/a	self.assertEqual(self.serv.last_rcpttos, ['Dinsdale'])
1192	n/a	self.assertEqual(self.serv.last_message.decode(), expected)
1193	n/a	self.assertIn('BODY=8BITMIME', self.serv.last_mail_options)
1194	n/a	self.assertIn('SMTPUTF8', self.serv.last_mail_options)
1195	n/a	self.assertEqual(self.serv.last_rcpt_options, [])
1196	n/a
1197	n/a	def test_send_message_error_on_non_ascii_addrs_if_no_smtputf8(self):
1198	n/a	msg = EmailMessage()
1199	n/a	msg['From'] = "PÃ¡olo <fÅ‘o@bar.com>"
1200	n/a	msg['To'] = 'Dinsdale'
1201	n/a	msg['Subject'] = 'Nudge nudge, wink, wink \u1F609'
1202	n/a	smtp = smtplib.SMTP(
1203	n/a	HOST, self.port, local_hostname='localhost', timeout=3)
1204	n/a	self.addCleanup(smtp.close)
1205	n/a	self.assertRaises(smtplib.SMTPNotSupportedError,
1206	n/a	smtp.send_message(msg))
1207	n/a
1208	n/a
1209	n/a	EXPECTED_RESPONSE = encode_base64(b'\0psu\0doesnotexist', eol='')
1210	n/a
1211	n/a	class SimSMTPAUTHInitialResponseChannel(SimSMTPChannel):
1212	n/a	def smtp_AUTH(self, arg):
1213	n/a	# RFC 4954's AUTH command allows for an optional initial-response.
1214	n/a	# Not all AUTH methods support this; some require a challenge. AUTH
1215	n/a	# PLAIN does those, so test that here. See issue #15014.
1216	n/a	args = arg.split()
1217	n/a	if args[0].lower() == 'plain':
1218	n/a	if len(args) == 2:
1219	n/a	# AUTH PLAIN <initial-response> with the response base 64
1220	n/a	# encoded. Hard code the expected response for the test.
1221	n/a	if args[1] == EXPECTED_RESPONSE:
1222	n/a	self.push('235 Ok')
1223	n/a	return
1224	n/a	self.push('571 Bad authentication')
1225	n/a
1226	n/a	class SimSMTPAUTHInitialResponseServer(SimSMTPServer):
1227	n/a	channel_class = SimSMTPAUTHInitialResponseChannel
1228	n/a
1229	n/a
1230	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
1231	n/a	class SMTPAUTHInitialResponseSimTests(unittest.TestCase):
1232	n/a	def setUp(self):
1233	n/a	self.real_getfqdn = socket.getfqdn
1234	n/a	socket.getfqdn = mock_socket.getfqdn
1235	n/a	self.serv_evt = threading.Event()
1236	n/a	self.client_evt = threading.Event()
1237	n/a	# Pick a random unused port by passing 0 for the port number
1238	n/a	self.serv = SimSMTPAUTHInitialResponseServer(
1239	n/a	(HOST, 0), ('nowhere', -1), decode_data=True)
1240	n/a	# Keep a note of what port was assigned
1241	n/a	self.port = self.serv.socket.getsockname()[1]
1242	n/a	serv_args = (self.serv, self.serv_evt, self.client_evt)
1243	n/a	self.thread = threading.Thread(target=debugging_server, args=serv_args)
1244	n/a	self.thread.start()
1245	n/a
1246	n/a	# wait until server thread has assigned a port number
1247	n/a	self.serv_evt.wait()
1248	n/a	self.serv_evt.clear()
1249	n/a
1250	n/a	def tearDown(self):
1251	n/a	socket.getfqdn = self.real_getfqdn
1252	n/a	# indicate that the client is finished
1253	n/a	self.client_evt.set()
1254	n/a	# wait for the server thread to terminate
1255	n/a	self.serv_evt.wait()
1256	n/a	self.thread.join()
1257	n/a
1258	n/a	def testAUTH_PLAIN_initial_response_login(self):
1259	n/a	self.serv.add_feature('AUTH PLAIN')
1260	n/a	smtp = smtplib.SMTP(HOST, self.port,
1261	n/a	local_hostname='localhost', timeout=15)
1262	n/a	smtp.login('psu', 'doesnotexist')
1263	n/a	smtp.close()
1264	n/a
1265	n/a	def testAUTH_PLAIN_initial_response_auth(self):
1266	n/a	self.serv.add_feature('AUTH PLAIN')
1267	n/a	smtp = smtplib.SMTP(HOST, self.port,
1268	n/a	local_hostname='localhost', timeout=15)
1269	n/a	smtp.user = 'psu'
1270	n/a	smtp.password = 'doesnotexist'
1271	n/a	code, response = smtp.auth('plain', smtp.auth_plain)
1272	n/a	smtp.close()
1273	n/a	self.assertEqual(code, 235)
1274	n/a
1275	n/a
1276	n/a	@support.reap_threads
1277	n/a	def test_main(verbose=None):
1278	n/a	support.run_unittest(
1279	n/a	BadHELOServerTests,
1280	n/a	DebuggingServerTests,
1281	n/a	GeneralTests,
1282	n/a	NonConnectingTests,
1283	n/a	SMTPAUTHInitialResponseSimTests,
1284	n/a	SMTPSimTests,
1285	n/a	TooLongLineTests,
1286	n/a	)
1287	n/a
1288	n/a
1289	n/a	if __name__ == '__main__':
1290	n/a	test_main()