#14731: refactor email policy framework.

This patch primarily does two things: (1) it adds some internal-interface
methods to Policy that allow for Policy to control the parsing and folding of
headers in such a way that we can construct a backward compatibility policy
that is 100% compatible with the 3.2 API, while allowing a new policy to
implement the email6 API.  (2) it adds that backward compatibility policy and
refactors the test suite so that the only differences between the 3.2
test_email.py file and the 3.3 test_email.py file is some small changes in
test framework and the addition of tests for bugs fixed that apply to the 3.2
API.

There are some additional teaks, such as moving just the code needed for the
compatibility policy into _policybase, so that the library code can import
only _policybase.  That way the new code that will be added for email6
will only get imported when a non-compatibility policy is imported.

1 files changed, 110 insertions, 4 deletions

diff --git a/Lib/test/test_email/test_policy.py b/Lib/test/test_email/test_policy.py
index 1c65901..07925a7 100644
--- a/Lib/test/test_email/test_policy.py
+++ b/Lib/test/test_email/test_policy.py
@@ -1,6 +1,10 @@
+import io
 import types
+import textwrap
 import unittest
 import email.policy
+import email.parser
+import email.generator
 
 class PolicyAPITests(unittest.TestCase):
 
@@ -11,14 +15,15 @@ class PolicyAPITests(unittest.TestCase):
     policy_defaults = {
         'max_line_length':          78,
         'linesep':                  '\n',
-        'must_be_7bit':             False,
+        'cte_type':                 '8bit',
         'raise_on_defect':          False,
         }
 
     # For each policy under test, we give here the values of the attributes
     # that are different from the defaults for that policy.
     policies = {
-        email.policy.Policy(): {},
+        email.policy.Compat32(): {},
+        email.policy.compat32: {},
         email.policy.default: {},
         email.policy.SMTP: {'linesep': '\r\n'},
         email.policy.HTTP: {'linesep': '\r\n', 'max_line_length': None},
@@ -44,6 +49,18 @@ class PolicyAPITests(unittest.TestCase):
                 self.assertIn(attr, self.policy_defaults,
                               "{} is not fully tested".format(attr))
 
+    def test_abc(self):
+        with self.assertRaises(TypeError) as cm:
+            email.policy.Policy()
+        msg = str(cm.exception)
+        abstract_methods = ('fold',
+                            'fold_binary',
+                            'header_fetch_parse',
+                            'header_source_parse',
+                            'header_store_parse')
+        for method in abstract_methods:
+            self.assertIn(method, msg)
+
     def test_policy_is_immutable(self):
         for policy in self.policies:
             for attr in self.policy_defaults:
@@ -88,7 +105,7 @@ class PolicyAPITests(unittest.TestCase):
                 self.defects = []
         obj = Dummy()
         defect = object()
-        policy = email.policy.Policy()
+        policy = email.policy.Compat32()
         policy.register_defect(obj, defect)
         self.assertEqual(obj.defects, [defect])
         defect2 = object()
@@ -117,7 +134,7 @@ class PolicyAPITests(unittest.TestCase):
         email.policy.default.handle_defect(foo, defect2)
         self.assertEqual(foo.defects, [defect1, defect2])
 
-    class MyPolicy(email.policy.Policy):
+    class MyPolicy(email.policy.Compat32):
         defects = None
         def __init__(self, *args, **kw):
             super().__init__(*args, defects=[], **kw)
@@ -146,5 +163,94 @@ class PolicyAPITests(unittest.TestCase):
     # For adding subclassed objects, make sure the usual rules apply (subclass
     # wins), but that the order still works (right overrides left).
 
+
+class TestPolicyPropagation(unittest.TestCase):
+
+    # The abstract methods are used by the parser but not by the wrapper
+    # functions that call it, so if the exception gets raised we know that the
+    # policy was actually propagated all the way to feedparser.
+    class MyPolicy(email.policy.Policy):
+        def badmethod(self, *args, **kw):
+            raise Exception("test")
+        fold = fold_binary = header_fetch_parser = badmethod
+        header_source_parse = header_store_parse = badmethod
+
+    def test_message_from_string(self):
+        with self.assertRaisesRegex(Exception, "^test$"):
+            email.message_from_string("Subject: test\n\n",
+                                      policy=self.MyPolicy)
+
+    def test_message_from_bytes(self):
+        with self.assertRaisesRegex(Exception, "^test$"):
+            email.message_from_bytes(b"Subject: test\n\n",
+                                     policy=self.MyPolicy)
+
+    def test_message_from_file(self):
+        f = io.StringIO('Subject: test\n\n')
+        with self.assertRaisesRegex(Exception, "^test$"):
+            email.message_from_file(f, policy=self.MyPolicy)
+
+    def test_message_from_binary_file(self):
+        f = io.BytesIO(b'Subject: test\n\n')
+        with self.assertRaisesRegex(Exception, "^test$"):
+            email.message_from_binary_file(f, policy=self.MyPolicy)
+
+    # These are redundant, but we need them for black-box completeness.
+
+    def test_parser(self):
+        p = email.parser.Parser(policy=self.MyPolicy)
+        with self.assertRaisesRegex(Exception, "^test$"):
+            p.parsestr('Subject: test\n\n')
+
+    def test_bytes_parser(self):
+        p = email.parser.BytesParser(policy=self.MyPolicy)
+        with self.assertRaisesRegex(Exception, "^test$"):
+            p.parsebytes(b'Subject: test\n\n')
+
+    # Now that we've established that all the parse methods get the
+    # policy in to feedparser, we can use message_from_string for
+    # the rest of the propagation tests.
+
+    def _make_msg(self, source='Subject: test\n\n', policy=None):
+        self.policy = email.policy.default.clone() if policy is None else policy
+        return email.message_from_string(source, policy=self.policy)
+
+    def test_parser_propagates_policy_to_message(self):
+        msg = self._make_msg()
+        self.assertIs(msg.policy, self.policy)
+
+    def test_parser_propagates_policy_to_sub_messages(self):
+        msg = self._make_msg(textwrap.dedent("""\
+            Subject: mime test
+            MIME-Version: 1.0
+            Content-Type: multipart/mixed, boundary="XXX"
+
+            --XXX
+            Content-Type: text/plain
+
+            test
+            --XXX
+            Content-Type: text/plain
+
+            test2
+            --XXX--
+            """))
+        for part in msg.walk():
+            self.assertIs(part.policy, self.policy)
+
+    def test_message_policy_propagates_to_generator(self):
+        msg = self._make_msg("Subject: test\nTo: foo\n\n",
+                             policy=email.policy.default.clone(linesep='X'))
+        s = io.StringIO()
+        g = email.generator.Generator(s)
+        g.flatten(msg)
+        self.assertEqual(s.getvalue(), "Subject: testXTo: fooXX")
+
+    def test_message_policy_used_by_as_string(self):
+        msg = self._make_msg("Subject: test\nTo: foo\n\n",
+                             policy=email.policy.default.clone(linesep='X'))
+        self.assertEqual(msg.as_string(), "Subject: testXTo: fooXX")
+
+
 if __name__ == '__main__':
     unittest.main()