Lines Matching refs:self
39 def handle_expt(self):
68 def setUp(self):
70 self.port = 25
72 def tearDown(self):
77 def testQuoteData(self):
80 self.assertEqual(expected, smtplib.quotedata(teststr))
82 def testBasic1(self):
85 client = self.client(HOST, self.port)
88 def testSourceAddress(self):
91 client = self.client(HOST, self.port,
93 self.assertEqual(client.source_address, ('127.0.0.1', 19876))
96 def testBasic2(self):
99 client = self.client("%s:%s" % (HOST, self.port))
102 def testLocalHostName(self):
105 client = self.client(HOST, self.port, local_hostname="testhost")
106 self.assertEqual(client.local_hostname, "testhost")
109 def testTimeoutDefault(self):
111 self.assertIsNone(mock_socket.getdefaulttimeout())
113 self.assertEqual(mock_socket.getdefaulttimeout(), 30)
115 client = self.client(HOST, self.port)
118 self.assertEqual(client.sock.gettimeout(), 30)
121 def testTimeoutNone(self):
123 self.assertIsNone(socket.getdefaulttimeout())
126 client = self.client(HOST, self.port, timeout=None)
129 self.assertIsNone(client.sock.gettimeout())
132 def testTimeoutZero(self):
134 with self.assertRaises(ValueError):
135 self.client(HOST, self.port, timeout=0)
137 def testTimeoutValue(self):
139 client = self.client(HOST, self.port, timeout=30)
140 self.assertEqual(client.sock.gettimeout(), 30)
143 def test_debuglevel(self):
145 client = self.client()
148 client.connect(HOST, self.port)
151 self.assertRegex(stderr.getvalue(), expected)
153 def test_debuglevel_2(self):
155 client = self.client()
158 client.connect(HOST, self.port)
162 self.assertRegex(stderr.getvalue(), expected)
175 def testUnixDomainSocketTimeoutDefault(self):
179 client = self.client(local_host, self.port)
182 self.assertIsNone(client.sock.gettimeout())
185 def testTimeoutZero(self):
188 with self.assertRaises(ValueError):
189 self.client(local_host, timeout=0)
236 def setUp(self):
237 self.thread_key = threading_helper.threading_setup()
238 self.real_getfqdn = socket.getfqdn
241 self.old_stdout = sys.stdout
242 self.output = io.StringIO()
243 sys.stdout = self.output
245 self.serv_evt = threading.Event()
246 self.client_evt = threading.Event()
248 self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
251 self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1),
254 self.host, self.port = self.serv.socket.getsockname()[:2]
255 serv_args = (self.serv, self.serv_evt, self.client_evt)
256 self.thread = threading.Thread(target=debugging_server, args=serv_args)
257 self.thread.start()
260 self.serv_evt.wait()
261 self.serv_evt.clear()
263 def tearDown(self):
264 socket.getfqdn = self.real_getfqdn
266 self.client_evt.set()
268 self.serv_evt.wait()
269 threading_helper.join_thread(self.thread)
271 sys.stdout = self.old_stdout
274 smtpd.DEBUGSTREAM = self.old_DEBUGSTREAM
275 del self.thread
276 self.doCleanups()
277 threading_helper.threading_cleanup(*self.thread_key)
279 def get_output_without_xpeer(self):
280 test_output = self.output.getvalue()
284 def testBasic(self):
286 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
290 def testSourceAddress(self):
294 smtp = smtplib.SMTP(self.host, self.port, local_hostname='localhost',
296 source_address=(self.host, src_port))
297 self.addCleanup(smtp.close)
298 self.assertEqual(smtp.source_address, (self.host, src_port))
299 self.assertEqual(smtp.local_hostname, 'localhost')
303 self.skipTest("couldn't bind to source port %d" % src_port)
306 def testNOOP(self):
307 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
309 self.addCleanup(smtp.close)
311 self.assertEqual(smtp.noop(), expected)
314 def testRSET(self):
315 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
317 self.addCleanup(smtp.close)
319 self.assertEqual(smtp.rset(), expected)
322 def testELHO(self):
324 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
326 self.addCleanup(smtp.close)
328 self.assertEqual(smtp.ehlo(), expected)
331 def testEXPNNotImplemented(self):
333 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
335 self.addCleanup(smtp.close)
338 self.assertEqual(smtp.getreply(), expected)
341 def test_issue43124_putcmd_escapes_newline(self):
343 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
345 self.addCleanup(smtp.close)
346 with self.assertRaises(ValueError) as exc:
348 self.assertIn("prohibited newline characters", str(exc.exception))
351 def testVRFY(self):
352 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
354 self.addCleanup(smtp.close)
357 self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
358 self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
361 def testSecondHELO(self):
364 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
366 self.addCleanup(smtp.close)
369 self.assertEqual(smtp.helo(), expected)
372 def testHELP(self):
373 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
375 self.addCleanup(smtp.close)
376 self.assertEqual(smtp.help(), b'Supported commands: EHLO HELO MAIL ' + \
380 def testSend(self):
383 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
385 self.addCleanup(smtp.close)
393 self.client_evt.set()
394 self.serv_evt.wait()
395 self.output.flush()
397 self.assertEqual(self.output.getvalue(), mexpect)
399 def testSendBinary(self):
401 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
403 self.addCleanup(smtp.close)
409 self.client_evt.set()
410 self.serv_evt.wait()
411 self.output.flush()
413 self.assertEqual(self.output.getvalue(), mexpect)
415 def testSendNeedingDotQuote(self):
418 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
420 self.addCleanup(smtp.close)
426 self.client_evt.set()
427 self.serv_evt.wait()
428 self.output.flush()
430 self.assertEqual(self.output.getvalue(), mexpect)
432 def test_issue43124_escape_localhostname(self):
436 smtp = smtplib.SMTP(HOST, self.port, local_hostname='hi\nX-INJECTED',
438 self.addCleanup(smtp.close)
439 with self.assertRaises(ValueError) as exc:
441 self.assertIn(
450 self.assertNotIn("X-INJECTED", debugout)
452 def test_issue43124_escape_options(self):
457 HOST, self.port, local_hostname='localhost',
460 self.addCleanup(smtp.close)
462 with self.assertRaises(ValueError) as exc:
465 self.assertIn("prohibited newline characters", msg)
466 self.assertIn("X-OPTION\\nX-INJECTED-1 X-OPTION2\\nX-INJECTED-2", msg)
472 self.assertNotIn("X-OPTION", debugout)
473 self.assertNotIn("X-OPTION2", debugout)
474 self.assertNotIn("X-INJECTED-1", debugout)
475 self.assertNotIn("X-INJECTED-2", debugout)
477 def testSendNullSender(self):
479 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
481 self.addCleanup(smtp.close)
487 self.client_evt.set()
488 self.serv_evt.wait()
489 self.output.flush()
491 self.assertEqual(self.output.getvalue(), mexpect)
494 self.assertRegex(debugout, sender)
496 def testSendMessage(self):
498 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
500 self.addCleanup(smtp.close)
506 self.client_evt.set()
507 self.serv_evt.wait()
508 self.output.flush()
513 test_output = self.get_output_without_xpeer()
516 self.assertEqual(test_output, mexpect)
518 def testSendMessageWithAddresses(self):
524 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
526 self.addCleanup(smtp.close)
532 self.assertEqual(m['Bcc'], 'John Root <root@localhost>, "Dinsdale" '
535 self.client_evt.set()
536 self.serv_evt.wait()
537 self.output.flush()
539 test_output = self.get_output_without_xpeer()
544 self.assertEqual(test_output, mexpect)
547 self.assertRegex(debugout, sender)
552 self.assertRegex(debugout, to_addr)
554 def testSendMessageWithSomeAddresses(self):
559 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
561 self.addCleanup(smtp.close)
567 self.client_evt.set()
568 self.serv_evt.wait()
569 self.output.flush()
571 test_output = self.get_output_without_xpeer()
574 self.assertEqual(test_output, mexpect)
577 self.assertRegex(debugout, sender)
581 self.assertRegex(debugout, to_addr)
583 def testSendMessageWithSpecifiedAddresses(self):
588 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
590 self.addCleanup(smtp.close)
596 self.client_evt.set()
597 self.serv_evt.wait()
598 self.output.flush()
600 test_output = self.get_output_without_xpeer()
603 self.assertEqual(test_output, mexpect)
606 self.assertRegex(debugout, sender)
610 self.assertNotRegex(debugout, to_addr)
612 self.assertRegex(debugout, recip)
614 def testSendMessageWithMultipleFrom(self):
620 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
622 self.addCleanup(smtp.close)
628 self.client_evt.set()
629 self.serv_evt.wait()
630 self.output.flush()
632 test_output = self.get_output_without_xpeer()
635 self.assertEqual(test_output, mexpect)
638 self.assertRegex(debugout, sender)
642 self.assertRegex(debugout, to_addr)
644 def testSendMessageResent(self):
654 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
656 self.addCleanup(smtp.close)
662 self.client_evt.set()
663 self.serv_evt.wait()
664 self.output.flush()
669 test_output = self.get_output_without_xpeer()
672 self.assertEqual(test_output, mexpect)
675 self.assertRegex(debugout, sender)
679 self.assertRegex(debugout, to_addr)
681 def testSendMessageMultipleResentRaises(self):
694 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
696 self.addCleanup(smtp.close)
697 with self.assertRaises(ValueError):
703 def testNotConnected(self):
709 self.assertRaises(smtplib.SMTPServerDisconnected, smtp.ehlo)
710 self.assertRaises(smtplib.SMTPServerDisconnected,
713 def testNonnumericPort(self):
715 self.assertRaises(OSError, smtplib.SMTP,
717 self.assertRaises(OSError, smtplib.SMTP,
720 def testSockAttributeExists(self):
725 self.assertIsNone(smtp.sock)
730 def setUp(self):
731 self.msg = EmailMessage()
732 self.msg['From'] = 'Páolo <főo@bar.com>'
733 self.smtp = smtplib.SMTP()
734 self.smtp.ehlo = Mock(return_value=(200, 'OK'))
735 self.smtp.has_extn, self.smtp.sendmail = Mock(), Mock()
737 def testSendMessage(self):
739 self.smtp.send_message(self.msg)
740 self.smtp.send_message(self.msg)
741 self.assertEqual(self.smtp.sendmail.call_args_list[0][0][3],
743 self.assertEqual(self.smtp.sendmail.call_args_list[1][0][3],
746 def testSendMessageWithMailOptions(self):
749 self.smtp.send_message(self.msg, None, None, mail_options)
750 self.assertEqual(mail_options, ['STARTTLS'])
751 self.assertEqual(self.smtp.sendmail.call_args_list[0][0][3],
758 def setUp(self):
761 self.old_stdout = sys.stdout
762 self.output = io.StringIO()
763 sys.stdout = self.output
764 self.port = 25
766 def tearDown(self):
768 sys.stdout = self.old_stdout
770 def testFailingHELO(self):
771 self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP,
772 HOST, self.port, 'localhost', 3)
778 def setUp(self):
779 self.thread_key = threading_helper.threading_setup()
780 self.old_stdout = sys.stdout
781 self.output = io.StringIO()
782 sys.stdout = self.output
784 self.evt = threading.Event()
785 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
786 self.sock.settimeout(15)
787 self.port = socket_helper.bind_port(self.sock)
788 servargs = (self.evt, self.respdata, self.sock)
789 self.thread = threading.Thread(target=server, args=servargs)
790 self.thread.start()
791 self.evt.wait()
792 self.evt.clear()
794 def tearDown(self):
795 self.evt.wait()
796 sys.stdout = self.old_stdout
797 threading_helper.join_thread(self.thread)
798 del self.thread
799 self.doCleanups()
800 threading_helper.threading_cleanup(*self.thread_key)
802 def testLineTooLong(self):
803 self.assertRaises(smtplib.SMTPResponseException, smtplib.SMTP,
804 HOST, self.port, 'localhost', 3)
833 def __init__(self, extra_features, *args, **kw):
834 self._extrafeatures = ''.join(
836 super(SimSMTPChannel, self).__init__(*args, **kw)
839 def found_terminator(self):
840 if self.smtp_state == self.AUTH:
841 line = self._emptystring.join(self.received_lines)
843 self.received_lines = []
845 self.auth_object(line)
847 self.smtp_state = self.COMMAND
848 self.push('%s %s' % (e.smtp_code, e.smtp_error))
853 def smtp_AUTH(self, arg):
854 if not self.seen_greeting:
855 self.push('503 Error: send EHLO first')
857 if not self.extended_smtp or 'AUTH' not in self._extrafeatures:
858 self.push('500 Error: command "AUTH" not recognized')
860 if self.authenticated_user is not None:
861 self.push(
866 self.push('501 Syntax: AUTH <mechanism> [initial-response]')
870 self.auth_object = getattr(self, auth_object_name)
872 self.push('504 Command parameter not implemented: unsupported '
875 self.smtp_state = self.AUTH
876 self.auth_object(args[1] if len(args) == 2 else None)
878 def _authenticated(self, user, valid):
880 self.authenticated_user = user
881 self.push('235 Authentication Succeeded')
883 self.push('535 Authentication credentials invalid')
884 self.smtp_state = self.COMMAND
886 def _decode_base64(self, string):
889 def _auth_plain(self, arg=None):
891 self.push('334 ')
893 logpass = self._decode_base64(arg)
897 self.push('535 Splitting response {!r} into user and password'
900 self._authenticated(user, password == sim_auth[1])
902 def _auth_login(self, arg=None):
905 self.push('334 VXNlcm5hbWU6')
906 elif not hasattr(self, '_auth_login_user'):
907 self._auth_login_user = self._decode_base64(arg)
909 self.push('334 UGFzc3dvcmQ6')
911 password = self._decode_base64(arg)
912 self._authenticated(self._auth_login_user, password == sim_auth[1])
913 del self._auth_login_user
915 def _auth_buggy(self, arg=None):
918 self.push('334 QnVHZ1liVWdHeQ==')
920 def _auth_cram_md5(self, arg=None):
922 self.push('334 {}'.format(sim_cram_md5_challenge))
924 logpass = self._decode_base64(arg)
928 self.push('535 Splitting response {!r} into user and password '
933 self._decode_base64(sim_cram_md5_challenge).encode('ascii'),
935 self._authenticated(user, hashed_pass == valid_hashed_pass)
938 def smtp_EHLO(self, arg):
944 resp = resp + self._extrafeatures + '250 HELP'
945 self.push(resp)
946 self.seen_greeting = arg
947 self.extended_smtp = True
949 def smtp_VRFY(self, arg):
952 self.push('250 %s %s' % (sim_users[arg], smtplib.quoteaddr(arg)))
954 self.push('550 No such user: %s' % arg)
956 def smtp_EXPN(self, arg):
963 self.push('250-%s %s' % (sim_users[user_email], quoted_addr))
965 self.push('250 %s %s' % (sim_users[user_email], quoted_addr))
967 self.push('550 No access for you!')
969 def smtp_QUIT(self, arg):
970 if self.quit_response is None:
971 super(SimSMTPChannel, self).smtp_QUIT(arg)
973 self.push(self.quit_response)
974 self.close_when_done()
976 def smtp_MAIL(self, arg):
977 if self.mail_response is None:
980 self.push(self.mail_response)
981 if self.disconnect:
982 self.close_when_done()
984 def smtp_RCPT(self, arg):
985 if self.rcpt_response is None:
988 self.rcpt_count += 1
989 self.push(self.rcpt_response[self.rcpt_count-1])
991 def smtp_RSET(self, arg):
992 self.rset_count += 1
995 def smtp_DATA(self, arg):
996 if self.data_response is None:
999 self.push(self.data_response)
1001 def handle_error(self):
1009 def __init__(self, *args, **kw):
1010 self._extra_features = []
1011 self._addresses = {}
1012 smtpd.SMTPServer.__init__(self, *args, **kw)
1014 def handle_accepted(self, conn, addr):
1015 self._SMTPchannel = self.channel_class(
1016 self._extra_features, self, conn, addr,
1017 decode_data=self._decode_data)
1019 def process_message(self, peer, mailfrom, rcpttos, data):
1020 self._addresses['from'] = mailfrom
1021 self._addresses['tos'] = rcpttos
1023 def add_feature(self, feature):
1024 self._extra_features.append(feature)
1026 def handle_error(self):
1034 def setUp(self):
1035 self.thread_key = threading_helper.threading_setup()
1036 self.real_getfqdn = socket.getfqdn
1038 self.serv_evt = threading.Event()
1039 self.client_evt = threading.Event()
1041 self.serv = SimSMTPServer((HOST, 0), ('nowhere', -1), decode_data=True)
1043 self.port = self.serv.socket.getsockname()[1]
1044 serv_args = (self.serv, self.serv_evt, self.client_evt)
1045 self.thread = threading.Thread(target=debugging_server, args=serv_args)
1046 self.thread.start()
1049 self.serv_evt.wait()
1050 self.serv_evt.clear()
1052 def tearDown(self):
1053 socket.getfqdn = self.real_getfqdn
1055 self.client_evt.set()
1057 self.serv_evt.wait()
1058 threading_helper.join_thread(self.thread)
1059 del self.thread
1060 self.doCleanups()
1061 threading_helper.threading_cleanup(*self.thread_key)
1063 def testBasic(self):
1065 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1069 def testEHLO(self):
1070 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1074 self.assertEqual(smtp.esmtp_features, {})
1085 self.assertEqual(smtp.esmtp_features, expected_features)
1087 self.assertTrue(smtp.has_extn(k))
1088 self.assertFalse(smtp.has_extn('unsupported-feature'))
1091 def testVRFY(self):
1092 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1099 self.assertEqual(smtp.vrfy(addr_spec), expected_known)
1103 self.assertEqual(smtp.vrfy(u), expected_unknown)
1106 def testEXPN(self):
1107 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1115 self.assertEqual(smtp.expn(listname), expected_known)
1119 self.assertEqual(smtp.expn(u), expected_unknown)
1122 def testAUTH_PLAIN(self):
1123 self.serv.add_feature("AUTH PLAIN")
1124 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1127 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1130 def testAUTH_LOGIN(self):
1131 self.serv.add_feature("AUTH LOGIN")
1132 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1135 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1138 def testAUTH_LOGIN_initial_response_ok(self):
1139 self.serv.add_feature("AUTH LOGIN")
1140 with smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1145 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1147 def testAUTH_LOGIN_initial_response_notok(self):
1148 self.serv.add_feature("AUTH LOGIN")
1149 with smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1154 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1156 def testAUTH_BUGGY(self):
1157 self.serv.add_feature("AUTH BUGGY")
1160 self.assertEqual(b"BuGgYbUgGy", challenge)
1164 HOST, self.port, local_hostname='localhost',
1171 with self.assertRaisesRegex(smtplib.SMTPException, expect) as cm:
1177 def testAUTH_CRAM_MD5(self):
1178 self.serv.add_feature("AUTH CRAM-MD5")
1179 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1182 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1186 def testAUTH_multiple(self):
1188 self.serv.add_feature("AUTH BOGUS PLAIN LOGIN CRAM-MD5")
1189 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1192 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1195 def test_auth_function(self):
1204 self.serv.add_feature("AUTH {}".format(mechanism))
1206 with self.subTest(mechanism=mechanism):
1207 smtp = smtplib.SMTP(HOST, self.port,
1214 self.assertEqual(resp, (235, b'Authentication Succeeded'))
1217 def test_quit_resets_greeting(self):
1218 smtp = smtplib.SMTP(HOST, self.port,
1222 self.assertEqual(code, 250)
1223 self.assertIn('size', smtp.esmtp_features)
1225 self.assertNotIn('size', smtp.esmtp_features)
1226 smtp.connect(HOST, self.port)
1227 self.assertNotIn('size', smtp.esmtp_features)
1229 self.assertIn('size', smtp.esmtp_features)
1232 def test_with_statement(self):
1233 with smtplib.SMTP(HOST, self.port) as smtp:
1235 self.assertEqual(code, 250)
1236 self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo')
1237 with smtplib.SMTP(HOST, self.port) as smtp:
1239 self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo')
1241 def test_with_statement_QUIT_failure(self):
1242 with self.assertRaises(smtplib.SMTPResponseException) as error:
1243 with smtplib.SMTP(HOST, self.port) as smtp:
1245 self.serv._SMTPchannel.quit_response = '421 QUIT FAILED'
1246 self.assertEqual(error.exception.smtp_code, 421)
1247 self.assertEqual(error.exception.smtp_error, b'QUIT FAILED')
1253 def test__rest_from_mail_cmd(self):
1254 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1257 self.serv._SMTPchannel.mail_response = '451 Requested action aborted'
1258 self.serv._SMTPchannel.disconnect = True
1259 with self.assertRaises(smtplib.SMTPSenderRefused):
1261 self.assertIsNone(smtp.sock)
1264 def test_421_from_mail_cmd(self):
1265 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1268 self.serv._SMTPchannel.mail_response = '421 closing connection'
1269 with self.assertRaises(smtplib.SMTPSenderRefused):
1271 self.assertIsNone(smtp.sock)
1272 self.assertEqual(self.serv._SMTPchannel.rset_count, 0)
1274 def test_421_from_rcpt_cmd(self):
1275 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1278 self.serv._SMTPchannel.rcpt_response = ['250 accepted', '421 closing']
1279 with self.assertRaises(smtplib.SMTPRecipientsRefused) as r:
1281 self.assertIsNone(smtp.sock)
1282 self.assertEqual(self.serv._SMTPchannel.rset_count, 0)
1283 self.assertDictEqual(r.exception.args[0], {'Frank': (421, b'closing')})
1285 def test_421_from_data_cmd(self):
1287 def found_terminator(self):
1288 if self.smtp_state == self.DATA:
1289 self.push('421 closing')
1292 self.serv.channel_class = MySimSMTPChannel
1293 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1296 with self.assertRaises(smtplib.SMTPDataError):
1298 self.assertIsNone(smtp.sock)
1299 self.assertEqual(self.serv._SMTPchannel.rcpt_count, 0)
1301 def test_smtputf8_NotSupportedError_if_no_server_support(self):
1303 HOST, self.port, local_hostname='localhost',
1305 self.addCleanup(smtp.close)
1307 self.assertTrue(smtp.does_esmtp)
1308 self.assertFalse(smtp.has_extn('smtputf8'))
1309 self.assertRaises(
1313 self.assertRaises(
1317 def test_send_unicode_without_SMTPUTF8(self):
1319 HOST, self.port, local_hostname='localhost',
1321 self.addCleanup(smtp.close)
1322 self.assertRaises(UnicodeEncodeError, smtp.sendmail, 'Alice', 'Böb', '')
1323 self.assertRaises(UnicodeEncodeError, smtp.mail, 'Älice')
1325 def test_send_message_error_on_non_ascii_addrs_if_no_smtputf8(self):
1333 HOST, self.port, local_hostname='localhost',
1335 self.addCleanup(smtp.close)
1336 with self.assertRaises(smtplib.SMTPNotSupportedError):
1339 def test_name_field_not_included_in_envelop_addresses(self):
1341 HOST, self.port, local_hostname='localhost',
1343 self.addCleanup(smtp.close)
1349 self.assertDictEqual(smtp.send_message(message), {})
1351 self.assertEqual(self.serv._addresses['from'], 'michael@example.com')
1352 self.assertEqual(self.serv._addresses['tos'], ['rene@example.com'])
1357 def __init__(self, *args, **kw):
1361 self._extra_features = ['SMTPUTF8', '8BITMIME']
1362 smtpd.SMTPServer.__init__(self, *args, **kw)
1364 def handle_accepted(self, conn, addr):
1365 self._SMTPchannel = self.channel_class(
1366 self._extra_features, self, conn, addr,
1367 decode_data=self._decode_data,
1368 enable_SMTPUTF8=self.enable_SMTPUTF8,
1371 def process_message(self, peer, mailfrom, rcpttos, data, mail_options=None,
1373 self.last_peer = peer
1374 self.last_mailfrom = mailfrom
1375 self.last_rcpttos = rcpttos
1376 self.last_message = data
1377 self.last_mail_options = mail_options
1378 self.last_rcpt_options = rcpt_options
1385 def setUp(self):
1386 self.thread_key = threading_helper.threading_setup()
1387 self.real_getfqdn = socket.getfqdn
1389 self.serv_evt = threading.Event()
1390 self.client_evt = threading.Event()
1392 self.serv = SimSMTPUTF8Server((HOST, 0), ('nowhere', -1),
1396 self.port = self.serv.socket.getsockname()[1]
1397 serv_args = (self.serv, self.serv_evt, self.client_evt)
1398 self.thread = threading.Thread(target=debugging_server, args=serv_args)
1399 self.thread.start()
1402 self.serv_evt.wait()
1403 self.serv_evt.clear()
1405 def tearDown(self):
1406 socket.getfqdn = self.real_getfqdn
1408 self.client_evt.set()
1410 self.serv_evt.wait()
1411 threading_helper.join_thread(self.thread)
1412 del self.thread
1413 self.doCleanups()
1414 threading_helper.threading_cleanup(*self.thread_key)
1416 def test_test_server_supports_extensions(self):
1418 HOST, self.port, local_hostname='localhost',
1420 self.addCleanup(smtp.close)
1422 self.assertTrue(smtp.does_esmtp)
1423 self.assertTrue(smtp.has_extn('smtputf8'))
1425 def test_send_unicode_with_SMTPUTF8_via_sendmail(self):
1428 HOST, self.port, local_hostname='localhost',
1430 self.addCleanup(smtp.close)
1433 self.assertEqual(self.serv.last_mailfrom, 'Jőhn')
1434 self.assertEqual(self.serv.last_rcpttos, ['Sálly'])
1435 self.assertEqual(self.serv.last_message, m)
1436 self.assertIn('BODY=8BITMIME', self.serv.last_mail_options)
1437 self.assertIn('SMTPUTF8', self.serv.last_mail_options)
1438 self.assertEqual(self.serv.last_rcpt_options, [])
1440 def test_send_unicode_with_SMTPUTF8_via_low_level_API(self):
1443 HOST, self.port, local_hostname='localhost',
1445 self.addCleanup(smtp.close)
1447 self.assertEqual(
1450 self.assertEqual(smtp.rcpt('János'), (250, b'OK'))
1451 self.assertEqual(smtp.data(m), (250, b'OK'))
1452 self.assertEqual(self.serv.last_mailfrom, 'Jő')
1453 self.assertEqual(self.serv.last_rcpttos, ['János'])
1454 self.assertEqual(self.serv.last_message, m)
1455 self.assertIn('BODY=8BITMIME', self.serv.last_mail_options)
1456 self.assertIn('SMTPUTF8', self.serv.last_mail_options)
1457 self.assertEqual(self.serv.last_rcpt_options, [])
1459 def test_send_message_uses_smtputf8_if_addrs_non_ascii(self):
1480 HOST, self.port, local_hostname='localhost',
1482 self.addCleanup(smtp.close)
1483 self.assertEqual(smtp.send_message(msg), {})
1484 self.assertEqual(self.serv.last_mailfrom, 'főo@bar.com')
1485 self.assertEqual(self.serv.last_rcpttos, ['Dinsdale'])
1486 self.assertEqual(self.serv.last_message.decode(), expected)
1487 self.assertIn('BODY=8BITMIME', self.serv.last_mail_options)
1488 self.assertIn('SMTPUTF8', self.serv.last_mail_options)
1489 self.assertEqual(self.serv.last_rcpt_options, [])
1495 def smtp_AUTH(self, arg):
1505 self.push('235 Ok')
1507 self.push('571 Bad authentication')
1514 def setUp(self):
1515 self.thread_key = threading_helper.threading_setup()
1516 self.real_getfqdn = socket.getfqdn
1518 self.serv_evt = threading.Event()
1519 self.client_evt = threading.Event()
1521 self.serv = SimSMTPAUTHInitialResponseServer(
1524 self.port = self.serv.socket.getsockname()[1]
1525 serv_args = (self.serv, self.serv_evt, self.client_evt)
1526 self.thread = threading.Thread(target=debugging_server, args=serv_args)
1527 self.thread.start()
1530 self.serv_evt.wait()
1531 self.serv_evt.clear()
1533 def tearDown(self):
1534 socket.getfqdn = self.real_getfqdn
1536 self.client_evt.set()
1538 self.serv_evt.wait()
1539 threading_helper.join_thread(self.thread)
1540 del self.thread
1541 self.doCleanups()
1542 threading_helper.threading_cleanup(*self.thread_key)
1544 def testAUTH_PLAIN_initial_response_login(self):
1545 self.serv.add_feature('AUTH PLAIN')
1546 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1551 def testAUTH_PLAIN_initial_response_auth(self):
1552 self.serv.add_feature('AUTH PLAIN')
1553 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
1559 self.assertEqual(code, 235)