Lines Matching refs:self
18 def __init__(self, *args, **kwargs):
19 smtpd.SMTPServer.__init__(self, *args, **kwargs)
20 self.messages = []
21 if self._decode_data:
22 self.return_status = 'return status'
24 self.return_status = b'return status'
26 def process_message(self, peer, mailfrom, rcpttos, data, **kw):
27 self.messages.append((peer, mailfrom, rcpttos, data))
28 if data == self.return_status:
39 def listen(self, num):
44 def setUp(self):
47 def test_process_message_unimplemented(self):
61 self.assertRaises(NotImplementedError, write_line, b'spam\r\n.\r\n')
63 def test_decode_data_and_enable_SMTPUTF8_raises(self):
64 self.assertRaises(
72 def tearDown(self):
79 def setUp(self):
82 def send_data(self, channel, data, enable_SMTPUTF8=False):
96 def test_process_message_with_decode_data_true(self):
102 self.send_data(channel, b'From: test\n\nhello\n')
104 self.assertEqual(stdout, textwrap.dedent("""\
113 def test_process_message_with_decode_data_false(self):
118 self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
120 self.assertEqual(stdout, textwrap.dedent("""\
129 def test_process_message_with_enable_SMTPUTF8_true(self):
135 self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
137 self.assertEqual(stdout, textwrap.dedent("""\
146 def test_process_SMTPUTF8_message_with_enable_SMTPUTF8_true(self):
152 self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n',
155 self.assertEqual(stdout, textwrap.dedent("""\
165 def tearDown(self):
171 def setUp(self):
174 def tearDown(self):
179 def test_socket_uses_IPv6(self):
181 self.assertEqual(server.socket.family, socket.AF_INET6)
183 def test_socket_uses_IPv4(self):
185 self.assertEqual(server.socket.family, socket.AF_INET)
192 def setUp(self):
194 self.old_debugstream = smtpd.DEBUGSTREAM
195 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
197 def tearDown(self):
200 smtpd.DEBUGSTREAM = self.old_debugstream
202 def write_line(self, channel, line):
206 def test_params_rejected(self):
210 self.write_line(channel, b'EHLO example')
211 self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
212 self.write_line(channel, b'RCPT to: <foo@example.com> foo=bar')
213 self.assertEqual(channel.socket.last, self.error_response)
215 def test_nothing_accepted(self):
219 self.write_line(channel, b'EHLO example')
220 self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
221 self.write_line(channel, b'RCPT to: <foo@example.com>')
222 self.assertEqual(channel.socket.last, b'250 OK\r\n')
229 def setUp(self):
231 self.old_debugstream = smtpd.DEBUGSTREAM
232 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
234 def tearDown(self):
237 smtpd.DEBUGSTREAM = self.old_debugstream
239 def write_line(self, channel, line):
243 def test_with_decode_data_true(self):
247 self.write_line(channel, b'EHLO example')
254 self.write_line(channel, line)
255 self.assertEqual(channel.socket.last, self.error_response)
256 self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
257 self.assertEqual(channel.socket.last, b'250 OK\r\n')
259 def test_with_decode_data_false(self):
263 self.write_line(channel, b'EHLO example')
268 self.write_line(channel, line)
269 self.assertEqual(channel.socket.last, self.error_response)
270 self.write_line(
273 self.assertEqual(
276 self.write_line(
278 self.assertEqual(channel.socket.last, b'250 OK\r\n')
280 def test_with_enable_smtputf8_true(self):
284 self.write_line(channel, b'EHLO example')
285 self.write_line(
288 self.assertEqual(channel.socket.last, b'250 OK\r\n')
292 def setUp(self):
294 self.old_debugstream = smtpd.DEBUGSTREAM
295 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
296 self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
298 conn, addr = self.server.accept()
299 self.channel = smtpd.SMTPChannel(self.server, conn, addr,
302 def tearDown(self):
305 smtpd.DEBUGSTREAM = self.old_debugstream
307 def write_line(self, line):
308 self.channel.socket.queue_recv(line)
309 self.channel.handle_read()
311 def test_broken_connect(self):
312 self.assertRaises(
316 def test_decode_data_and_enable_SMTPUTF8_raises(self):
317 self.assertRaises(
319 self.server, self.channel.conn, self.channel.addr,
322 def test_server_accept(self):
323 self.server.handle_accept()
325 def test_missing_data(self):
326 self.write_line(b'')
327 self.assertEqual(self.channel.socket.last,
330 def test_EHLO(self):
331 self.write_line(b'EHLO example')
332 self.assertEqual(self.channel.socket.last, b'250 HELP\r\n')
334 def test_EHLO_bad_syntax(self):
335 self.write_line(b'EHLO')
336 self.assertEqual(self.channel.socket.last,
339 def test_EHLO_duplicate(self):
340 self.write_line(b'EHLO example')
341 self.write_line(b'EHLO example')
342 self.assertEqual(self.channel.socket.last,
345 def test_EHLO_HELO_duplicate(self):
346 self.write_line(b'EHLO example')
347 self.write_line(b'HELO example')
348 self.assertEqual(self.channel.socket.last,
351 def test_HELO(self):
353 self.write_line(b'HELO example')
354 self.assertEqual(self.channel.socket.last,
357 def test_HELO_EHLO_duplicate(self):
358 self.write_line(b'HELO example')
359 self.write_line(b'EHLO example')
360 self.assertEqual(self.channel.socket.last,
363 def test_HELP(self):
364 self.write_line(b'HELP')
365 self.assertEqual(self.channel.socket.last,
369 def test_HELP_command(self):
370 self.write_line(b'HELP MAIL')
371 self.assertEqual(self.channel.socket.last,
374 def test_HELP_command_unknown(self):
375 self.write_line(b'HELP SPAM')
376 self.assertEqual(self.channel.socket.last,
380 def test_HELO_bad_syntax(self):
381 self.write_line(b'HELO')
382 self.assertEqual(self.channel.socket.last,
385 def test_HELO_duplicate(self):
386 self.write_line(b'HELO example')
387 self.write_line(b'HELO example')
388 self.assertEqual(self.channel.socket.last,
391 def test_HELO_parameter_rejected_when_extensions_not_enabled(self):
392 self.extended_smtp = False
393 self.write_line(b'HELO example')
394 self.write_line(b'MAIL from:<foo@example.com> SIZE=1234')
395 self.assertEqual(self.channel.socket.last,
398 def test_MAIL_allows_space_after_colon(self):
399 self.write_line(b'HELO example')
400 self.write_line(b'MAIL from: <foo@example.com>')
401 self.assertEqual(self.channel.socket.last,
404 def test_extended_MAIL_allows_space_after_colon(self):
405 self.write_line(b'EHLO example')
406 self.write_line(b'MAIL from: <foo@example.com> size=20')
407 self.assertEqual(self.channel.socket.last,
410 def test_NOOP(self):
411 self.write_line(b'NOOP')
412 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
414 def test_HELO_NOOP(self):
415 self.write_line(b'HELO example')
416 self.write_line(b'NOOP')
417 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
419 def test_NOOP_bad_syntax(self):
420 self.write_line(b'NOOP hi')
421 self.assertEqual(self.channel.socket.last,
424 def test_QUIT(self):
425 self.write_line(b'QUIT')
426 self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
428 def test_HELO_QUIT(self):
429 self.write_line(b'HELO example')
430 self.write_line(b'QUIT')
431 self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
433 def test_QUIT_arg_ignored(self):
434 self.write_line(b'QUIT bye bye')
435 self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
437 def test_bad_state(self):
438 self.channel.smtp_state = 'BAD STATE'
439 self.write_line(b'HELO example')
440 self.assertEqual(self.channel.socket.last,
443 def test_command_too_long(self):
444 self.write_line(b'HELO example')
445 self.write_line(b'MAIL from: ' +
446 b'a' * self.channel.command_size_limit +
448 self.assertEqual(self.channel.socket.last,
451 def test_MAIL_command_limit_extended_with_SIZE(self):
452 self.write_line(b'EHLO example')
453 fill_len = self.channel.command_size_limit - len('MAIL from:<@example>')
454 self.write_line(b'MAIL from:<' +
457 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
459 self.write_line(b'MAIL from:<' +
462 self.assertEqual(self.channel.socket.last,
465 def test_MAIL_command_rejects_SMTPUTF8_by_default(self):
466 self.write_line(b'EHLO example')
467 self.write_line(
469 self.assertEqual(self.channel.socket.last[0:1], b'5')
471 def test_data_longer_than_default_data_size_limit(self):
473 self.channel.data_size_limit = 1048
474 self.write_line(b'HELO example')
475 self.write_line(b'MAIL From:eggs@example')
476 self.write_line(b'RCPT To:spam@example')
477 self.write_line(b'DATA')
478 self.write_line(b'A' * self.channel.data_size_limit +
480 self.assertEqual(self.channel.socket.last,
483 def test_MAIL_size_parameter(self):
484 self.write_line(b'EHLO example')
485 self.write_line(b'MAIL FROM:<eggs@example> SIZE=512')
486 self.assertEqual(self.channel.socket.last,
489 def test_MAIL_invalid_size_parameter(self):
490 self.write_line(b'EHLO example')
491 self.write_line(b'MAIL FROM:<eggs@example> SIZE=invalid')
492 self.assertEqual(self.channel.socket.last,
495 def test_MAIL_RCPT_unknown_parameters(self):
496 self.write_line(b'EHLO example')
497 self.write_line(b'MAIL FROM:<eggs@example> ham=green')
498 self.assertEqual(self.channel.socket.last,
501 self.write_line(b'MAIL FROM:<eggs@example>')
502 self.write_line(b'RCPT TO:<eggs@example> ham=green')
503 self.assertEqual(self.channel.socket.last,
506 def test_MAIL_size_parameter_larger_than_default_data_size_limit(self):
507 self.channel.data_size_limit = 1048
508 self.write_line(b'EHLO example')
509 self.write_line(b'MAIL FROM:<eggs@example> SIZE=2096')
510 self.assertEqual(self.channel.socket.last,
513 def test_need_MAIL(self):
514 self.write_line(b'HELO example')
515 self.write_line(b'RCPT to:spam@example')
516 self.assertEqual(self.channel.socket.last,
519 def test_MAIL_syntax_HELO(self):
520 self.write_line(b'HELO example')
521 self.write_line(b'MAIL from eggs@example')
522 self.assertEqual(self.channel.socket.last,
525 def test_MAIL_syntax_EHLO(self):
526 self.write_line(b'EHLO example')
527 self.write_line(b'MAIL from eggs@example')
528 self.assertEqual(self.channel.socket.last,
531 def test_MAIL_missing_address(self):
532 self.write_line(b'HELO example')
533 self.write_line(b'MAIL from:')
534 self.assertEqual(self.channel.socket.last,
537 def test_MAIL_chevrons(self):
538 self.write_line(b'HELO example')
539 self.write_line(b'MAIL from:<eggs@example>')
540 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
542 def test_MAIL_empty_chevrons(self):
543 self.write_line(b'EHLO example')
544 self.write_line(b'MAIL from:<>')
545 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
547 def test_MAIL_quoted_localpart(self):
548 self.write_line(b'EHLO example')
549 self.write_line(b'MAIL from: <"Fred Blogs"@example.com>')
550 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
551 self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
553 def test_MAIL_quoted_localpart_no_angles(self):
554 self.write_line(b'EHLO example')
555 self.write_line(b'MAIL from: "Fred Blogs"@example.com')
556 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
557 self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
559 def test_MAIL_quoted_localpart_with_size(self):
560 self.write_line(b'EHLO example')
561 self.write_line(b'MAIL from: <"Fred Blogs"@example.com> SIZE=1000')
562 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
563 self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
565 def test_MAIL_quoted_localpart_with_size_no_angles(self):
566 self.write_line(b'EHLO example')
567 self.write_line(b'MAIL from: "Fred Blogs"@example.com SIZE=1000')
568 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
569 self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
571 def test_nested_MAIL(self):
572 self.write_line(b'HELO example')
573 self.write_line(b'MAIL from:eggs@example')
574 self.write_line(b'MAIL from:spam@example')
575 self.assertEqual(self.channel.socket.last,
578 def test_VRFY(self):
579 self.write_line(b'VRFY eggs@example')
580 self.assertEqual(self.channel.socket.last,
584 def test_VRFY_syntax(self):
585 self.write_line(b'VRFY')
586 self.assertEqual(self.channel.socket.last,
589 def test_EXPN_not_implemented(self):
590 self.write_line(b'EXPN')
591 self.assertEqual(self.channel.socket.last,
594 def test_no_HELO_MAIL(self):
595 self.write_line(b'MAIL from:<foo@example.com>')
596 self.assertEqual(self.channel.socket.last,
599 def test_need_RCPT(self):
600 self.write_line(b'HELO example')
601 self.write_line(b'MAIL From:eggs@example')
602 self.write_line(b'DATA')
603 self.assertEqual(self.channel.socket.last,
606 def test_RCPT_syntax_HELO(self):
607 self.write_line(b'HELO example')
608 self.write_line(b'MAIL From: eggs@example')
609 self.write_line(b'RCPT to eggs@example')
610 self.assertEqual(self.channel.socket.last,
613 def test_RCPT_syntax_EHLO(self):
614 self.write_line(b'EHLO example')
615 self.write_line(b'MAIL From: eggs@example')
616 self.write_line(b'RCPT to eggs@example')
617 self.assertEqual(self.channel.socket.last,
620 def test_RCPT_lowercase_to_OK(self):
621 self.write_line(b'HELO example')
622 self.write_line(b'MAIL From: eggs@example')
623 self.write_line(b'RCPT to: <eggs@example>')
624 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
626 def test_no_HELO_RCPT(self):
627 self.write_line(b'RCPT to eggs@example')
628 self.assertEqual(self.channel.socket.last,
631 def test_data_dialog(self):
632 self.write_line(b'HELO example')
633 self.write_line(b'MAIL From:eggs@example')
634 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
635 self.write_line(b'RCPT To:spam@example')
636 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
638 self.write_line(b'DATA')
639 self.assertEqual(self.channel.socket.last,
641 self.write_line(b'data\r\nmore\r\n.')
642 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
643 self.assertEqual(self.server.messages,
649 def test_DATA_syntax(self):
650 self.write_line(b'HELO example')
651 self.write_line(b'MAIL From:eggs@example')
652 self.write_line(b'RCPT To:spam@example')
653 self.write_line(b'DATA spam')
654 self.assertEqual(self.channel.socket.last, b'501 Syntax: DATA\r\n')
656 def test_no_HELO_DATA(self):
657 self.write_line(b'DATA spam')
658 self.assertEqual(self.channel.socket.last,
661 def test_data_transparency_section_4_5_2(self):
662 self.write_line(b'HELO example')
663 self.write_line(b'MAIL From:eggs@example')
664 self.write_line(b'RCPT To:spam@example')
665 self.write_line(b'DATA')
666 self.write_line(b'..\r\n.\r\n')
667 self.assertEqual(self.channel.received_data, '.')
669 def test_multiple_RCPT(self):
670 self.write_line(b'HELO example')
671 self.write_line(b'MAIL From:eggs@example')
672 self.write_line(b'RCPT To:spam@example')
673 self.write_line(b'RCPT To:ham@example')
674 self.write_line(b'DATA')
675 self.write_line(b'data\r\n.')
676 self.assertEqual(self.server.messages,
682 def test_manual_status(self):
684 self.write_line(b'HELO example')
685 self.write_line(b'MAIL From:eggs@example')
686 self.write_line(b'RCPT To:spam@example')
687 self.write_line(b'DATA')
688 self.write_line(b'return status\r\n.')
689 self.assertEqual(self.channel.socket.last, b'250 Okish\r\n')
691 def test_RSET(self):
692 self.write_line(b'HELO example')
693 self.write_line(b'MAIL From:eggs@example')
694 self.write_line(b'RCPT To:spam@example')
695 self.write_line(b'RSET')
696 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
697 self.write_line(b'MAIL From:foo@example')
698 self.write_line(b'RCPT To:eggs@example')
699 self.write_line(b'DATA')
700 self.write_line(b'data\r\n.')
701 self.assertEqual(self.server.messages,
707 def test_HELO_RSET(self):
708 self.write_line(b'HELO example')
709 self.write_line(b'RSET')
710 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
712 def test_RSET_syntax(self):
713 self.write_line(b'RSET hi')
714 self.assertEqual(self.channel.socket.last, b'501 Syntax: RSET\r\n')
716 def test_unknown_command(self):
717 self.write_line(b'UNKNOWN_CMD')
718 self.assertEqual(self.channel.socket.last,
722 def test_attribute_deprecations(self):
724 spam = self.channel._SMTPChannel__server
726 self.channel._SMTPChannel__server = 'spam'
728 spam = self.channel._SMTPChannel__line
730 self.channel._SMTPChannel__line = 'spam'
732 spam = self.channel._SMTPChannel__state
734 self.channel._SMTPChannel__state = 'spam'
736 spam = self.channel._SMTPChannel__greeting
738 self.channel._SMTPChannel__greeting = 'spam'
740 spam = self.channel._SMTPChannel__mailfrom
742 self.channel._SMTPChannel__mailfrom = 'spam'
744 spam = self.channel._SMTPChannel__rcpttos
746 self.channel._SMTPChannel__rcpttos = 'spam'
748 spam = self.channel._SMTPChannel__data
750 self.channel._SMTPChannel__data = 'spam'
752 spam = self.channel._SMTPChannel__fqdn
754 self.channel._SMTPChannel__fqdn = 'spam'
756 spam = self.channel._SMTPChannel__peer
758 self.channel._SMTPChannel__peer = 'spam'
760 spam = self.channel._SMTPChannel__conn
762 self.channel._SMTPChannel__conn = 'spam'
764 spam = self.channel._SMTPChannel__addr
766 self.channel._SMTPChannel__addr = 'spam'
770 def setUp(self):
772 self.old_debugstream = smtpd.DEBUGSTREAM
773 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
774 self.server = DummyServer((socket_helper.HOSTv6, 0), ('b', 0),
776 conn, addr = self.server.accept()
777 self.channel = smtpd.SMTPChannel(self.server, conn, addr,
782 def setUp(self):
784 self.old_debugstream = smtpd.DEBUGSTREAM
785 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
786 self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
788 conn, addr = self.server.accept()
790 self.channel = smtpd.SMTPChannel(self.server, conn, addr, 32,
793 def tearDown(self):
796 smtpd.DEBUGSTREAM = self.old_debugstream
798 def write_line(self, line):
799 self.channel.socket.queue_recv(line)
800 self.channel.handle_read()
802 def test_data_limit_dialog(self):
803 self.write_line(b'HELO example')
804 self.write_line(b'MAIL From:eggs@example')
805 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
806 self.write_line(b'RCPT To:spam@example')
807 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
809 self.write_line(b'DATA')
810 self.assertEqual(self.channel.socket.last,
812 self.write_line(b'data\r\nmore\r\n.')
813 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
814 self.assertEqual(self.server.messages,
820 def test_data_limit_dialog_too_much_data(self):
821 self.write_line(b'HELO example')
822 self.write_line(b'MAIL From:eggs@example')
823 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
824 self.write_line(b'RCPT To:spam@example')
825 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
827 self.write_line(b'DATA')
828 self.assertEqual(self.channel.socket.last,
830 self.write_line(b'This message is longer than 32 bytes\r\n.')
831 self.assertEqual(self.channel.socket.last,
837 def setUp(self):
839 self.old_debugstream = smtpd.DEBUGSTREAM
840 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
841 self.server = DummyServer((socket_helper.HOST, 0), ('b', 0))
842 conn, addr = self.server.accept()
843 self.channel = smtpd.SMTPChannel(self.server, conn, addr)
845 def tearDown(self):
848 smtpd.DEBUGSTREAM = self.old_debugstream
850 def write_line(self, line):
851 self.channel.socket.queue_recv(line)
852 self.channel.handle_read()
854 def test_ascii_data(self):
855 self.write_line(b'HELO example')
856 self.write_line(b'MAIL From:eggs@example')
857 self.write_line(b'RCPT To:spam@example')
858 self.write_line(b'DATA')
859 self.write_line(b'plain ascii text')
860 self.write_line(b'.')
861 self.assertEqual(self.channel.received_data, b'plain ascii text')
863 def test_utf8_data(self):
864 self.write_line(b'HELO example')
865 self.write_line(b'MAIL From:eggs@example')
866 self.write_line(b'RCPT To:spam@example')
867 self.write_line(b'DATA')
868 self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
869 self.write_line(b'and some plain ascii')
870 self.write_line(b'.')
871 self.assertEqual(
872 self.channel.received_data,
879 def setUp(self):
881 self.old_debugstream = smtpd.DEBUGSTREAM
882 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
883 self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
885 conn, addr = self.server.accept()
887 self.channel = smtpd.SMTPChannel(self.server, conn, addr,
890 def tearDown(self):
893 smtpd.DEBUGSTREAM = self.old_debugstream
895 def write_line(self, line):
896 self.channel.socket.queue_recv(line)
897 self.channel.handle_read()
899 def test_ascii_data(self):
900 self.write_line(b'HELO example')
901 self.write_line(b'MAIL From:eggs@example')
902 self.write_line(b'RCPT To:spam@example')
903 self.write_line(b'DATA')
904 self.write_line(b'plain ascii text')
905 self.write_line(b'.')
906 self.assertEqual(self.channel.received_data, 'plain ascii text')
908 def test_utf8_data(self):
909 self.write_line(b'HELO example')
910 self.write_line(b'MAIL From:eggs@example')
911 self.write_line(b'RCPT To:spam@example')
912 self.write_line(b'DATA')
913 self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
914 self.write_line(b'and some plain ascii')
915 self.write_line(b'.')
916 self.assertEqual(
917 self.channel.received_data,
922 def setUp(self):
924 self.old_debugstream = smtpd.DEBUGSTREAM
925 self.debug = smtpd.DEBUGSTREAM = io.StringIO()
926 self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
928 conn, addr = self.server.accept()
929 self.channel = smtpd.SMTPChannel(self.server, conn, addr,
932 def tearDown(self):
935 smtpd.DEBUGSTREAM = self.old_debugstream
937 def write_line(self, line):
938 self.channel.socket.queue_recv(line)
939 self.channel.handle_read()
941 def test_MAIL_command_accepts_SMTPUTF8_when_announced(self):
942 self.write_line(b'EHLO example')
943 self.write_line(
947 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
949 def test_process_smtputf8_message(self):
950 self.write_line(b'EHLO example')
952 self.write_line(b'MAIL from: <a@example> ' + mail_parameters)
953 self.assertEqual(self.channel.socket.last[0:3], b'250')
954 self.write_line(b'rcpt to:<b@example.com>')
955 self.assertEqual(self.channel.socket.last[0:3], b'250')
956 self.write_line(b'data')
957 self.assertEqual(self.channel.socket.last[0:3], b'354')
958 self.write_line(b'c\r\n.')
960 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
962 self.assertEqual(self.channel.socket.last,
965 def test_utf8_data(self):
966 self.write_line(b'EHLO example')
967 self.write_line(
969 self.assertEqual(self.channel.socket.last[0:3], b'250')
970 self.write_line('RCPT To:späm@examplé'.encode('utf-8'))
971 self.assertEqual(self.channel.socket.last[0:3], b'250')
972 self.write_line(b'DATA')
973 self.assertEqual(self.channel.socket.last[0:3], b'354')
974 self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
975 self.write_line(b'.')
976 self.assertEqual(
977 self.channel.received_data,
980 def test_MAIL_command_limit_extended_with_SIZE_and_SMTPUTF8(self):
981 self.write_line(b'ehlo example')
983 self.write_line(b'MAIL from:<' +
986 self.assertEqual(self.channel.socket.last,
988 self.write_line(b'MAIL from:<' +
991 self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
993 def test_multiple_emails_with_extended_command_length(self):
994 self.write_line(b'ehlo example')
997 self.write_line(b'MAIL from:<' + char * fill_len + b'a@example>')
998 self.assertEqual(self.channel.socket.last[0:3], b'500')
999 self.write_line(b'MAIL from:<' + char * fill_len + b'@example>')
1000 self.assertEqual(self.channel.socket.last[0:3], b'250')
1001 self.write_line(b'rcpt to:<hans@example.com>')
1002 self.assertEqual(self.channel.socket.last[0:3], b'250')
1003 self.write_line(b'data')
1004 self.assertEqual(self.channel.socket.last[0:3], b'354')
1005 self.write_line(b'test\r\n.')
1006 self.assertEqual(self.channel.socket.last[0:3], b'250')
1010 def test__all__(self):
1015 support.check__all__(self, smtpd, not_exported=not_exported)