xxx better would be to actually detect the duplicate with self raisepointerrors startboundarynotfounddefect msg self strmsgsource if self raiseexpected return inner msg getpayload0 self asserttruehasattrinner defects self assertequallenself getdefectsinner 1 self assertisinstanceself getdefectsinner0 errors startboundarynotfounddefect def testmultipartnoboundaryself source textwrap dedent date fri 6 apr 2001 09 23 06 0800 gmt0800 from foobar subject broken mail mimeversion 1 0 contenttype multipartreport reporttypedeliverystatus jab03225 986577786zinfandel lacita com one part jab03225 986577786zinfandel lacita com contenttype messagedeliverystatus header another part jab03225 986577786zinfandel lacita com def testmultipartinvalidcteself with self raisepoint errors invalidmultipartcontenttransferencodingdefect msg self strmsg self multipartmsg format ncontenttransferencoding base64 if self raiseexpected return self assertequallenself getdefectsmsg 1 self assertisinstanceself getdefectsmsg0 errors invalidmultipartcontenttransferencodingdefect def testmultipartnoctenodefectself if self raiseexpected return msg self strmsgself multipartmsg format self assertequallenself getdefectsmsg 0 def testmultipartvalidctenodefectself if self raiseexpected return for cte in 7bit 8bit binary msg self strmsg self multipartmsg formatncontenttransferencoding cte self assertequallenself getdefectsmsg 0 ctecte def testlyingmultipartself source textwrap dedent from allison dunlap xxxexample com to yyyexample com subject 64423 date sun 11 jul 2004 16 09 27 0300 mimeversion 1 0 contenttype multipartalternative blah blah blah the message structure is multipartmixed textplain messagerfc822 multipartmixed this message is missing its start boundary with self raisepointerrors startboundarynotfounddefect outer self strmsgsource if self raiseexpected return bad outer getpayload1 getpayload0 self assertequallenself getdefectsbad 1 self assertisinstanceself getdefectsbad0 errors startboundarynotfounddefect def testfirstlineiscontinuationheaderself with self raisepointerrors firstheaderlineiscontinuationdefect msg self strmsg line 1nsubject testnnbody if self raiseexpected return self assertequalmsg keys subject self assertequalmsg getpayload body self assertequallenself getdefectsmsg 1 self assertdefectsequalself getdefectsmsg errors firstheaderlineiscontinuationdefect self assertequalself getdefectsmsg0 line line 1n def testmissingheaderbodyseparatorself our heuristic if we see a line that doesn t look like a header no leading whitespace but no is to assume that the blank line that separates the header from the body is missing and to stop parsing headers and start parsing the body with self raisepointerrors missingheaderbodyseparatordefect msg self strmsg subject testnnot a headernto abcnnbn if self raiseexpected return self assertequalmsg keys subject self assertequalmsg getpayload not a headernto abcnnbn self assertdefectsequalself getdefectsmsg errors missingheaderbodyseparatordefect def testbadpaddinginbase64payloadself source textwrap dedent subject test mimeversion 1 0 contenttype textplain charsetutf8 contenttransferencoding base64 dmk msg self strmsgsource with self raisepointerrors invalidbase64charactersdefect payload msg getpayloaddecodetrue if self raiseexpected return self assertequalpayload b vi self assertdefectsequalself getdefectsmsg errors invalidbase64charactersdefect def testinvalidlengthofbase64payloadself source textwrap dedent subject test mimeversion 1 0 contenttype textplain charsetutf8 contenttransferencoding base64 abcde with self raisepointerrors closeboundarynotfounddefect msg self strmsgsource if self raiseexpected return self assertequallenmsg getpayload 2 self assertequalmsg getpayload1 getpayload alternative 2n self assertdefectsequalself getdefectsmsg errors closeboundarynotfounddefect class testdefectdetectiontestdefectsbase testemailbase def getdefectsself obj return obj defects class testdefectcapturetestdefectsbase testemailbase class capturepolicypolicy emailpolicy captured none def registerdefectself obj defect self captured appenddefect def setupself self policy self capturepolicycapturedlist def getdefectsself obj return self policy captured class testdefectraisingtestdefectsbase testemailbase policy testdefectsbase policy policy policy cloneraiseondefecttrue raiseexpected true contextlib contextmanager def raisepointself defect with self assertraisesdefect yield if name main unittest main subject xx from xx xx dk to xx mime version 1 0 content type multipart mixed boundary ms_mac_oe_3071477847_720252_mime_part ms_mac_oe_3071477847_720252_mime_part content type multipart alternative boundary ms_mac_oe_3071477847_720252_mime_part ms_mac_oe_3071477847_720252_mime_part content type text plain charset iso 8859 1 content transfer encoding quoted printable text ms_mac_oe_3071477847_720252_mime_part content type text html charset iso 8859 1 content transfer encoding quoted printable html html ms_mac_oe_3071477847_720252_mime_part ms_mac_oe_3071477847_720252_mime_part content type image gif name xx gif content disposition attachment content transfer encoding base64 some removed base64 encoded chars ms_mac_oe_3071477847_720252_mime_part xxx better would be to actually detect the duplicate date fri 6 apr 2001 09 23 06 0800 gmt 0800 from foobar subject broken mail mime version 1 0 content type multipart report report type delivery status jab03225 986577786 zinfandel lacita com one part jab03225 986577786 zinfandel lacita com content type message delivery status header another part jab03225 986577786 zinfandel lacita com date wed 14 nov 2007 12 56 23 gmt from foo bar invalid to foo bar invalid subject content transfer encoding base64 and multipart mime version 1 0 content type multipart mixed boundary 3344438784458119861 3344438784458119861 content type text plain test message 3344438784458119861 content type application octet stream content transfer encoding base64 ywjj 3344438784458119861 from allison dunlap xxx example com to yyy example com subject 64423 date sun 11 jul 2004 16 09 27 0300 mime version 1 0 content type multipart alternative blah blah blah content type multipart mixed boundary aaa from mail delivery subsystem xxx example com to yyy example com aaa stuff aaa content type message rfc822 from webmaster python org to zzz example com content type multipart mixed boundary bbb bbb aaa the message structure is multipart mixed text plain message rfc822 multipart mixed this message is missing its start boundary our heuristic if we see a line that doesn t look like a header no leading whitespace but no is to assume that the blank line that separates the header from the body is missing and to stop parsing headers and start parsing the body subject test mime version 1 0 content type text plain charset utf 8 content transfer encoding base64 dmk subject test mime version 1 0 content type text plain charset utf 8 content transfer encoding base64 dm x01k subject test mime version 1 0 content type text plain charset utf 8 content transfer encoding base64 abcde to 1 harrydomain4 com subject fwd 1 mime version 1 0 content type multipart alternative boundary 000101020201080900040301 000101020201080900040301 content type text plain charset iso 8859 1 content transfer encoding 7bit alternative 1 000101020201080900040301 content type text html charset iso 8859 1 content transfer encoding 7bit alternative 2
import textwrap
import unittest
import contextlib
from email import policy
from email import errors
from test.test_email import TestEmailBase


class TestDefectsBase:

    policy = policy.default
    raise_expected = False

    @contextlib.contextmanager
    def _raise_point(self, defect):
        yield

    def test_same_boundary_inner_outer(self):
        source = textwrap.dedent()
        
        with self._raise_point(errors.StartBoundaryNotFoundDefect):
            msg = self._str_msg(source)
        if self.raise_expected: return
        inner = msg.get_payload(0)
        self.assertTrue(hasattr(inner, 'defects'))
        self.assertEqual(len(self.get_defects(inner)), 1)
        self.assertIsInstance(self.get_defects(inner)[0],
                              errors.StartBoundaryNotFoundDefect)

    def test_multipart_no_boundary(self):
        source = textwrap.dedent()
        with self._raise_point(errors.NoBoundaryInMultipartDefect):
            msg = self._str_msg(source)
        if self.raise_expected: return
        self.assertIsInstance(msg.get_payload(), str)
        self.assertEqual(len(self.get_defects(msg)), 2)
        self.assertIsInstance(self.get_defects(msg)[0],
                              errors.NoBoundaryInMultipartDefect)
        self.assertIsInstance(self.get_defects(msg)[1],
                              errors.MultipartInvariantViolationDefect)

    multipart_msg = textwrap.dedent()

    def test_multipart_invalid_cte(self):
        with self._raise_point(
                errors.InvalidMultipartContentTransferEncodingDefect):
            msg = self._str_msg(
                    self.multipart_msg.format(
                        "\nContent-Transfer-Encoding: base64"))
        if self.raise_expected: return
        self.assertEqual(len(self.get_defects(msg)), 1)
        self.assertIsInstance(self.get_defects(msg)[0],
            errors.InvalidMultipartContentTransferEncodingDefect)

    def test_multipart_no_cte_no_defect(self):
        if self.raise_expected: return
        msg = self._str_msg(self.multipart_msg.format(''))
        self.assertEqual(len(self.get_defects(msg)), 0)

    def test_multipart_valid_cte_no_defect(self):
        if self.raise_expected: return
        for cte in ('7bit', '8bit', 'BINary'):
            msg = self._str_msg(
                self.multipart_msg.format("\nContent-Transfer-Encoding: "+cte))
            self.assertEqual(len(self.get_defects(msg)), 0, "cte="+cte)

    def test_lying_multipart(self):
        source = textwrap.dedent()
        with self._raise_point(errors.NoBoundaryInMultipartDefect):
            msg = self._str_msg(source)
        if self.raise_expected: return
        self.assertTrue(hasattr(msg, 'defects'))
        self.assertEqual(len(self.get_defects(msg)), 2)
        self.assertIsInstance(self.get_defects(msg)[0],
                              errors.NoBoundaryInMultipartDefect)
        self.assertIsInstance(self.get_defects(msg)[1],
                              errors.MultipartInvariantViolationDefect)

    def test_missing_start_boundary(self):
        source = textwrap.dedent()
        
        
        
        
        
        
        
        
        with self._raise_point(errors.StartBoundaryNotFoundDefect):
            outer = self._str_msg(source)
        if self.raise_expected: return
        bad = outer.get_payload(1).get_payload(0)
        self.assertEqual(len(self.get_defects(bad)), 1)
        self.assertIsInstance(self.get_defects(bad)[0],
                              errors.StartBoundaryNotFoundDefect)

    def test_first_line_is_continuation_header(self):
        with self._raise_point(errors.FirstHeaderLineIsContinuationDefect):
            msg = self._str_msg(' Line 1\nSubject: test\n\nbody')
        if self.raise_expected: return
        self.assertEqual(msg.keys(), ['Subject'])
        self.assertEqual(msg.get_payload(), 'body')
        self.assertEqual(len(self.get_defects(msg)), 1)
        self.assertDefectsEqual(self.get_defects(msg),
                                 [errors.FirstHeaderLineIsContinuationDefect])
        self.assertEqual(self.get_defects(msg)[0].line, ' Line 1\n')

    def test_missing_header_body_separator(self):
        
        
        
        
        with self._raise_point(errors.MissingHeaderBodySeparatorDefect):
            msg = self._str_msg('Subject: test\nnot a header\nTo: abc\n\nb\n')
        if self.raise_expected: return
        self.assertEqual(msg.keys(), ['Subject'])
        self.assertEqual(msg.get_payload(), 'not a header\nTo: abc\n\nb\n')
        self.assertDefectsEqual(self.get_defects(msg),
                                [errors.MissingHeaderBodySeparatorDefect])

    def test_bad_padding_in_base64_payload(self):
        source = textwrap.dedent()
        msg = self._str_msg(source)
        with self._raise_point(errors.InvalidBase64PaddingDefect):
            payload = msg.get_payload(decode=True)
        if self.raise_expected: return
        self.assertEqual(payload, b'vi')
        self.assertDefectsEqual(self.get_defects(msg),
                                [errors.InvalidBase64PaddingDefect])

    def test_invalid_chars_in_base64_payload(self):
        source = textwrap.dedent()
        msg = self._str_msg(source)
        with self._raise_point(errors.InvalidBase64CharactersDefect):
            payload = msg.get_payload(decode=True)
        if self.raise_expected: return
        self.assertEqual(payload, b'vi')
        self.assertDefectsEqual(self.get_defects(msg),
                                [errors.InvalidBase64CharactersDefect])

    def test_invalid_length_of_base64_payload(self):
        source = textwrap.dedent()
        msg = self._str_msg(source)
        with self._raise_point(errors.InvalidBase64LengthDefect):
            payload = msg.get_payload(decode=True)
        if self.raise_expected: return
        self.assertEqual(payload, b'abcde')
        self.assertDefectsEqual(self.get_defects(msg),
                                [errors.InvalidBase64LengthDefect])

    def test_missing_ending_boundary(self):
        source = textwrap.dedent()
        with self._raise_point(errors.CloseBoundaryNotFoundDefect):
            msg = self._str_msg(source)
        if self.raise_expected: return
        self.assertEqual(len(msg.get_payload()), 2)
        self.assertEqual(msg.get_payload(1).get_payload(), 'Alternative 2\n')
        self.assertDefectsEqual(self.get_defects(msg),
                                [errors.CloseBoundaryNotFoundDefect])


class TestDefectDetection(TestDefectsBase, TestEmailBase):

    def get_defects(self, obj):
        return obj.defects


class TestDefectCapture(TestDefectsBase, TestEmailBase):

    class CapturePolicy(policy.EmailPolicy):
        captured = None
        def register_defect(self, obj, defect):
            self.captured.append(defect)

    def setUp(self):
        self.policy = self.CapturePolicy(captured=list())

    def get_defects(self, obj):
        return self.policy.captured


class TestDefectRaising(TestDefectsBase, TestEmailBase):

    policy = TestDefectsBase.policy
    policy = policy.clone(raise_on_defect=True)
    raise_expected = True

    @contextlib.contextmanager
    def _raise_point(self, defect):
        with self.assertRaises(defect):
            yield


if __name__ == '__main__':
    unittest.main()