#11999: sync based on comparing mtimes, not mtime to system clock

1 files changed, 217 insertions, 46 deletions

diff --git a/Lib/test/test_mailbox.py b/Lib/test/test_mailbox.py
index 9be1332a33..8dc73262dd 100644
--- a/Lib/test/test_mailbox.py
+++ b/Lib/test/test_mailbox.py
@@ -7,8 +7,10 @@ import email
 import email.message
 import re
 import io
+import tempfile
 from test import support
 import unittest
+import textwrap
 import mailbox
 import glob
 try:
@@ -21,16 +23,16 @@ class TestBase(unittest.TestCase):
 
     def _check_sample(self, msg):
         # Inspect a mailbox.Message representation of the sample message
-        self.assertTrue(isinstance(msg, email.message.Message))
-        self.assertTrue(isinstance(msg, mailbox.Message))
+        self.assertIsInstance(msg, email.message.Message)
+        self.assertIsInstance(msg, mailbox.Message)
         for key, value in _sample_headers.items():
             self.assertIn(value, msg.get_all(key))
         self.assertTrue(msg.is_multipart())
         self.assertEqual(len(msg.get_payload()), len(_sample_payloads))
         for i, payload in enumerate(_sample_payloads):
             part = msg.get_payload(i)
-            self.assertTrue(isinstance(part, email.message.Message))
-            self.assertFalse(isinstance(part, mailbox.Message))
+            self.assertIsInstance(part, email.message.Message)
+            self.assertNotIsInstance(part, mailbox.Message)
             self.assertEqual(part.get_payload(), payload)
 
     def _delete_recursively(self, target):
@@ -48,6 +50,8 @@ class TestBase(unittest.TestCase):
 
 class TestMailbox(TestBase):
 
+    maxDiff = None
+
     _factory = None     # Overridden by subclasses to reuse tests
     _template = 'From: foo\n\n%s'
 
@@ -69,14 +73,124 @@ class TestMailbox(TestBase):
         self.assertEqual(len(self._box), 2)
         keys.append(self._box.add(email.message_from_string(_sample_message)))
         self.assertEqual(len(self._box), 3)
-        keys.append(self._box.add(io.StringIO(_sample_message)))
+        keys.append(self._box.add(io.BytesIO(_bytes_sample_message)))
         self.assertEqual(len(self._box), 4)
         keys.append(self._box.add(_sample_message))
         self.assertEqual(len(self._box), 5)
+        keys.append(self._box.add(_bytes_sample_message))
+        self.assertEqual(len(self._box), 6)
+        with self.assertWarns(DeprecationWarning):
+            keys.append(self._box.add(
+                io.TextIOWrapper(io.BytesIO(_bytes_sample_message))))
+        self.assertEqual(len(self._box), 7)
         self.assertEqual(self._box.get_string(keys[0]), self._template % 0)
-        for i in (1, 2, 3, 4):
+        for i in (1, 2, 3, 4, 5, 6):
             self._check_sample(self._box[keys[i]])
 
+    _nonascii_msg = textwrap.dedent("""\
+            From: foo
+            Subject: Falinaptár házhozszállítással. Már rendeltél?
+
+            0
+            """)
+
+    def test_add_invalid_8bit_bytes_header(self):
+        key = self._box.add(self._nonascii_msg.encode('latin1'))
+        self.assertEqual(len(self._box), 1)
+        self.assertEqual(self._box.get_bytes(key),
+            self._nonascii_msg.encode('latin1'))
+
+    def test_invalid_nonascii_header_as_string(self):
+        subj = self._nonascii_msg.splitlines()[1]
+        key = self._box.add(subj.encode('latin1'))
+        self.assertEqual(self._box.get_string(key),
+            'Subject: =?unknown-8bit?b?RmFsaW5hcHThciBo4Xpob3pzeuFsbO104XNz'
+            'YWwuIE3hciByZW5kZWx06Ww/?=\n\n')
+
+    def test_add_nonascii_string_header_raises(self):
+        with self.assertRaisesRegex(ValueError, "ASCII-only"):
+            self._box.add(self._nonascii_msg)
+        self._box.flush()
+        self.assertEqual(len(self._box), 0)
+        self.assertMailboxEmpty()
+
+    def test_add_that_raises_leaves_mailbox_empty(self):
+        # XXX This test will start failing when Message learns to handle
+        # non-ASCII string headers, and a different internal failure will
+        # need to be found or manufactured.
+        with self.assertRaises(ValueError):
+            self._box.add(email.message_from_string("From: Alphöso"))
+        self.assertEqual(len(self._box), 0)
+        self._box.close()
+        self.assertMailboxEmpty()
+
+    _non_latin_bin_msg = textwrap.dedent("""\
+        From: foo@bar.com
+        To: báz
+        Subject: Maintenant je vous présente mon collègue, le pouf célèbre
+        \tJean de Baddie
+        Mime-Version: 1.0
+        Content-Type: text/plain; charset="utf-8"
+        Content-Transfer-Encoding: 8bit
+
+        Да, они летят.
+        """).encode('utf-8')
+
+    def test_add_8bit_body(self):
+        key = self._box.add(self._non_latin_bin_msg)
+        self.assertEqual(self._box.get_bytes(key),
+                         self._non_latin_bin_msg)
+        with self._box.get_file(key) as f:
+            self.assertEqual(f.read(),
+                             self._non_latin_bin_msg.replace(b'\n',
+                                os.linesep.encode()))
+        self.assertEqual(self._box[key].get_payload(),
+                        "Да, они летят.\n")
+
+    def test_add_binary_file(self):
+        with tempfile.TemporaryFile('wb+') as f:
+            f.write(_bytes_sample_message)
+            f.seek(0)
+            key = self._box.add(f)
+        # See issue 11062
+        if not isinstance(self._box, mailbox.Babyl):
+            self.assertEqual(self._box.get_bytes(key).split(b'\n'),
+                _bytes_sample_message.split(b'\n'))
+
+    def test_add_binary_nonascii_file(self):
+        with tempfile.TemporaryFile('wb+') as f:
+            f.write(self._non_latin_bin_msg)
+            f.seek(0)
+            key = self._box.add(f)
+        # See issue 11062
+        if not isinstance(self._box, mailbox.Babyl):
+            self.assertEqual(self._box.get_bytes(key).split(b'\n'),
+                self._non_latin_bin_msg.split(b'\n'))
+
+    def test_add_text_file_warns(self):
+        with tempfile.TemporaryFile('w+') as f:
+            f.write(_sample_message)
+            f.seek(0)
+            with self.assertWarns(DeprecationWarning):
+                key = self._box.add(f)
+        # See issue 11062
+        if not isinstance(self._box, mailbox.Babyl):
+            self.assertEqual(self._box.get_bytes(key).split(b'\n'),
+                _bytes_sample_message.split(b'\n'))
+
+    def test_add_StringIO_warns(self):
+        with self.assertWarns(DeprecationWarning):
+            key = self._box.add(io.StringIO(self._template % "0"))
+        self.assertEqual(self._box.get_string(key), self._template % "0")
+
+    def test_add_nonascii_StringIO_raises(self):
+        with self.assertWarns(DeprecationWarning):
+            with self.assertRaisesRegex(ValueError, "ASCII-only"):
+                self._box.add(io.StringIO(self._nonascii_msg))
+        self.assertEqual(len(self._box), 0)
+        self._box.close()
+        self.assertMailboxEmpty()
+
     def test_remove(self):
         # Remove messages using remove()
         self._test_remove_or_delitem(self._box.remove)
@@ -149,27 +263,38 @@ class TestMailbox(TestBase):
         key0 = self._box.add(self._template % 0)
         key1 = self._box.add(_sample_message)
         msg0 = self._box.get_message(key0)
-        self.assertTrue(isinstance(msg0, mailbox.Message))
+        self.assertIsInstance(msg0, mailbox.Message)
         self.assertEqual(msg0['from'], 'foo')
         self.assertEqual(msg0.get_payload(), '0')
         self._check_sample(self._box.get_message(key1))
 
+    def test_get_bytes(self):
+        # Get bytes representations of messages
+        key0 = self._box.add(self._template % 0)
+        key1 = self._box.add(_sample_message)
+        self.assertEqual(self._box.get_bytes(key0),
+            (self._template % 0).encode('ascii'))
+        self.assertEqual(self._box.get_bytes(key1), _bytes_sample_message)
+
     def test_get_string(self):
         # Get string representations of messages
         key0 = self._box.add(self._template % 0)
         key1 = self._box.add(_sample_message)
         self.assertEqual(self._box.get_string(key0), self._template % 0)
-        self.assertEqual(self._box.get_string(key1), _sample_message)
+        self.assertEqual(self._box.get_string(key1).split('\n'),
+                         _sample_message.split('\n'))
 
     def test_get_file(self):
         # Get file representations of messages
         key0 = self._box.add(self._template % 0)
         key1 = self._box.add(_sample_message)
-        data0 = self._box.get_file(key0).read()
-        data1 = self._box.get_file(key1).read()
-        self.assertEqual(data0.replace(os.linesep, '\n'),
+        with self._box.get_file(key0) as file:
+            data0 = file.read()
+        with self._box.get_file(key1) as file:
+            data1 = file.read()
+        self.assertEqual(data0.decode('ascii').replace(os.linesep, '\n'),
                          self._template % 0)
-        self.assertEqual(data1.replace(os.linesep, '\n'),
+        self.assertEqual(data1.decode('ascii').replace(os.linesep, '\n'),
                          _sample_message)
 
     def test_iterkeys(self):
@@ -403,11 +528,12 @@ class TestMailbox(TestBase):
     def test_dump_message(self):
         # Write message representations to disk
         for input in (email.message_from_string(_sample_message),
-                      _sample_message, io.StringIO(_sample_message)):
-            output = io.StringIO()
+                      _sample_message, io.BytesIO(_bytes_sample_message)):
+            output = io.BytesIO()
             self._box._dump_message(input, output)
-            self.assertEqual(output.getvalue(), _sample_message)
-        output = io.StringIO()
+            self.assertEqual(output.getvalue(),
+                _bytes_sample_message.replace(b'\n', os.linesep.encode()))
+        output = io.BytesIO()
         self.assertRaises(TypeError,
                           lambda: self._box._dump_message(None, output))
 
@@ -437,6 +563,7 @@ class TestMailboxSuperclass(TestBase):
         self.assertRaises(NotImplementedError, lambda: box.__getitem__(''))
         self.assertRaises(NotImplementedError, lambda: box.get_message(''))
         self.assertRaises(NotImplementedError, lambda: box.get_string(''))
+        self.assertRaises(NotImplementedError, lambda: box.get_bytes(''))
         self.assertRaises(NotImplementedError, lambda: box.get_file(''))
         self.assertRaises(NotImplementedError, lambda: '' in box)
         self.assertRaises(NotImplementedError, lambda: box.__contains__(''))
@@ -460,6 +587,9 @@ class TestMaildir(TestMailbox):
         if os.name in ('nt', 'os2') or sys.platform == 'cygwin':
             self._box.colon = '!'
 
+    def assertMailboxEmpty(self):
+        self.assertEqual(os.listdir(os.path.join(self._path, 'tmp')), [])
+
     def test_add_MM(self):
         # Add a MaildirMessage instance
         msg = mailbox.MaildirMessage(self._template % 0)
@@ -476,7 +606,7 @@ class TestMaildir(TestMailbox):
         msg.set_flags('RF')
         key = self._box.add(msg)
         msg_returned = self._box.get_message(key)
-        self.assertTrue(isinstance(msg_returned, mailbox.MaildirMessage))
+        self.assertIsInstance(msg_returned, mailbox.MaildirMessage)
         self.assertEqual(msg_returned.get_subdir(), 'cur')
         self.assertEqual(msg_returned.get_flags(), 'FR')
 
@@ -516,7 +646,7 @@ class TestMaildir(TestMailbox):
         box = mailbox.Maildir(self._path, factory=FakeMessage)
         box.colon = self._box.colon
         msg2 = box.get_message(key)
-        self.assertTrue(isinstance(msg2, FakeMessage))
+        self.assertIsInstance(msg2, FakeMessage)
 
     def test_initialize_new(self):
         # Initialize a non-existent mailbox
@@ -586,12 +716,10 @@ class TestMaildir(TestMailbox):
         # Remove old files from 'tmp'
         foo_path = os.path.join(self._path, 'tmp', 'foo')
         bar_path = os.path.join(self._path, 'tmp', 'bar')
-        f = open(foo_path, 'w')
-        f.write("@")
-        f.close()
-        f = open(bar_path, 'w')
-        f.write("@")
-        f.close()
+        with open(foo_path, 'w') as f:
+            f.write("@")
+        with open(bar_path, 'w') as f:
+            f.write("@")
         self._box.clean()
         self.assertTrue(os.path.exists(foo_path))
         self.assertTrue(os.path.exists(bar_path))
@@ -640,9 +768,9 @@ class TestMaildir(TestMailbox):
                              "Host name mismatch: '%s' should be '%s'" %
                              (groups[4], hostname))
             previous_groups = groups
-            tmp_file.write(_sample_message)
+            tmp_file.write(_bytes_sample_message)
             tmp_file.seek(0)
-            self.assertEqual(tmp_file.read(), _sample_message)
+            self.assertEqual(tmp_file.read(), _bytes_sample_message)
             tmp_file.close()
         file_count = len(os.listdir(os.path.join(self._path, "tmp")))
         self.assertEqual(file_count, repetitions,
@@ -776,17 +904,28 @@ class TestMaildir(TestMailbox):
 class _TestMboxMMDF(TestMailbox):
 
     def tearDown(self):
+        super().tearDown()
         self._box.close()
         self._delete_recursively(self._path)
         for lock_remnant in glob.glob(self._path + '.*'):
             support.unlink(lock_remnant)
 
+    def assertMailboxEmpty(self):
+        with open(self._path) as f:
+            self.assertEqual(f.readlines(), [])
+
     def test_add_from_string(self):
         # Add a string starting with 'From ' to the mailbox
         key = self._box.add('From foo@bar blah\nFrom: foo\n\n0')
         self.assertEqual(self._box[key].get_from(), 'foo@bar blah')
         self.assertEqual(self._box[key].get_payload(), '0')
 
+    def test_add_from_bytes(self):
+        # Add a byte string starting with 'From ' to the mailbox
+        key = self._box.add(b'From foo@bar blah\nFrom: foo\n\n0')
+        self.assertEqual(self._box[key].get_from(), 'foo@bar blah')
+        self.assertEqual(self._box[key].get_payload(), '0')
+
     def test_add_mbox_or_mmdf_message(self):
         # Add an mboxMessage or MMDFMessage
         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
@@ -817,7 +956,8 @@ class _TestMboxMMDF(TestMailbox):
         self._box._file.seek(0)
         contents = self._box._file.read()
         self._box.close()
-        self.assertEqual(contents, open(self._path, 'r', newline='').read())
+        with open(self._path, 'rb') as f:
+            self.assertEqual(contents, f.read())
         self._box = self._factory(self._path)
 
     def test_lock_conflict(self):
@@ -896,6 +1036,9 @@ class TestMH(TestMailbox):
 
     _factory = lambda self, path, factory=None: mailbox.MH(path, factory)
 
+    def assertMailboxEmpty(self):
+        self.assertEqual(os.listdir(self._path), ['.mh_sequences'])
+
     def test_list_folders(self):
         # List folders
         self._box.add_folder('one')
@@ -969,6 +1112,13 @@ class TestMH(TestMailbox):
         key0 = self._box.add(msg0)
         refmsg0 = self._box.get_message(key0)
 
+    def test_issue7627(self):
+        msg0 = mailbox.MHMessage(self._template % 0)
+        key0 = self._box.add(msg0)
+        self._box.lock()
+        self._box.remove(key0)
+        self._box.unlock()
+
     def test_pack(self):
         # Pack the contents of the mailbox
         msg0 = mailbox.MHMessage(self._template % 0)
@@ -1021,7 +1171,12 @@ class TestBabyl(TestMailbox):
 
     _factory = lambda self, path, factory=None: mailbox.Babyl(path, factory)
 
+    def assertMailboxEmpty(self):
+        with open(self._path) as f:
+            self.assertEqual(f.readlines(), [])
+
     def tearDown(self):
+        super().tearDown()
         self._box.close()
         self._delete_recursively(self._path)
         for lock_remnant in glob.glob(self._path + '.*'):
@@ -1071,21 +1226,29 @@ class TestMessage(TestBase):
 
     def test_initialize_with_file(self):
         # Initialize based on contents of file
-        f = open(self._path, 'w+')
-        f.write(_sample_message)
-        f.seek(0)
-        msg = self._factory(f)
-        self._post_initialize_hook(msg)
-        self._check_sample(msg)
-        f.close()
+        with open(self._path, 'w+') as f:
+            f.write(_sample_message)
+            f.seek(0)
+            msg = self._factory(f)
+            self._post_initialize_hook(msg)
+            self._check_sample(msg)
+
+    def test_initialize_with_binary_file(self):
+        # Initialize based on contents of binary file
+        with open(self._path, 'wb+') as f:
+            f.write(_bytes_sample_message)
+            f.seek(0)
+            msg = self._factory(f)
+            self._post_initialize_hook(msg)
+            self._check_sample(msg)
 
     def test_initialize_with_nothing(self):
         # Initialize without arguments
         msg = self._factory()
         self._post_initialize_hook(msg)
-        self.assertTrue(isinstance(msg, email.message.Message))
-        self.assertTrue(isinstance(msg, mailbox.Message))
-        self.assertTrue(isinstance(msg, self._factory))
+        self.assertIsInstance(msg, email.message.Message)
+        self.assertIsInstance(msg, mailbox.Message)
+        self.assertIsInstance(msg, self._factory)
         self.assertEqual(msg.keys(), [])
         self.assertFalse(msg.is_multipart())
         self.assertEqual(msg.get_payload(), None)
@@ -1355,6 +1518,14 @@ class TestMessageConversion(TestBase):
             msg_plain = mailbox.Message(msg)
             self._check_sample(msg_plain)
 
+    def test_x_from_bytes(self):
+        # Convert all formats to Message
+        for class_ in (mailbox.Message, mailbox.MaildirMessage,
+                       mailbox.mboxMessage, mailbox.MHMessage,
+                       mailbox.BabylMessage, mailbox.MMDFMessage):
+            msg = class_(_bytes_sample_message)
+            self._check_sample(msg)
+
     def test_x_to_invalid(self):
         # Convert all formats to an invalid format
         for class_ in (mailbox.Message, mailbox.MaildirMessage,
@@ -1801,18 +1972,16 @@ class MaildirTestCase(unittest.TestCase):
         filename = ".".join((str(t), str(pid), "myhostname", "mydomain"))
         tmpname = os.path.join(self._dir, "tmp", filename)
         newname = os.path.join(self._dir, dir, filename)
-        fp = open(tmpname, "w")
-        self._msgfiles.append(tmpname)
-        if mbox:
-            fp.write(FROM_)
-        fp.write(DUMMY_MESSAGE)
-        fp.close()
+        with open(tmpname, "w") as fp:
+            self._msgfiles.append(tmpname)
+            if mbox:
+                fp.write(FROM_)
+            fp.write(DUMMY_MESSAGE)
         if hasattr(os, "link"):
             os.link(tmpname, newname)
         else:
-            fp = open(newname, "w")
-            fp.write(DUMMY_MESSAGE)
-            fp.close()
+            with open(newname, "w") as fp:
+                fp.write(DUMMY_MESSAGE)
         self._msgfiles.append(newname)
         return tmpname
 
@@ -1902,6 +2071,8 @@ H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs
 --NMuMz9nt05w80d4+--
 """
 
+_bytes_sample_message = _sample_message.encode('ascii')
+
 _sample_headers = {
     "Return-Path":"<gkj@gregorykjohnson.com>",
     "X-Original-To":"gkj+person@localhost",