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

Lib/test/test_email/test_policy.py

+import io
 import types
+import textwrap
 import unittest
 import email.policy
+import email.parser
+import email.generator
 
 class PolicyAPITests(unittest.TestCase):
 
     longMessage = True
 
     # These default values are the ones set on email.policy.default.
     # If any of these defaults change, the docs must be updated.
     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},
         email.policy.strict: {'raise_on_defect': True},
         }
 
     def test_defaults(self):
         for policy, changed_defaults in self.policies.items():
             expected = self.policy_defaults.copy()
             expected.update(changed_defaults)
             for attr, value in expected.items():
                 self.assertEqual(getattr(policy, attr), value,
                                 ("change {} docs/docstrings if defaults have "
                                 "changed").format(policy))
 
     def test_all_attributes_covered(self):
         for attr in dir(email.policy.default):
             if (attr.startswith('_') or
                isinstance(getattr(email.policy.Policy, attr),
                           types.FunctionType)):
                 continue
             else:
                 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:
                 with self.assertRaisesRegex(AttributeError, attr+".*read-only"):
                     setattr(policy, attr, None)
             with self.assertRaisesRegex(AttributeError, 'no attribute.*foo'):
                 policy.foo = None
 
     def test_set_policy_attrs_when_calledl(self):
@@ skipping 25 matching lines @@
         added = added + email.policy.default
         for attr, value in expected.items():
             self.assertEqual(getattr(added, attr), value)
 
     def test_register_defect(self):
         class Dummy:
             def __init__(self):
                 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()
         policy.register_defect(obj, defect2)
         self.assertEqual(obj.defects, [defect, defect2])
 
     class MyObj:
         def __init__(self):
             self.defects = []
 
     class MyDefect(Exception):
         pass
 
     def test_handle_defect_raises_on_strict(self):
         foo = self.MyObj()
         defect = self.MyDefect("the telly is broken")
         with self.assertRaisesRegex(self.MyDefect, "the telly is broken"):
             email.policy.strict.handle_defect(foo, defect)
 
     def test_handle_defect_registers_defect(self):
         foo = self.MyObj()
         defect1 = self.MyDefect("one")
         email.policy.default.handle_defect(foo, defect1)
         self.assertEqual(foo.defects, [defect1])
         defect2 = self.MyDefect("two")
         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)
         def register_defect(self, obj, defect):
             self.defects.append(defect)
 
     def test_overridden_register_defect_still_raises(self):
         foo = self.MyObj()
         defect = self.MyDefect("the telly is broken")
         with self.assertRaisesRegex(self.MyDefect, "the telly is broken"):
             self.MyPolicy(raise_on_defect=True).handle_defect(foo, defect)
 
     def test_overriden_register_defect_works(self):
         foo = self.MyObj()
         defect1 = self.MyDefect("one")
         my_policy = self.MyPolicy()
         my_policy.handle_defect(foo, defect1)
         self.assertEqual(my_policy.defects, [defect1])
         self.assertEqual(foo.defects, [])
         defect2 = self.MyDefect("two")
         my_policy.handle_defect(foo, defect2)
         self.assertEqual(my_policy.defects, [defect1, defect2])
         self.assertEqual(foo.defects, [])
 
     # XXX: Need subclassing tests.
     # 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()