Enhance Policy framework in preparation for adding "eamil6" policy as provisional

Lib/test/test_email/test_email.py

 # Copyright (C) 2001-2010 Python Software Foundation
 # Contact: email-sig@python.org
 # email package unit tests
 
 import os
 import re
 import sys
 import time
 import base64
 import difflib
 import unittest
 import warnings
 import textwrap
 
 from io import StringIO, BytesIO
 from itertools import chain
 
 import email
+import email.policy
 
 from email.charset import Charset
 from email.header import Header, decode_header, make_header
 from email.parser import Parser, HeaderParser
 from email.generator import Generator, DecodedGenerator
 from email.message import Message
 from email.mime.application import MIMEApplication
 from email.mime.audio import MIMEAudio
 from email.mime.text import MIMEText
 from email.mime.image import MIMEImage
@@ skipping 1769 matching lines @@
 MIME-Version: 1.0
 Content-Transfer-Encoding: base64
 
 YXNkZg==
 --===============0012394164==--""")
         self.assertEqual(m.get_payload(0).get_payload(), 'YXNkZg==')
 
 
 
 # Test some badly formatted messages
-class TestNonConformantBase:
-
-    def _msgobj(self, filename):
-        with openfile(filename) as fp:
-            return email.message_from_file(fp, policy=self.policy)
+class TestNonConformant(TestEmailBase):
 
     def test_parse_missing_minor_type(self):
         eq = self.assertEqual
         msg = self._msgobj('msg_14.txt')
         eq(msg.get_content_type(), 'text/plain')
         eq(msg.get_content_maintype(), 'text')
         eq(msg.get_content_subtype(), 'plain')
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_same_boundary_inner_outer(self):
         unless = self.assertTrue
         msg = self._msgobj('msg_15.txt')
         # XXX We can probably eventually do better
         inner = msg.get_payload(0)
         unless(hasattr(inner, 'defects'))
-        self.assertEqual(len(self.get_defects(inner)), 1)
-        unless(isinstance(self.get_defects(inner)[0],
+        self.assertEqual(len(inner.defects), 1)
+        unless(isinstance(inner.defects[0],
                           errors.StartBoundaryNotFoundDefect))
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_multipart_no_boundary(self):
         unless = self.assertTrue
         msg = self._msgobj('msg_25.txt')
         unless(isinstance(msg.get_payload(), str))
-        self.assertEqual(len(self.get_defects(msg)), 2)
-        unless(isinstance(self.get_defects(msg)[0],
+        self.assertEqual(len(msg.defects), 2)
+        unless(isinstance(msg.defects[0],
                           errors.NoBoundaryInMultipartDefect))
-        unless(isinstance(self.get_defects(msg)[1],
+        unless(isinstance(msg.defects[1],
                           errors.MultipartInvariantViolationDefect))
 
     multipart_msg = textwrap.dedent("""\
         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==--
         """)
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_multipart_invalid_cte(self):
-        msg = email.message_from_string(
-            self.multipart_msg.format("\nContent-Transfer-Encoding: base64"),
-            policy = self.policy)
-        self.assertEqual(len(self.get_defects(msg)), 1)
-        self.assertIsInstance(self.get_defects(msg)[0],
+        msg = self._str_msg(
+            self.multipart_msg.format("\nContent-Transfer-Encoding: base64"))
+        self.assertEqual(len(msg.defects), 1)
+        self.assertIsInstance(msg.defects[0],
             errors.InvalidMultipartContentTransferEncodingDefect)
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_multipart_no_cte_no_defect(self):
-        msg = email.message_from_string(
-            self.multipart_msg.format(''),
-            policy = self.policy)
-        self.assertEqual(len(self.get_defects(msg)), 0)
+        msg = self._str_msg(self.multipart_msg.format(''))
+        self.assertEqual(len(msg.defects), 0)
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_multipart_valid_cte_no_defect(self):
         for cte in ('7bit', '8bit', 'BINary'):
-            msg = email.message_from_string(
+            msg = self._str_msg(
                 self.multipart_msg.format(
-                    "\nContent-Transfer-Encoding: {}".format(cte)),
-                policy = self.policy)
-            self.assertEqual(len(self.get_defects(msg)), 0)
+                    "\nContent-Transfer-Encoding: {}".format(cte)))
+            self.assertEqual(len(msg.defects), 0)
 
     def test_invalid_content_type(self):
         eq = self.assertEqual
         neq = self.ndiffAssertEqual
         msg = Message()
         # RFC 2045, $5.2 says invalid yields text/plain
         msg['Content-Type'] = 'text'
         eq(msg.get_content_maintype(), 'text')
         eq(msg.get_content_subtype(), 'plain')
         eq(msg.get_content_type(), 'text/plain')
@@ skipping 30 matching lines @@
         eq = self.ndiffAssertEqual
         msg = self._msgobj('msg_35.txt')
         eq(msg.as_string(), """\
 From: aperson@dom.ain
 To: bperson@dom.ain
 Subject: here's something interesting
 
 counter to RFC 2822, there's no separating newline here
 """)
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_lying_multipart(self):
         unless = self.assertTrue
         msg = self._msgobj('msg_41.txt')
         unless(hasattr(msg, 'defects'))
-        self.assertEqual(len(self.get_defects(msg)), 2)
-        unless(isinstance(self.get_defects(msg)[0],
+        self.assertEqual(len(msg.defects), 2)
+        unless(isinstance(msg.defects[0],
                           errors.NoBoundaryInMultipartDefect))
-        unless(isinstance(self.get_defects(msg)[1],
+        unless(isinstance(msg.defects[1],
                           errors.MultipartInvariantViolationDefect))
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_missing_start_boundary(self):
         outer = self._msgobj('msg_42.txt')
         # The message structure is:
         #
         # multipart/mixed
         #    text/plain
         #    message/rfc822
         #        multipart/mixed [*]
         #
         # [*] This message is missing its start boundary
         bad = outer.get_payload(1).get_payload(0)
-        self.assertEqual(len(self.get_defects(bad)), 1)
-        self.assertTrue(isinstance(self.get_defects(bad)[0],
+        self.assertEqual(len(bad.defects), 1)
+        self.assertTrue(isinstance(bad.defects[0],
                                    errors.StartBoundaryNotFoundDefect))
 
+    # test_parser.TestMessageDefectDetectionBase
     def test_first_line_is_continuation_header(self):
         eq = self.assertEqual
         m = ' Line 1\nLine 2\nLine 3'
-        msg = email.message_from_string(m, policy=self.policy)
+        msg = email.message_from_string(m)
         eq(msg.keys(), [])
         eq(msg.get_payload(), 'Line 2\nLine 3')
-        eq(len(self.get_defects(msg)), 1)
-        self.assertTrue(isinstance(self.get_defects(msg)[0],
+        eq(len(msg.defects), 1)
+        self.assertTrue(isinstance(msg.defects[0],
                                    errors.FirstHeaderLineIsContinuationDefect))
-        eq(self.get_defects(msg)[0].line, ' Line 1\n')
-
-
-class TestNonConformant(TestNonConformantBase, TestEmailBase):
-
-    policy=email.policy.default
-
-    def get_defects(self, obj):
-        return obj.defects
-
-
-class TestNonConformantCapture(TestNonConformantBase, TestEmailBase):
-
-    class CapturePolicy(email.policy.Policy):
-        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 TestRaisingDefects(TestEmailBase):
-
-    def _msgobj(self, filename):
-        with openfile(filename) as fp:
-            return email.message_from_file(fp, policy=email.policy.strict)
-
-    def test_same_boundary_inner_outer(self):
-        with self.assertRaises(errors.StartBoundaryNotFoundDefect):
-            self._msgobj('msg_15.txt')
-
-    def test_multipart_no_boundary(self):
-        with self.assertRaises(errors.NoBoundaryInMultipartDefect):
-            self._msgobj('msg_25.txt')
-
-    def test_lying_multipart(self):
-        with self.assertRaises(errors.NoBoundaryInMultipartDefect):
-            self._msgobj('msg_41.txt')
-
-
-    def test_missing_start_boundary(self):
-        with self.assertRaises(errors.StartBoundaryNotFoundDefect):
-            self._msgobj('msg_42.txt')
-
-    def test_first_line_is_continuation_header(self):
-        m = ' Line 1\nLine 2\nLine 3'
-        with self.assertRaises(errors.FirstHeaderLineIsContinuationDefect):
-            msg = email.message_from_string(m, policy=email.policy.strict)
+        eq(msg.defects[0].line, ' Line 1\n')
 
 
 # Test RFC 2047 header encoding and decoding
 class TestRFC2047(TestEmailBase):
     def test_rfc2047_multiline(self):
         eq = self.assertEqual
         s = """Re: =?mac-iceland?q?r=8Aksm=9Arg=8Cs?= baz
  foo bar =?mac-iceland?q?r=8Aksm=9Arg=8Cs?="""
         dh = decode_header(s)
         eq(dh, [
@@ skipping 571 matching lines @@
             pass
 
         with openfile('msg_01.txt') as fp:
             msg = email.message_from_file(fp, MyMessage)
         unless(isinstance(msg, MyMessage))
         # Try something more complicated
         with openfile('msg_02.txt') as fp:
             msg = email.message_from_file(fp, MyMessage)
         for subpart in msg.walk():
             unless(isinstance(subpart, MyMessage))
+
+    def test_custom_message_does_not_require_arguments(self):
+        class MyMessage(Message):
+            def __init__(self):
+                super().__init__()
+        msg = self._str_msg("Subject: test\n\ntest", MyMessage)
+        self.assertTrue(isinstance(msg, MyMessage))
 
     def test__all__(self):
         module = __import__('email')
         self.assertEqual(sorted(module.__all__), [
             'base64mime', 'charset', 'encoders', 'errors', 'feedparser',
             'generator', 'header', 'iterators', 'message',
             'message_from_binary_file', 'message_from_bytes',
             'message_from_file', 'message_from_string', 'mime', 'parser',
             'quoprimime', 'utils',
             ])
@@ skipping 506 matching lines @@
         eq(part2.get_content_type(), 'application/riscos')
 
     def test_crlf_flatten(self):
         # Using newline='\n' preserves the crlfs in this input file.
         with openfile('msg_26.txt', newline='\n') as fp:
             text = fp.read()
         msg = email.message_from_string(text)
         s = StringIO()
         g = Generator(s)
         g.flatten(msg, linesep='\r\n')
-        self.assertEqual(s.getvalue(), text)
-
-    def test_crlf_control_via_policy(self):
-        with openfile('msg_26.txt', newline='\n') as fp:
-            text = fp.read()
-        msg = email.message_from_string(text)
-        s = StringIO()
-        g = email.generator.Generator(s, policy=email.policy.SMTP)
-        g.flatten(msg)
-        self.assertEqual(s.getvalue(), text)
-
-    def test_flatten_linesep_overrides_policy(self):
-        # msg_27 is lf separated
-        with openfile('msg_27.txt', newline='\n') as fp:
-            text = fp.read()
-        msg = email.message_from_string(text)
-        s = StringIO()
-        g = email.generator.Generator(s, policy=email.policy.SMTP)
-        g.flatten(msg, linesep='\n')
         self.assertEqual(s.getvalue(), text)
 
     maxDiff = None
 
     def test_multipart_digest_with_extra_mime_headers(self):
         eq = self.assertEqual
         neq = self.ndiffAssertEqual
         with openfile('msg_28.txt') as fp:
             msg = email.message_from_file(fp)
         # Structure is:
@@ skipping 471 matching lines @@
         # Since this broke in ByteGenerator, test Generator for completeness.
         m = Message()
         # This has enough non-ascii that it should always end up b encoded.
         m['Subject'] = Header('žluťoučký kůň')
         s = StringIO()
         g = email.generator.Generator(s)
         g.flatten(m, linesep='\r\n')
         self.assertEqual(
             s.getvalue(),
             'Subject: =?utf-8?b?xb5sdcWlb3XEjWvDvSBrxa/FiA==?=\r\n\r\n')
-
-    def test_crlf_control_via_policy(self):
-        # msg_26 is crlf terminated
-        with openfile('msg_26.txt', 'rb') as fp:
-            text = fp.read()
-        msg = email.message_from_bytes(text)
-        s = BytesIO()
-        g = email.generator.BytesGenerator(s, policy=email.policy.SMTP)
-        g.flatten(msg)
-        self.assertEqual(s.getvalue(), text)
-
-    def test_flatten_linesep_overrides_policy(self):
-        # msg_27 is lf separated
-        with openfile('msg_27.txt', 'rb') as fp:
-            text = fp.read()
-        msg = email.message_from_bytes(text)
-        s = BytesIO()
-        g = email.generator.BytesGenerator(s, policy=email.policy.SMTP)
-        g.flatten(msg, linesep='\n')
-        self.assertEqual(s.getvalue(), text)
-
-    def test_must_be_7bit_handles_unknown_8bit(self):
-        msg = email.message_from_bytes(self.non_latin_bin_msg)
-        out = BytesIO()
-        g = email.generator.BytesGenerator(out,
-                        policy=email.policy.default.clone(must_be_7bit=True))
-        g.flatten(msg)
-        self.assertEqual(out.getvalue(),
-            self.non_latin_bin_msg_as7bit_wrapped.encode('ascii'))
-
-    def test_must_be_7bit_transforms_8bit_cte(self):
-        msg = email.message_from_bytes(self.latin_bin_msg)
-        out = BytesIO()
-        g = email.generator.BytesGenerator(out,
-                        policy=email.policy.default.clone(must_be_7bit=True))
-        g.flatten(msg)
-        self.assertEqual(out.getvalue(),
-                        self.latin_bin_msg_as7bit.encode('ascii'))
 
     maxDiff = None
 
 
 class BaseTestBytesGeneratorIdempotent:
 
     maxDiff = None
 
     def _msgobj(self, filename):
         with openfile(filename, 'rb') as fp:
@@ skipping 1185 matching lines @@
         original, msg = self._msg_and_obj('msg_45.txt')
         fp = StringIO()
         Generator(fp).flatten(msg)
         result = fp.getvalue()
         self._signed_parts_eq(original, result)
 
 
 
 if __name__ == '__main__':
     unittest.main()