Python code coverage for Lib/test/test_smtpd.py

#	count	content
1	n/a	import unittest
2	n/a	import textwrap
3	n/a	from test import support, mock_socket
4	n/a	import socket
5	n/a	import io
6	n/a	import smtpd
7	n/a	import asyncore
8	n/a
9	n/a
10	n/a	class DummyServer(smtpd.SMTPServer):
11	n/a	def __init__(self, args, *kwargs):
12	n/a	smtpd.SMTPServer.__init__(self, args, *kwargs)
13	n/a	self.messages = []
14	n/a	if self._decode_data:
15	n/a	self.return_status = 'return status'
16	n/a	else:
17	n/a	self.return_status = b'return status'
18	n/a
19	n/a	def process_message(self, peer, mailfrom, rcpttos, data, **kw):
20	n/a	self.messages.append((peer, mailfrom, rcpttos, data))
21	n/a	if data == self.return_status:
22	n/a	return '250 Okish'
23	n/a	if 'mail_options' in kw and 'SMTPUTF8' in kw['mail_options']:
24	n/a	return '250 SMTPUTF8 message okish'
25	n/a
26	n/a
27	n/a	class DummyDispatcherBroken(Exception):
28	n/a	pass
29	n/a
30	n/a
31	n/a	class BrokenDummyServer(DummyServer):
32	n/a	def listen(self, num):
33	n/a	raise DummyDispatcherBroken()
34	n/a
35	n/a
36	n/a	class SMTPDServerTest(unittest.TestCase):
37	n/a	def setUp(self):
38	n/a	smtpd.socket = asyncore.socket = mock_socket
39	n/a
40	n/a	def test_process_message_unimplemented(self):
41	n/a	server = smtpd.SMTPServer((support.HOST, 0), ('b', 0),
42	n/a	decode_data=True)
43	n/a	conn, addr = server.accept()
44	n/a	channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
45	n/a
46	n/a	def write_line(line):
47	n/a	channel.socket.queue_recv(line)
48	n/a	channel.handle_read()
49	n/a
50	n/a	write_line(b'HELO example')
51	n/a	write_line(b'MAIL From:eggs@example')
52	n/a	write_line(b'RCPT To:spam@example')
53	n/a	write_line(b'DATA')
54	n/a	self.assertRaises(NotImplementedError, write_line, b'spam\r\n.\r\n')
55	n/a
56	n/a	def test_decode_data_and_enable_SMTPUTF8_raises(self):
57	n/a	self.assertRaises(
58	n/a	ValueError,
59	n/a	smtpd.SMTPServer,
60	n/a	(support.HOST, 0),
61	n/a	('b', 0),
62	n/a	enable_SMTPUTF8=True,
63	n/a	decode_data=True)
64	n/a
65	n/a	def tearDown(self):
66	n/a	asyncore.close_all()
67	n/a	asyncore.socket = smtpd.socket = socket
68	n/a
69	n/a
70	n/a	class DebuggingServerTest(unittest.TestCase):
71	n/a
72	n/a	def setUp(self):
73	n/a	smtpd.socket = asyncore.socket = mock_socket
74	n/a
75	n/a	def send_data(self, channel, data, enable_SMTPUTF8=False):
76	n/a	def write_line(line):
77	n/a	channel.socket.queue_recv(line)
78	n/a	channel.handle_read()
79	n/a	write_line(b'EHLO example')
80	n/a	if enable_SMTPUTF8:
81	n/a	write_line(b'MAIL From:eggs@example BODY=8BITMIME SMTPUTF8')
82	n/a	else:
83	n/a	write_line(b'MAIL From:eggs@example')
84	n/a	write_line(b'RCPT To:spam@example')
85	n/a	write_line(b'DATA')
86	n/a	write_line(data)
87	n/a	write_line(b'.')
88	n/a
89	n/a	def test_process_message_with_decode_data_true(self):
90	n/a	server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
91	n/a	decode_data=True)
92	n/a	conn, addr = server.accept()
93	n/a	channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
94	n/a	with support.captured_stdout() as s:
95	n/a	self.send_data(channel, b'From: test\n\nhello\n')
96	n/a	stdout = s.getvalue()
97	n/a	self.assertEqual(stdout, textwrap.dedent("""\
98	n/a	---------- MESSAGE FOLLOWS ----------
99	n/a	From: test
100	n/a	X-Peer: peer-address
101	n/a
102	n/a	hello
103	n/a	------------ END MESSAGE ------------
104	n/a	"""))
105	n/a
106	n/a	def test_process_message_with_decode_data_false(self):
107	n/a	server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0))
108	n/a	conn, addr = server.accept()
109	n/a	channel = smtpd.SMTPChannel(server, conn, addr)
110	n/a	with support.captured_stdout() as s:
111	n/a	self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
112	n/a	stdout = s.getvalue()
113	n/a	self.assertEqual(stdout, textwrap.dedent("""\
114	n/a	---------- MESSAGE FOLLOWS ----------
115	n/a	b'From: test'
116	n/a	b'X-Peer: peer-address'
117	n/a	b''
118	n/a	b'h\\xc3\\xa9llo\\xff'
119	n/a	------------ END MESSAGE ------------
120	n/a	"""))
121	n/a
122	n/a	def test_process_message_with_enable_SMTPUTF8_true(self):
123	n/a	server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
124	n/a	enable_SMTPUTF8=True)
125	n/a	conn, addr = server.accept()
126	n/a	channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
127	n/a	with support.captured_stdout() as s:
128	n/a	self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
129	n/a	stdout = s.getvalue()
130	n/a	self.assertEqual(stdout, textwrap.dedent("""\
131	n/a	---------- MESSAGE FOLLOWS ----------
132	n/a	b'From: test'
133	n/a	b'X-Peer: peer-address'
134	n/a	b''
135	n/a	b'h\\xc3\\xa9llo\\xff'
136	n/a	------------ END MESSAGE ------------
137	n/a	"""))
138	n/a
139	n/a	def test_process_SMTPUTF8_message_with_enable_SMTPUTF8_true(self):
140	n/a	server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
141	n/a	enable_SMTPUTF8=True)
142	n/a	conn, addr = server.accept()
143	n/a	channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
144	n/a	with support.captured_stdout() as s:
145	n/a	self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n',
146	n/a	enable_SMTPUTF8=True)
147	n/a	stdout = s.getvalue()
148	n/a	self.assertEqual(stdout, textwrap.dedent("""\
149	n/a	---------- MESSAGE FOLLOWS ----------
150	n/a	mail options: ['BODY=8BITMIME', 'SMTPUTF8']
151	n/a	b'From: test'
152	n/a	b'X-Peer: peer-address'
153	n/a	b''
154	n/a	b'h\\xc3\\xa9llo\\xff'
155	n/a	------------ END MESSAGE ------------
156	n/a	"""))
157	n/a
158	n/a	def tearDown(self):
159	n/a	asyncore.close_all()
160	n/a	asyncore.socket = smtpd.socket = socket
161	n/a
162	n/a
163	n/a	class TestFamilyDetection(unittest.TestCase):
164	n/a	def setUp(self):
165	n/a	smtpd.socket = asyncore.socket = mock_socket
166	n/a
167	n/a	def tearDown(self):
168	n/a	asyncore.close_all()
169	n/a	asyncore.socket = smtpd.socket = socket
170	n/a
171	n/a	@unittest.skipUnless(support.IPV6_ENABLED, "IPv6 not enabled")
172	n/a	def test_socket_uses_IPv6(self):
173	n/a	server = smtpd.SMTPServer((support.HOSTv6, 0), (support.HOST, 0))
174	n/a	self.assertEqual(server.socket.family, socket.AF_INET6)
175	n/a
176	n/a	def test_socket_uses_IPv4(self):
177	n/a	server = smtpd.SMTPServer((support.HOST, 0), (support.HOSTv6, 0))
178	n/a	self.assertEqual(server.socket.family, socket.AF_INET)
179	n/a
180	n/a
181	n/a	class TestRcptOptionParsing(unittest.TestCase):
182	n/a	error_response = (b'555 RCPT TO parameters not recognized or not '
183	n/a	b'implemented\r\n')
184	n/a
185	n/a	def setUp(self):
186	n/a	smtpd.socket = asyncore.socket = mock_socket
187	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
188	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
189	n/a
190	n/a	def tearDown(self):
191	n/a	asyncore.close_all()
192	n/a	asyncore.socket = smtpd.socket = socket
193	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
194	n/a
195	n/a	def write_line(self, channel, line):
196	n/a	channel.socket.queue_recv(line)
197	n/a	channel.handle_read()
198	n/a
199	n/a	def test_params_rejected(self):
200	n/a	server = DummyServer((support.HOST, 0), ('b', 0))
201	n/a	conn, addr = server.accept()
202	n/a	channel = smtpd.SMTPChannel(server, conn, addr)
203	n/a	self.write_line(channel, b'EHLO example')
204	n/a	self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
205	n/a	self.write_line(channel, b'RCPT to: <foo@example.com> foo=bar')
206	n/a	self.assertEqual(channel.socket.last, self.error_response)
207	n/a
208	n/a	def test_nothing_accepted(self):
209	n/a	server = DummyServer((support.HOST, 0), ('b', 0))
210	n/a	conn, addr = server.accept()
211	n/a	channel = smtpd.SMTPChannel(server, conn, addr)
212	n/a	self.write_line(channel, b'EHLO example')
213	n/a	self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
214	n/a	self.write_line(channel, b'RCPT to: <foo@example.com>')
215	n/a	self.assertEqual(channel.socket.last, b'250 OK\r\n')
216	n/a
217	n/a
218	n/a	class TestMailOptionParsing(unittest.TestCase):
219	n/a	error_response = (b'555 MAIL FROM parameters not recognized or not '
220	n/a	b'implemented\r\n')
221	n/a
222	n/a	def setUp(self):
223	n/a	smtpd.socket = asyncore.socket = mock_socket
224	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
225	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
226	n/a
227	n/a	def tearDown(self):
228	n/a	asyncore.close_all()
229	n/a	asyncore.socket = smtpd.socket = socket
230	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
231	n/a
232	n/a	def write_line(self, channel, line):
233	n/a	channel.socket.queue_recv(line)
234	n/a	channel.handle_read()
235	n/a
236	n/a	def test_with_decode_data_true(self):
237	n/a	server = DummyServer((support.HOST, 0), ('b', 0), decode_data=True)
238	n/a	conn, addr = server.accept()
239	n/a	channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
240	n/a	self.write_line(channel, b'EHLO example')
241	n/a	for line in [
242	n/a	b'MAIL from: <foo@example.com> size=20 SMTPUTF8',
243	n/a	b'MAIL from: <foo@example.com> size=20 SMTPUTF8 BODY=8BITMIME',
244	n/a	b'MAIL from: <foo@example.com> size=20 BODY=UNKNOWN',
245	n/a	b'MAIL from: <foo@example.com> size=20 body=8bitmime',
246	n/a	]:
247	n/a	self.write_line(channel, line)
248	n/a	self.assertEqual(channel.socket.last, self.error_response)
249	n/a	self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
250	n/a	self.assertEqual(channel.socket.last, b'250 OK\r\n')
251	n/a
252	n/a	def test_with_decode_data_false(self):
253	n/a	server = DummyServer((support.HOST, 0), ('b', 0))
254	n/a	conn, addr = server.accept()
255	n/a	channel = smtpd.SMTPChannel(server, conn, addr)
256	n/a	self.write_line(channel, b'EHLO example')
257	n/a	for line in [
258	n/a	b'MAIL from: <foo@example.com> size=20 SMTPUTF8',
259	n/a	b'MAIL from: <foo@example.com> size=20 SMTPUTF8 BODY=8BITMIME',
260	n/a	]:
261	n/a	self.write_line(channel, line)
262	n/a	self.assertEqual(channel.socket.last, self.error_response)
263	n/a	self.write_line(
264	n/a	channel,
265	n/a	b'MAIL from: <foo@example.com> size=20 SMTPUTF8 BODY=UNKNOWN')
266	n/a	self.assertEqual(
267	n/a	channel.socket.last,
268	n/a	b'501 Error: BODY can only be one of 7BIT, 8BITMIME\r\n')
269	n/a	self.write_line(
270	n/a	channel, b'MAIL from: <foo@example.com> size=20 body=8bitmime')
271	n/a	self.assertEqual(channel.socket.last, b'250 OK\r\n')
272	n/a
273	n/a	def test_with_enable_smtputf8_true(self):
274	n/a	server = DummyServer((support.HOST, 0), ('b', 0), enable_SMTPUTF8=True)
275	n/a	conn, addr = server.accept()
276	n/a	channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
277	n/a	self.write_line(channel, b'EHLO example')
278	n/a	self.write_line(
279	n/a	channel,
280	n/a	b'MAIL from: <foo@example.com> size=20 body=8bitmime smtputf8')
281	n/a	self.assertEqual(channel.socket.last, b'250 OK\r\n')
282	n/a
283	n/a
284	n/a	class SMTPDChannelTest(unittest.TestCase):
285	n/a	def setUp(self):
286	n/a	smtpd.socket = asyncore.socket = mock_socket
287	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
288	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
289	n/a	self.server = DummyServer((support.HOST, 0), ('b', 0),
290	n/a	decode_data=True)
291	n/a	conn, addr = self.server.accept()
292	n/a	self.channel = smtpd.SMTPChannel(self.server, conn, addr,
293	n/a	decode_data=True)
294	n/a
295	n/a	def tearDown(self):
296	n/a	asyncore.close_all()
297	n/a	asyncore.socket = smtpd.socket = socket
298	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
299	n/a
300	n/a	def write_line(self, line):
301	n/a	self.channel.socket.queue_recv(line)
302	n/a	self.channel.handle_read()
303	n/a
304	n/a	def test_broken_connect(self):
305	n/a	self.assertRaises(
306	n/a	DummyDispatcherBroken, BrokenDummyServer,
307	n/a	(support.HOST, 0), ('b', 0), decode_data=True)
308	n/a
309	n/a	def test_decode_data_and_enable_SMTPUTF8_raises(self):
310	n/a	self.assertRaises(
311	n/a	ValueError, smtpd.SMTPChannel,
312	n/a	self.server, self.channel.conn, self.channel.addr,
313	n/a	enable_SMTPUTF8=True, decode_data=True)
314	n/a
315	n/a	def test_server_accept(self):
316	n/a	self.server.handle_accept()
317	n/a
318	n/a	def test_missing_data(self):
319	n/a	self.write_line(b'')
320	n/a	self.assertEqual(self.channel.socket.last,
321	n/a	b'500 Error: bad syntax\r\n')
322	n/a
323	n/a	def test_EHLO(self):
324	n/a	self.write_line(b'EHLO example')
325	n/a	self.assertEqual(self.channel.socket.last, b'250 HELP\r\n')
326	n/a
327	n/a	def test_EHLO_bad_syntax(self):
328	n/a	self.write_line(b'EHLO')
329	n/a	self.assertEqual(self.channel.socket.last,
330	n/a	b'501 Syntax: EHLO hostname\r\n')
331	n/a
332	n/a	def test_EHLO_duplicate(self):
333	n/a	self.write_line(b'EHLO example')
334	n/a	self.write_line(b'EHLO example')
335	n/a	self.assertEqual(self.channel.socket.last,
336	n/a	b'503 Duplicate HELO/EHLO\r\n')
337	n/a
338	n/a	def test_EHLO_HELO_duplicate(self):
339	n/a	self.write_line(b'EHLO example')
340	n/a	self.write_line(b'HELO example')
341	n/a	self.assertEqual(self.channel.socket.last,
342	n/a	b'503 Duplicate HELO/EHLO\r\n')
343	n/a
344	n/a	def test_HELO(self):
345	n/a	name = smtpd.socket.getfqdn()
346	n/a	self.write_line(b'HELO example')
347	n/a	self.assertEqual(self.channel.socket.last,
348	n/a	'250 {}\r\n'.format(name).encode('ascii'))
349	n/a
350	n/a	def test_HELO_EHLO_duplicate(self):
351	n/a	self.write_line(b'HELO example')
352	n/a	self.write_line(b'EHLO example')
353	n/a	self.assertEqual(self.channel.socket.last,
354	n/a	b'503 Duplicate HELO/EHLO\r\n')
355	n/a
356	n/a	def test_HELP(self):
357	n/a	self.write_line(b'HELP')
358	n/a	self.assertEqual(self.channel.socket.last,
359	n/a	b'250 Supported commands: EHLO HELO MAIL RCPT ' + \
360	n/a	b'DATA RSET NOOP QUIT VRFY\r\n')
361	n/a
362	n/a	def test_HELP_command(self):
363	n/a	self.write_line(b'HELP MAIL')
364	n/a	self.assertEqual(self.channel.socket.last,
365	n/a	b'250 Syntax: MAIL FROM: <address>\r\n')
366	n/a
367	n/a	def test_HELP_command_unknown(self):
368	n/a	self.write_line(b'HELP SPAM')
369	n/a	self.assertEqual(self.channel.socket.last,
370	n/a	b'501 Supported commands: EHLO HELO MAIL RCPT ' + \
371	n/a	b'DATA RSET NOOP QUIT VRFY\r\n')
372	n/a
373	n/a	def test_HELO_bad_syntax(self):
374	n/a	self.write_line(b'HELO')
375	n/a	self.assertEqual(self.channel.socket.last,
376	n/a	b'501 Syntax: HELO hostname\r\n')
377	n/a
378	n/a	def test_HELO_duplicate(self):
379	n/a	self.write_line(b'HELO example')
380	n/a	self.write_line(b'HELO example')
381	n/a	self.assertEqual(self.channel.socket.last,
382	n/a	b'503 Duplicate HELO/EHLO\r\n')
383	n/a
384	n/a	def test_HELO_parameter_rejected_when_extensions_not_enabled(self):
385	n/a	self.extended_smtp = False
386	n/a	self.write_line(b'HELO example')
387	n/a	self.write_line(b'MAIL from:<foo@example.com> SIZE=1234')
388	n/a	self.assertEqual(self.channel.socket.last,
389	n/a	b'501 Syntax: MAIL FROM: <address>\r\n')
390	n/a
391	n/a	def test_MAIL_allows_space_after_colon(self):
392	n/a	self.write_line(b'HELO example')
393	n/a	self.write_line(b'MAIL from: <foo@example.com>')
394	n/a	self.assertEqual(self.channel.socket.last,
395	n/a	b'250 OK\r\n')
396	n/a
397	n/a	def test_extended_MAIL_allows_space_after_colon(self):
398	n/a	self.write_line(b'EHLO example')
399	n/a	self.write_line(b'MAIL from: <foo@example.com> size=20')
400	n/a	self.assertEqual(self.channel.socket.last,
401	n/a	b'250 OK\r\n')
402	n/a
403	n/a	def test_NOOP(self):
404	n/a	self.write_line(b'NOOP')
405	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
406	n/a
407	n/a	def test_HELO_NOOP(self):
408	n/a	self.write_line(b'HELO example')
409	n/a	self.write_line(b'NOOP')
410	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
411	n/a
412	n/a	def test_NOOP_bad_syntax(self):
413	n/a	self.write_line(b'NOOP hi')
414	n/a	self.assertEqual(self.channel.socket.last,
415	n/a	b'501 Syntax: NOOP\r\n')
416	n/a
417	n/a	def test_QUIT(self):
418	n/a	self.write_line(b'QUIT')
419	n/a	self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
420	n/a
421	n/a	def test_HELO_QUIT(self):
422	n/a	self.write_line(b'HELO example')
423	n/a	self.write_line(b'QUIT')
424	n/a	self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
425	n/a
426	n/a	def test_QUIT_arg_ignored(self):
427	n/a	self.write_line(b'QUIT bye bye')
428	n/a	self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
429	n/a
430	n/a	def test_bad_state(self):
431	n/a	self.channel.smtp_state = 'BAD STATE'
432	n/a	self.write_line(b'HELO example')
433	n/a	self.assertEqual(self.channel.socket.last,
434	n/a	b'451 Internal confusion\r\n')
435	n/a
436	n/a	def test_command_too_long(self):
437	n/a	self.write_line(b'HELO example')
438	n/a	self.write_line(b'MAIL from: ' +
439	n/a	b'a' * self.channel.command_size_limit +
440	n/a	b'@example')
441	n/a	self.assertEqual(self.channel.socket.last,
442	n/a	b'500 Error: line too long\r\n')
443	n/a
444	n/a	def test_MAIL_command_limit_extended_with_SIZE(self):
445	n/a	self.write_line(b'EHLO example')
446	n/a	fill_len = self.channel.command_size_limit - len('MAIL from:<@example>')
447	n/a	self.write_line(b'MAIL from:<' +
448	n/a	b'a' * fill_len +
449	n/a	b'@example> SIZE=1234')
450	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
451	n/a
452	n/a	self.write_line(b'MAIL from:<' +
453	n/a	b'a' * (fill_len + 26) +
454	n/a	b'@example> SIZE=1234')
455	n/a	self.assertEqual(self.channel.socket.last,
456	n/a	b'500 Error: line too long\r\n')
457	n/a
458	n/a	def test_MAIL_command_rejects_SMTPUTF8_by_default(self):
459	n/a	self.write_line(b'EHLO example')
460	n/a	self.write_line(
461	n/a	b'MAIL from: <naive@example.com> BODY=8BITMIME SMTPUTF8')
462	n/a	self.assertEqual(self.channel.socket.last[0:1], b'5')
463	n/a
464	n/a	def test_data_longer_than_default_data_size_limit(self):
465	n/a	# Hack the default so we don't have to generate so much data.
466	n/a	self.channel.data_size_limit = 1048
467	n/a	self.write_line(b'HELO example')
468	n/a	self.write_line(b'MAIL From:eggs@example')
469	n/a	self.write_line(b'RCPT To:spam@example')
470	n/a	self.write_line(b'DATA')
471	n/a	self.write_line(b'A' * self.channel.data_size_limit +
472	n/a	b'A\r\n.')
473	n/a	self.assertEqual(self.channel.socket.last,
474	n/a	b'552 Error: Too much mail data\r\n')
475	n/a
476	n/a	def test_MAIL_size_parameter(self):
477	n/a	self.write_line(b'EHLO example')
478	n/a	self.write_line(b'MAIL FROM:<eggs@example> SIZE=512')
479	n/a	self.assertEqual(self.channel.socket.last,
480	n/a	b'250 OK\r\n')
481	n/a
482	n/a	def test_MAIL_invalid_size_parameter(self):
483	n/a	self.write_line(b'EHLO example')
484	n/a	self.write_line(b'MAIL FROM:<eggs@example> SIZE=invalid')
485	n/a	self.assertEqual(self.channel.socket.last,
486	n/a	b'501 Syntax: MAIL FROM: <address> [SP <mail-parameters>]\r\n')
487	n/a
488	n/a	def test_MAIL_RCPT_unknown_parameters(self):
489	n/a	self.write_line(b'EHLO example')
490	n/a	self.write_line(b'MAIL FROM:<eggs@example> ham=green')
491	n/a	self.assertEqual(self.channel.socket.last,
492	n/a	b'555 MAIL FROM parameters not recognized or not implemented\r\n')
493	n/a
494	n/a	self.write_line(b'MAIL FROM:<eggs@example>')
495	n/a	self.write_line(b'RCPT TO:<eggs@example> ham=green')
496	n/a	self.assertEqual(self.channel.socket.last,
497	n/a	b'555 RCPT TO parameters not recognized or not implemented\r\n')
498	n/a
499	n/a	def test_MAIL_size_parameter_larger_than_default_data_size_limit(self):
500	n/a	self.channel.data_size_limit = 1048
501	n/a	self.write_line(b'EHLO example')
502	n/a	self.write_line(b'MAIL FROM:<eggs@example> SIZE=2096')
503	n/a	self.assertEqual(self.channel.socket.last,
504	n/a	b'552 Error: message size exceeds fixed maximum message size\r\n')
505	n/a
506	n/a	def test_need_MAIL(self):
507	n/a	self.write_line(b'HELO example')
508	n/a	self.write_line(b'RCPT to:spam@example')
509	n/a	self.assertEqual(self.channel.socket.last,
510	n/a	b'503 Error: need MAIL command\r\n')
511	n/a
512	n/a	def test_MAIL_syntax_HELO(self):
513	n/a	self.write_line(b'HELO example')
514	n/a	self.write_line(b'MAIL from eggs@example')
515	n/a	self.assertEqual(self.channel.socket.last,
516	n/a	b'501 Syntax: MAIL FROM: <address>\r\n')
517	n/a
518	n/a	def test_MAIL_syntax_EHLO(self):
519	n/a	self.write_line(b'EHLO example')
520	n/a	self.write_line(b'MAIL from eggs@example')
521	n/a	self.assertEqual(self.channel.socket.last,
522	n/a	b'501 Syntax: MAIL FROM: <address> [SP <mail-parameters>]\r\n')
523	n/a
524	n/a	def test_MAIL_missing_address(self):
525	n/a	self.write_line(b'HELO example')
526	n/a	self.write_line(b'MAIL from:')
527	n/a	self.assertEqual(self.channel.socket.last,
528	n/a	b'501 Syntax: MAIL FROM: <address>\r\n')
529	n/a
530	n/a	def test_MAIL_chevrons(self):
531	n/a	self.write_line(b'HELO example')
532	n/a	self.write_line(b'MAIL from:<eggs@example>')
533	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
534	n/a
535	n/a	def test_MAIL_empty_chevrons(self):
536	n/a	self.write_line(b'EHLO example')
537	n/a	self.write_line(b'MAIL from:<>')
538	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
539	n/a
540	n/a	def test_MAIL_quoted_localpart(self):
541	n/a	self.write_line(b'EHLO example')
542	n/a	self.write_line(b'MAIL from: <"Fred Blogs"@example.com>')
543	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
544	n/a	self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
545	n/a
546	n/a	def test_MAIL_quoted_localpart_no_angles(self):
547	n/a	self.write_line(b'EHLO example')
548	n/a	self.write_line(b'MAIL from: "Fred Blogs"@example.com')
549	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
550	n/a	self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
551	n/a
552	n/a	def test_MAIL_quoted_localpart_with_size(self):
553	n/a	self.write_line(b'EHLO example')
554	n/a	self.write_line(b'MAIL from: <"Fred Blogs"@example.com> SIZE=1000')
555	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
556	n/a	self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
557	n/a
558	n/a	def test_MAIL_quoted_localpart_with_size_no_angles(self):
559	n/a	self.write_line(b'EHLO example')
560	n/a	self.write_line(b'MAIL from: "Fred Blogs"@example.com SIZE=1000')
561	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
562	n/a	self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
563	n/a
564	n/a	def test_nested_MAIL(self):
565	n/a	self.write_line(b'HELO example')
566	n/a	self.write_line(b'MAIL from:eggs@example')
567	n/a	self.write_line(b'MAIL from:spam@example')
568	n/a	self.assertEqual(self.channel.socket.last,
569	n/a	b'503 Error: nested MAIL command\r\n')
570	n/a
571	n/a	def test_VRFY(self):
572	n/a	self.write_line(b'VRFY eggs@example')
573	n/a	self.assertEqual(self.channel.socket.last,
574	n/a	b'252 Cannot VRFY user, but will accept message and attempt ' + \
575	n/a	b'delivery\r\n')
576	n/a
577	n/a	def test_VRFY_syntax(self):
578	n/a	self.write_line(b'VRFY')
579	n/a	self.assertEqual(self.channel.socket.last,
580	n/a	b'501 Syntax: VRFY <address>\r\n')
581	n/a
582	n/a	def test_EXPN_not_implemented(self):
583	n/a	self.write_line(b'EXPN')
584	n/a	self.assertEqual(self.channel.socket.last,
585	n/a	b'502 EXPN not implemented\r\n')
586	n/a
587	n/a	def test_no_HELO_MAIL(self):
588	n/a	self.write_line(b'MAIL from:<foo@example.com>')
589	n/a	self.assertEqual(self.channel.socket.last,
590	n/a	b'503 Error: send HELO first\r\n')
591	n/a
592	n/a	def test_need_RCPT(self):
593	n/a	self.write_line(b'HELO example')
594	n/a	self.write_line(b'MAIL From:eggs@example')
595	n/a	self.write_line(b'DATA')
596	n/a	self.assertEqual(self.channel.socket.last,
597	n/a	b'503 Error: need RCPT command\r\n')
598	n/a
599	n/a	def test_RCPT_syntax_HELO(self):
600	n/a	self.write_line(b'HELO example')
601	n/a	self.write_line(b'MAIL From: eggs@example')
602	n/a	self.write_line(b'RCPT to eggs@example')
603	n/a	self.assertEqual(self.channel.socket.last,
604	n/a	b'501 Syntax: RCPT TO: <address>\r\n')
605	n/a
606	n/a	def test_RCPT_syntax_EHLO(self):
607	n/a	self.write_line(b'EHLO example')
608	n/a	self.write_line(b'MAIL From: eggs@example')
609	n/a	self.write_line(b'RCPT to eggs@example')
610	n/a	self.assertEqual(self.channel.socket.last,
611	n/a	b'501 Syntax: RCPT TO: <address> [SP <mail-parameters>]\r\n')
612	n/a
613	n/a	def test_RCPT_lowercase_to_OK(self):
614	n/a	self.write_line(b'HELO example')
615	n/a	self.write_line(b'MAIL From: eggs@example')
616	n/a	self.write_line(b'RCPT to: <eggs@example>')
617	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
618	n/a
619	n/a	def test_no_HELO_RCPT(self):
620	n/a	self.write_line(b'RCPT to eggs@example')
621	n/a	self.assertEqual(self.channel.socket.last,
622	n/a	b'503 Error: send HELO first\r\n')
623	n/a
624	n/a	def test_data_dialog(self):
625	n/a	self.write_line(b'HELO example')
626	n/a	self.write_line(b'MAIL From:eggs@example')
627	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
628	n/a	self.write_line(b'RCPT To:spam@example')
629	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
630	n/a
631	n/a	self.write_line(b'DATA')
632	n/a	self.assertEqual(self.channel.socket.last,
633	n/a	b'354 End data with <CR><LF>.<CR><LF>\r\n')
634	n/a	self.write_line(b'data\r\nmore\r\n.')
635	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
636	n/a	self.assertEqual(self.server.messages,
637	n/a	[(('peer-address', 'peer-port'),
638	n/a	'eggs@example',
639	n/a	['spam@example'],
640	n/a	'data\nmore')])
641	n/a
642	n/a	def test_DATA_syntax(self):
643	n/a	self.write_line(b'HELO example')
644	n/a	self.write_line(b'MAIL From:eggs@example')
645	n/a	self.write_line(b'RCPT To:spam@example')
646	n/a	self.write_line(b'DATA spam')
647	n/a	self.assertEqual(self.channel.socket.last, b'501 Syntax: DATA\r\n')
648	n/a
649	n/a	def test_no_HELO_DATA(self):
650	n/a	self.write_line(b'DATA spam')
651	n/a	self.assertEqual(self.channel.socket.last,
652	n/a	b'503 Error: send HELO first\r\n')
653	n/a
654	n/a	def test_data_transparency_section_4_5_2(self):
655	n/a	self.write_line(b'HELO example')
656	n/a	self.write_line(b'MAIL From:eggs@example')
657	n/a	self.write_line(b'RCPT To:spam@example')
658	n/a	self.write_line(b'DATA')
659	n/a	self.write_line(b'..\r\n.\r\n')
660	n/a	self.assertEqual(self.channel.received_data, '.')
661	n/a
662	n/a	def test_multiple_RCPT(self):
663	n/a	self.write_line(b'HELO example')
664	n/a	self.write_line(b'MAIL From:eggs@example')
665	n/a	self.write_line(b'RCPT To:spam@example')
666	n/a	self.write_line(b'RCPT To:ham@example')
667	n/a	self.write_line(b'DATA')
668	n/a	self.write_line(b'data\r\n.')
669	n/a	self.assertEqual(self.server.messages,
670	n/a	[(('peer-address', 'peer-port'),
671	n/a	'eggs@example',
672	n/a	['spam@example','ham@example'],
673	n/a	'data')])
674	n/a
675	n/a	def test_manual_status(self):
676	n/a	# checks that the Channel is able to return a custom status message
677	n/a	self.write_line(b'HELO example')
678	n/a	self.write_line(b'MAIL From:eggs@example')
679	n/a	self.write_line(b'RCPT To:spam@example')
680	n/a	self.write_line(b'DATA')
681	n/a	self.write_line(b'return status\r\n.')
682	n/a	self.assertEqual(self.channel.socket.last, b'250 Okish\r\n')
683	n/a
684	n/a	def test_RSET(self):
685	n/a	self.write_line(b'HELO example')
686	n/a	self.write_line(b'MAIL From:eggs@example')
687	n/a	self.write_line(b'RCPT To:spam@example')
688	n/a	self.write_line(b'RSET')
689	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
690	n/a	self.write_line(b'MAIL From:foo@example')
691	n/a	self.write_line(b'RCPT To:eggs@example')
692	n/a	self.write_line(b'DATA')
693	n/a	self.write_line(b'data\r\n.')
694	n/a	self.assertEqual(self.server.messages,
695	n/a	[(('peer-address', 'peer-port'),
696	n/a	'foo@example',
697	n/a	['eggs@example'],
698	n/a	'data')])
699	n/a
700	n/a	def test_HELO_RSET(self):
701	n/a	self.write_line(b'HELO example')
702	n/a	self.write_line(b'RSET')
703	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
704	n/a
705	n/a	def test_RSET_syntax(self):
706	n/a	self.write_line(b'RSET hi')
707	n/a	self.assertEqual(self.channel.socket.last, b'501 Syntax: RSET\r\n')
708	n/a
709	n/a	def test_unknown_command(self):
710	n/a	self.write_line(b'UNKNOWN_CMD')
711	n/a	self.assertEqual(self.channel.socket.last,
712	n/a	b'500 Error: command "UNKNOWN_CMD" not ' + \
713	n/a	b'recognized\r\n')
714	n/a
715	n/a	def test_attribute_deprecations(self):
716	n/a	with support.check_warnings(('', DeprecationWarning)):
717	n/a	spam = self.channel._SMTPChannel__server
718	n/a	with support.check_warnings(('', DeprecationWarning)):
719	n/a	self.channel._SMTPChannel__server = 'spam'
720	n/a	with support.check_warnings(('', DeprecationWarning)):
721	n/a	spam = self.channel._SMTPChannel__line
722	n/a	with support.check_warnings(('', DeprecationWarning)):
723	n/a	self.channel._SMTPChannel__line = 'spam'
724	n/a	with support.check_warnings(('', DeprecationWarning)):
725	n/a	spam = self.channel._SMTPChannel__state
726	n/a	with support.check_warnings(('', DeprecationWarning)):
727	n/a	self.channel._SMTPChannel__state = 'spam'
728	n/a	with support.check_warnings(('', DeprecationWarning)):
729	n/a	spam = self.channel._SMTPChannel__greeting
730	n/a	with support.check_warnings(('', DeprecationWarning)):
731	n/a	self.channel._SMTPChannel__greeting = 'spam'
732	n/a	with support.check_warnings(('', DeprecationWarning)):
733	n/a	spam = self.channel._SMTPChannel__mailfrom
734	n/a	with support.check_warnings(('', DeprecationWarning)):
735	n/a	self.channel._SMTPChannel__mailfrom = 'spam'
736	n/a	with support.check_warnings(('', DeprecationWarning)):
737	n/a	spam = self.channel._SMTPChannel__rcpttos
738	n/a	with support.check_warnings(('', DeprecationWarning)):
739	n/a	self.channel._SMTPChannel__rcpttos = 'spam'
740	n/a	with support.check_warnings(('', DeprecationWarning)):
741	n/a	spam = self.channel._SMTPChannel__data
742	n/a	with support.check_warnings(('', DeprecationWarning)):
743	n/a	self.channel._SMTPChannel__data = 'spam'
744	n/a	with support.check_warnings(('', DeprecationWarning)):
745	n/a	spam = self.channel._SMTPChannel__fqdn
746	n/a	with support.check_warnings(('', DeprecationWarning)):
747	n/a	self.channel._SMTPChannel__fqdn = 'spam'
748	n/a	with support.check_warnings(('', DeprecationWarning)):
749	n/a	spam = self.channel._SMTPChannel__peer
750	n/a	with support.check_warnings(('', DeprecationWarning)):
751	n/a	self.channel._SMTPChannel__peer = 'spam'
752	n/a	with support.check_warnings(('', DeprecationWarning)):
753	n/a	spam = self.channel._SMTPChannel__conn
754	n/a	with support.check_warnings(('', DeprecationWarning)):
755	n/a	self.channel._SMTPChannel__conn = 'spam'
756	n/a	with support.check_warnings(('', DeprecationWarning)):
757	n/a	spam = self.channel._SMTPChannel__addr
758	n/a	with support.check_warnings(('', DeprecationWarning)):
759	n/a	self.channel._SMTPChannel__addr = 'spam'
760	n/a
761	n/a	@unittest.skipUnless(support.IPV6_ENABLED, "IPv6 not enabled")
762	n/a	class SMTPDChannelIPv6Test(SMTPDChannelTest):
763	n/a	def setUp(self):
764	n/a	smtpd.socket = asyncore.socket = mock_socket
765	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
766	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
767	n/a	self.server = DummyServer((support.HOSTv6, 0), ('b', 0),
768	n/a	decode_data=True)
769	n/a	conn, addr = self.server.accept()
770	n/a	self.channel = smtpd.SMTPChannel(self.server, conn, addr,
771	n/a	decode_data=True)
772	n/a
773	n/a	class SMTPDChannelWithDataSizeLimitTest(unittest.TestCase):
774	n/a
775	n/a	def setUp(self):
776	n/a	smtpd.socket = asyncore.socket = mock_socket
777	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
778	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
779	n/a	self.server = DummyServer((support.HOST, 0), ('b', 0),
780	n/a	decode_data=True)
781	n/a	conn, addr = self.server.accept()
782	n/a	# Set DATA size limit to 32 bytes for easy testing
783	n/a	self.channel = smtpd.SMTPChannel(self.server, conn, addr, 32,
784	n/a	decode_data=True)
785	n/a
786	n/a	def tearDown(self):
787	n/a	asyncore.close_all()
788	n/a	asyncore.socket = smtpd.socket = socket
789	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
790	n/a
791	n/a	def write_line(self, line):
792	n/a	self.channel.socket.queue_recv(line)
793	n/a	self.channel.handle_read()
794	n/a
795	n/a	def test_data_limit_dialog(self):
796	n/a	self.write_line(b'HELO example')
797	n/a	self.write_line(b'MAIL From:eggs@example')
798	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
799	n/a	self.write_line(b'RCPT To:spam@example')
800	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
801	n/a
802	n/a	self.write_line(b'DATA')
803	n/a	self.assertEqual(self.channel.socket.last,
804	n/a	b'354 End data with <CR><LF>.<CR><LF>\r\n')
805	n/a	self.write_line(b'data\r\nmore\r\n.')
806	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
807	n/a	self.assertEqual(self.server.messages,
808	n/a	[(('peer-address', 'peer-port'),
809	n/a	'eggs@example',
810	n/a	['spam@example'],
811	n/a	'data\nmore')])
812	n/a
813	n/a	def test_data_limit_dialog_too_much_data(self):
814	n/a	self.write_line(b'HELO example')
815	n/a	self.write_line(b'MAIL From:eggs@example')
816	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
817	n/a	self.write_line(b'RCPT To:spam@example')
818	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
819	n/a
820	n/a	self.write_line(b'DATA')
821	n/a	self.assertEqual(self.channel.socket.last,
822	n/a	b'354 End data with <CR><LF>.<CR><LF>\r\n')
823	n/a	self.write_line(b'This message is longer than 32 bytes\r\n.')
824	n/a	self.assertEqual(self.channel.socket.last,
825	n/a	b'552 Error: Too much mail data\r\n')
826	n/a
827	n/a
828	n/a	class SMTPDChannelWithDecodeDataFalse(unittest.TestCase):
829	n/a
830	n/a	def setUp(self):
831	n/a	smtpd.socket = asyncore.socket = mock_socket
832	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
833	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
834	n/a	self.server = DummyServer((support.HOST, 0), ('b', 0))
835	n/a	conn, addr = self.server.accept()
836	n/a	self.channel = smtpd.SMTPChannel(self.server, conn, addr)
837	n/a
838	n/a	def tearDown(self):
839	n/a	asyncore.close_all()
840	n/a	asyncore.socket = smtpd.socket = socket
841	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
842	n/a
843	n/a	def write_line(self, line):
844	n/a	self.channel.socket.queue_recv(line)
845	n/a	self.channel.handle_read()
846	n/a
847	n/a	def test_ascii_data(self):
848	n/a	self.write_line(b'HELO example')
849	n/a	self.write_line(b'MAIL From:eggs@example')
850	n/a	self.write_line(b'RCPT To:spam@example')
851	n/a	self.write_line(b'DATA')
852	n/a	self.write_line(b'plain ascii text')
853	n/a	self.write_line(b'.')
854	n/a	self.assertEqual(self.channel.received_data, b'plain ascii text')
855	n/a
856	n/a	def test_utf8_data(self):
857	n/a	self.write_line(b'HELO example')
858	n/a	self.write_line(b'MAIL From:eggs@example')
859	n/a	self.write_line(b'RCPT To:spam@example')
860	n/a	self.write_line(b'DATA')
861	n/a	self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
862	n/a	self.write_line(b'and some plain ascii')
863	n/a	self.write_line(b'.')
864	n/a	self.assertEqual(
865	n/a	self.channel.received_data,
866	n/a	b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87\n'
867	n/a	b'and some plain ascii')
868	n/a
869	n/a
870	n/a	class SMTPDChannelWithDecodeDataTrue(unittest.TestCase):
871	n/a
872	n/a	def setUp(self):
873	n/a	smtpd.socket = asyncore.socket = mock_socket
874	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
875	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
876	n/a	self.server = DummyServer((support.HOST, 0), ('b', 0),
877	n/a	decode_data=True)
878	n/a	conn, addr = self.server.accept()
879	n/a	# Set decode_data to True
880	n/a	self.channel = smtpd.SMTPChannel(self.server, conn, addr,
881	n/a	decode_data=True)
882	n/a
883	n/a	def tearDown(self):
884	n/a	asyncore.close_all()
885	n/a	asyncore.socket = smtpd.socket = socket
886	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
887	n/a
888	n/a	def write_line(self, line):
889	n/a	self.channel.socket.queue_recv(line)
890	n/a	self.channel.handle_read()
891	n/a
892	n/a	def test_ascii_data(self):
893	n/a	self.write_line(b'HELO example')
894	n/a	self.write_line(b'MAIL From:eggs@example')
895	n/a	self.write_line(b'RCPT To:spam@example')
896	n/a	self.write_line(b'DATA')
897	n/a	self.write_line(b'plain ascii text')
898	n/a	self.write_line(b'.')
899	n/a	self.assertEqual(self.channel.received_data, 'plain ascii text')
900	n/a
901	n/a	def test_utf8_data(self):
902	n/a	self.write_line(b'HELO example')
903	n/a	self.write_line(b'MAIL From:eggs@example')
904	n/a	self.write_line(b'RCPT To:spam@example')
905	n/a	self.write_line(b'DATA')
906	n/a	self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
907	n/a	self.write_line(b'and some plain ascii')
908	n/a	self.write_line(b'.')
909	n/a	self.assertEqual(
910	n/a	self.channel.received_data,
911	n/a	'utf8 enriched text: Å¼ÅºÄ‡\nand some plain ascii')
912	n/a
913	n/a
914	n/a	class SMTPDChannelTestWithEnableSMTPUTF8True(unittest.TestCase):
915	n/a	def setUp(self):
916	n/a	smtpd.socket = asyncore.socket = mock_socket
917	n/a	self.old_debugstream = smtpd.DEBUGSTREAM
918	n/a	self.debug = smtpd.DEBUGSTREAM = io.StringIO()
919	n/a	self.server = DummyServer((support.HOST, 0), ('b', 0),
920	n/a	enable_SMTPUTF8=True)
921	n/a	conn, addr = self.server.accept()
922	n/a	self.channel = smtpd.SMTPChannel(self.server, conn, addr,
923	n/a	enable_SMTPUTF8=True)
924	n/a
925	n/a	def tearDown(self):
926	n/a	asyncore.close_all()
927	n/a	asyncore.socket = smtpd.socket = socket
928	n/a	smtpd.DEBUGSTREAM = self.old_debugstream
929	n/a
930	n/a	def write_line(self, line):
931	n/a	self.channel.socket.queue_recv(line)
932	n/a	self.channel.handle_read()
933	n/a
934	n/a	def test_MAIL_command_accepts_SMTPUTF8_when_announced(self):
935	n/a	self.write_line(b'EHLO example')
936	n/a	self.write_line(
937	n/a	'MAIL from: <naiÌˆve@example.com> BODY=8BITMIME SMTPUTF8'.encode(
938	n/a	'utf-8')
939	n/a	)
940	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
941	n/a
942	n/a	def test_process_smtputf8_message(self):
943	n/a	self.write_line(b'EHLO example')
944	n/a	for mail_parameters in [b'', b'BODY=8BITMIME SMTPUTF8']:
945	n/a	self.write_line(b'MAIL from: <a@example> ' + mail_parameters)
946	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
947	n/a	self.write_line(b'rcpt to:<b@example.com>')
948	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
949	n/a	self.write_line(b'data')
950	n/a	self.assertEqual(self.channel.socket.last[0:3], b'354')
951	n/a	self.write_line(b'c\r\n.')
952	n/a	if mail_parameters == b'':
953	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
954	n/a	else:
955	n/a	self.assertEqual(self.channel.socket.last,
956	n/a	b'250 SMTPUTF8 message okish\r\n')
957	n/a
958	n/a	def test_utf8_data(self):
959	n/a	self.write_line(b'EHLO example')
960	n/a	self.write_line(
961	n/a	'MAIL From: naiÌˆve@examplÃ© BODY=8BITMIME SMTPUTF8'.encode('utf-8'))
962	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
963	n/a	self.write_line('RCPT To:spÃ¤m@examplÃ©'.encode('utf-8'))
964	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
965	n/a	self.write_line(b'DATA')
966	n/a	self.assertEqual(self.channel.socket.last[0:3], b'354')
967	n/a	self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
968	n/a	self.write_line(b'.')
969	n/a	self.assertEqual(
970	n/a	self.channel.received_data,
971	n/a	b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
972	n/a
973	n/a	def test_MAIL_command_limit_extended_with_SIZE_and_SMTPUTF8(self):
974	n/a	self.write_line(b'ehlo example')
975	n/a	fill_len = (512 + 26 + 10) - len('mail from:<@example>')
976	n/a	self.write_line(b'MAIL from:<' +
977	n/a	b'a' * (fill_len + 1) +
978	n/a	b'@example>')
979	n/a	self.assertEqual(self.channel.socket.last,
980	n/a	b'500 Error: line too long\r\n')
981	n/a	self.write_line(b'MAIL from:<' +
982	n/a	b'a' * fill_len +
983	n/a	b'@example>')
984	n/a	self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
985	n/a
986	n/a	def test_multiple_emails_with_extended_command_length(self):
987	n/a	self.write_line(b'ehlo example')
988	n/a	fill_len = (512 + 26 + 10) - len('mail from:<@example>')
989	n/a	for char in [b'a', b'b', b'c']:
990	n/a	self.write_line(b'MAIL from:<' + char * fill_len + b'a@example>')
991	n/a	self.assertEqual(self.channel.socket.last[0:3], b'500')
992	n/a	self.write_line(b'MAIL from:<' + char * fill_len + b'@example>')
993	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
994	n/a	self.write_line(b'rcpt to:<hans@example.com>')
995	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
996	n/a	self.write_line(b'data')
997	n/a	self.assertEqual(self.channel.socket.last[0:3], b'354')
998	n/a	self.write_line(b'test\r\n.')
999	n/a	self.assertEqual(self.channel.socket.last[0:3], b'250')
1000	n/a
1001	n/a
1002	n/a	class MiscTestCase(unittest.TestCase):
1003	n/a	def test__all__(self):
1004	n/a	blacklist = {
1005	n/a	"program", "Devnull", "DEBUGSTREAM", "NEWLINE", "COMMASPACE",
1006	n/a	"DATA_SIZE_DEFAULT", "usage", "Options", "parseargs",
1007	n/a
1008	n/a	}
1009	n/a	support.check__all__(self, smtpd, blacklist=blacklist)
1010	n/a
1011	n/a
1012	n/a	if __name__ == "__main__":
1013	n/a	unittest.main()