diff options
Diffstat (limited to 'lib/python2.7/test/test_mailbox.py')
| -rw-r--r-- | lib/python2.7/test/test_mailbox.py | 2122 |
1 files changed, 0 insertions, 2122 deletions
diff --git a/lib/python2.7/test/test_mailbox.py b/lib/python2.7/test/test_mailbox.py deleted file mode 100644 index 5e8e3c1..0000000 --- a/lib/python2.7/test/test_mailbox.py +++ /dev/null @@ -1,2122 +0,0 @@ -import os -import sys -import time -import stat -import socket -import email -import email.message -import re -import shutil -import StringIO -import tempfile -from test import test_support -import unittest -import mailbox -import glob -try: - import fcntl -except ImportError: - pass - -# Silence Py3k warning -rfc822 = test_support.import_module('rfc822', deprecated=True) - -class TestBase: - - def _check_sample(self, msg): - # Inspect a mailbox.Message representation of the sample message - self.assertIsInstance(msg, email.message.Message) - self.assertIsInstance(msg, mailbox.Message) - for key, value in _sample_headers.iteritems(): - 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.assertIsInstance(part, email.message.Message) - self.assertNotIsInstance(part, mailbox.Message) - self.assertEqual(part.get_payload(), payload) - - def _delete_recursively(self, target): - # Delete a file or delete a directory recursively - if os.path.isdir(target): - test_support.rmtree(target) - elif os.path.exists(target): - test_support.unlink(target) - - -class TestMailbox(TestBase): - - _factory = None # Overridden by subclasses to reuse tests - _template = 'From: foo\n\n%s\n' - - def setUp(self): - self._path = test_support.TESTFN - self._delete_recursively(self._path) - self._box = self._factory(self._path) - - def tearDown(self): - self._box.close() - self._delete_recursively(self._path) - - def test_add(self): - # Add copies of a sample message - keys = [] - keys.append(self._box.add(self._template % 0)) - self.assertEqual(len(self._box), 1) - keys.append(self._box.add(mailbox.Message(_sample_message))) - 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(StringIO.StringIO(_sample_message))) - self.assertEqual(len(self._box), 4) - keys.append(self._box.add(_sample_message)) - self.assertEqual(len(self._box), 5) - self.assertEqual(self._box.get_string(keys[0]), self._template % 0) - for i in (1, 2, 3, 4): - self._check_sample(self._box[keys[i]]) - - def test_add_file(self): - with tempfile.TemporaryFile('w+') as f: - f.write(_sample_message) - f.seek(0) - key = self._box.add(f) - self.assertEqual(self._box.get_string(key).split('\n'), - _sample_message.split('\n')) - - def test_add_StringIO(self): - key = self._box.add(StringIO.StringIO(self._template % "0")) - self.assertEqual(self._box.get_string(key), self._template % "0") - - def test_remove(self): - # Remove messages using remove() - self._test_remove_or_delitem(self._box.remove) - - def test_delitem(self): - # Remove messages using __delitem__() - self._test_remove_or_delitem(self._box.__delitem__) - - def _test_remove_or_delitem(self, method): - # (Used by test_remove() and test_delitem().) - key0 = self._box.add(self._template % 0) - key1 = self._box.add(self._template % 1) - self.assertEqual(len(self._box), 2) - method(key0) - l = len(self._box) - self.assertEqual(l, 1) - self.assertRaises(KeyError, lambda: self._box[key0]) - self.assertRaises(KeyError, lambda: method(key0)) - self.assertEqual(self._box.get_string(key1), self._template % 1) - key2 = self._box.add(self._template % 2) - self.assertEqual(len(self._box), 2) - method(key2) - l = len(self._box) - self.assertEqual(l, 1) - self.assertRaises(KeyError, lambda: self._box[key2]) - self.assertRaises(KeyError, lambda: method(key2)) - self.assertEqual(self._box.get_string(key1), self._template % 1) - method(key1) - self.assertEqual(len(self._box), 0) - self.assertRaises(KeyError, lambda: self._box[key1]) - self.assertRaises(KeyError, lambda: method(key1)) - - def test_discard(self, repetitions=10): - # Discard messages - key0 = self._box.add(self._template % 0) - key1 = self._box.add(self._template % 1) - self.assertEqual(len(self._box), 2) - self._box.discard(key0) - self.assertEqual(len(self._box), 1) - self.assertRaises(KeyError, lambda: self._box[key0]) - self._box.discard(key0) - self.assertEqual(len(self._box), 1) - self.assertRaises(KeyError, lambda: self._box[key0]) - - def test_get(self): - # Retrieve messages using get() - key0 = self._box.add(self._template % 0) - msg = self._box.get(key0) - self.assertEqual(msg['from'], 'foo') - self.assertEqual(msg.get_payload(), '0\n') - self.assertIs(self._box.get('foo'), None) - self.assertFalse(self._box.get('foo', False)) - self._box.close() - self._box = self._factory(self._path, factory=rfc822.Message) - key1 = self._box.add(self._template % 1) - msg = self._box.get(key1) - self.assertEqual(msg['from'], 'foo') - self.assertEqual(msg.fp.read(), '1' + os.linesep) - msg.fp.close() - - def test_getitem(self): - # Retrieve message using __getitem__() - key0 = self._box.add(self._template % 0) - msg = self._box[key0] - self.assertEqual(msg['from'], 'foo') - self.assertEqual(msg.get_payload(), '0\n') - self.assertRaises(KeyError, lambda: self._box['foo']) - self._box.discard(key0) - self.assertRaises(KeyError, lambda: self._box[key0]) - - def test_get_message(self): - # Get Message representations of messages - key0 = self._box.add(self._template % 0) - key1 = self._box.add(_sample_message) - msg0 = self._box.get_message(key0) - self.assertIsInstance(msg0, mailbox.Message) - self.assertEqual(msg0['from'], 'foo') - self.assertEqual(msg0.get_payload(), '0\n') - self._check_sample(self._box.get_message(key1)) - - 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) - - def test_get_file(self): - # Get file representations of messages - key0 = self._box.add(self._template % 0) - key1 = self._box.add(_sample_message) - msg0 = self._box.get_file(key0) - self.assertEqual(msg0.read().replace(os.linesep, '\n'), - self._template % 0) - msg1 = self._box.get_file(key1) - self.assertEqual(msg1.read().replace(os.linesep, '\n'), - _sample_message) - msg0.close() - msg1.close() - - def test_get_file_can_be_closed_twice(self): - # Issue 11700 - key = self._box.add(_sample_message) - f = self._box.get_file(key) - f.close() - f.close() - - def test_iterkeys(self): - # Get keys using iterkeys() - self._check_iteration(self._box.iterkeys, do_keys=True, do_values=False) - - def test_keys(self): - # Get keys using keys() - self._check_iteration(self._box.keys, do_keys=True, do_values=False) - - def test_itervalues(self): - # Get values using itervalues() - self._check_iteration(self._box.itervalues, do_keys=False, - do_values=True) - - def test_iter(self): - # Get values using __iter__() - self._check_iteration(self._box.__iter__, do_keys=False, - do_values=True) - - def test_values(self): - # Get values using values() - self._check_iteration(self._box.values, do_keys=False, do_values=True) - - def test_iteritems(self): - # Get keys and values using iteritems() - self._check_iteration(self._box.iteritems, do_keys=True, - do_values=True) - - def test_items(self): - # Get keys and values using items() - self._check_iteration(self._box.items, do_keys=True, do_values=True) - - def _check_iteration(self, method, do_keys, do_values, repetitions=10): - for value in method(): - self.fail("Not empty") - keys, values = [], [] - for i in xrange(repetitions): - keys.append(self._box.add(self._template % i)) - values.append(self._template % i) - if do_keys and not do_values: - returned_keys = list(method()) - elif do_values and not do_keys: - returned_values = list(method()) - else: - returned_keys, returned_values = [], [] - for key, value in method(): - returned_keys.append(key) - returned_values.append(value) - if do_keys: - self.assertEqual(len(keys), len(returned_keys)) - self.assertEqual(set(keys), set(returned_keys)) - if do_values: - count = 0 - for value in returned_values: - self.assertEqual(value['from'], 'foo') - self.assertTrue(int(value.get_payload()) < repetitions, - (value.get_payload(), repetitions)) - count += 1 - self.assertEqual(len(values), count) - - def test_has_key(self): - # Check existence of keys using has_key() - self._test_has_key_or_contains(self._box.has_key) - - def test_contains(self): - # Check existence of keys using __contains__() - self._test_has_key_or_contains(self._box.__contains__) - - def _test_has_key_or_contains(self, method): - # (Used by test_has_key() and test_contains().) - self.assertFalse(method('foo')) - key0 = self._box.add(self._template % 0) - self.assertTrue(method(key0)) - self.assertFalse(method('foo')) - key1 = self._box.add(self._template % 1) - self.assertTrue(method(key1)) - self.assertTrue(method(key0)) - self.assertFalse(method('foo')) - self._box.remove(key0) - self.assertFalse(method(key0)) - self.assertTrue(method(key1)) - self.assertFalse(method('foo')) - self._box.remove(key1) - self.assertFalse(method(key1)) - self.assertFalse(method(key0)) - self.assertFalse(method('foo')) - - def test_len(self, repetitions=10): - # Get message count - keys = [] - for i in xrange(repetitions): - self.assertEqual(len(self._box), i) - keys.append(self._box.add(self._template % i)) - self.assertEqual(len(self._box), i + 1) - for i in xrange(repetitions): - self.assertEqual(len(self._box), repetitions - i) - self._box.remove(keys[i]) - self.assertEqual(len(self._box), repetitions - i - 1) - - def test_set_item(self): - # Modify messages using __setitem__() - key0 = self._box.add(self._template % 'original 0') - self.assertEqual(self._box.get_string(key0), - self._template % 'original 0') - key1 = self._box.add(self._template % 'original 1') - self.assertEqual(self._box.get_string(key1), - self._template % 'original 1') - self._box[key0] = self._template % 'changed 0' - self.assertEqual(self._box.get_string(key0), - self._template % 'changed 0') - self._box[key1] = self._template % 'changed 1' - self.assertEqual(self._box.get_string(key1), - self._template % 'changed 1') - self._box[key0] = _sample_message - self._check_sample(self._box[key0]) - self._box[key1] = self._box[key0] - self._check_sample(self._box[key1]) - self._box[key0] = self._template % 'original 0' - self.assertEqual(self._box.get_string(key0), - self._template % 'original 0') - self._check_sample(self._box[key1]) - self.assertRaises(KeyError, - lambda: self._box.__setitem__('foo', 'bar')) - self.assertRaises(KeyError, lambda: self._box['foo']) - self.assertEqual(len(self._box), 2) - - def test_clear(self, iterations=10): - # Remove all messages using clear() - keys = [] - for i in xrange(iterations): - self._box.add(self._template % i) - for i, key in enumerate(keys): - self.assertEqual(self._box.get_string(key), self._template % i) - self._box.clear() - self.assertEqual(len(self._box), 0) - for i, key in enumerate(keys): - self.assertRaises(KeyError, lambda: self._box.get_string(key)) - - def test_pop(self): - # Get and remove a message using pop() - key0 = self._box.add(self._template % 0) - self.assertIn(key0, self._box) - key1 = self._box.add(self._template % 1) - self.assertIn(key1, self._box) - self.assertEqual(self._box.pop(key0).get_payload(), '0\n') - self.assertNotIn(key0, self._box) - self.assertIn(key1, self._box) - key2 = self._box.add(self._template % 2) - self.assertIn(key2, self._box) - self.assertEqual(self._box.pop(key2).get_payload(), '2\n') - self.assertNotIn(key2, self._box) - self.assertIn(key1, self._box) - self.assertEqual(self._box.pop(key1).get_payload(), '1\n') - self.assertNotIn(key1, self._box) - self.assertEqual(len(self._box), 0) - - def test_popitem(self, iterations=10): - # Get and remove an arbitrary (key, message) using popitem() - keys = [] - for i in xrange(10): - keys.append(self._box.add(self._template % i)) - seen = [] - for i in xrange(10): - key, msg = self._box.popitem() - self.assertIn(key, keys) - self.assertNotIn(key, seen) - seen.append(key) - self.assertEqual(int(msg.get_payload()), keys.index(key)) - self.assertEqual(len(self._box), 0) - for key in keys: - self.assertRaises(KeyError, lambda: self._box[key]) - - def test_update(self): - # Modify multiple messages using update() - key0 = self._box.add(self._template % 'original 0') - key1 = self._box.add(self._template % 'original 1') - key2 = self._box.add(self._template % 'original 2') - self._box.update({key0: self._template % 'changed 0', - key2: _sample_message}) - self.assertEqual(len(self._box), 3) - self.assertEqual(self._box.get_string(key0), - self._template % 'changed 0') - self.assertEqual(self._box.get_string(key1), - self._template % 'original 1') - self._check_sample(self._box[key2]) - self._box.update([(key2, self._template % 'changed 2'), - (key1, self._template % 'changed 1'), - (key0, self._template % 'original 0')]) - self.assertEqual(len(self._box), 3) - self.assertEqual(self._box.get_string(key0), - self._template % 'original 0') - self.assertEqual(self._box.get_string(key1), - self._template % 'changed 1') - self.assertEqual(self._box.get_string(key2), - self._template % 'changed 2') - self.assertRaises(KeyError, - lambda: self._box.update({'foo': 'bar', - key0: self._template % "changed 0"})) - self.assertEqual(len(self._box), 3) - self.assertEqual(self._box.get_string(key0), - self._template % "changed 0") - self.assertEqual(self._box.get_string(key1), - self._template % "changed 1") - self.assertEqual(self._box.get_string(key2), - self._template % "changed 2") - - def test_flush(self): - # Write changes to disk - self._test_flush_or_close(self._box.flush, True) - - def test_popitem_and_flush_twice(self): - # See #15036. - self._box.add(self._template % 0) - self._box.add(self._template % 1) - self._box.flush() - - self._box.popitem() - self._box.flush() - self._box.popitem() - self._box.flush() - - def test_lock_unlock(self): - # Lock and unlock the mailbox - self.assertFalse(os.path.exists(self._get_lock_path())) - self._box.lock() - self.assertTrue(os.path.exists(self._get_lock_path())) - self._box.unlock() - self.assertFalse(os.path.exists(self._get_lock_path())) - - def test_close(self): - # Close mailbox and flush changes to disk - self._test_flush_or_close(self._box.close, False) - - def _test_flush_or_close(self, method, should_call_close): - contents = [self._template % i for i in xrange(3)] - self._box.add(contents[0]) - self._box.add(contents[1]) - self._box.add(contents[2]) - oldbox = self._box - method() - if should_call_close: - self._box.close() - self._box = self._factory(self._path) - keys = self._box.keys() - self.assertEqual(len(keys), 3) - for key in keys: - self.assertIn(self._box.get_string(key), contents) - oldbox.close() - - def test_dump_message(self): - # Write message representations to disk - for input in (email.message_from_string(_sample_message), - _sample_message, StringIO.StringIO(_sample_message)): - output = StringIO.StringIO() - self._box._dump_message(input, output) - self.assertEqual(output.getvalue(), - _sample_message.replace('\n', os.linesep)) - output = StringIO.StringIO() - self.assertRaises(TypeError, - lambda: self._box._dump_message(None, output)) - - def _get_lock_path(self): - # Return the path of the dot lock file. May be overridden. - return self._path + '.lock' - - -class TestMailboxSuperclass(TestBase, unittest.TestCase): - - def test_notimplemented(self): - # Test that all Mailbox methods raise NotImplementedException. - box = mailbox.Mailbox('path') - self.assertRaises(NotImplementedError, lambda: box.add('')) - self.assertRaises(NotImplementedError, lambda: box.remove('')) - self.assertRaises(NotImplementedError, lambda: box.__delitem__('')) - self.assertRaises(NotImplementedError, lambda: box.discard('')) - self.assertRaises(NotImplementedError, lambda: box.__setitem__('', '')) - self.assertRaises(NotImplementedError, lambda: box.iterkeys()) - self.assertRaises(NotImplementedError, lambda: box.keys()) - self.assertRaises(NotImplementedError, lambda: box.itervalues().next()) - self.assertRaises(NotImplementedError, lambda: box.__iter__().next()) - self.assertRaises(NotImplementedError, lambda: box.values()) - self.assertRaises(NotImplementedError, lambda: box.iteritems().next()) - self.assertRaises(NotImplementedError, lambda: box.items()) - self.assertRaises(NotImplementedError, lambda: box.get('')) - 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_file('')) - self.assertRaises(NotImplementedError, lambda: box.has_key('')) - self.assertRaises(NotImplementedError, lambda: box.__contains__('')) - self.assertRaises(NotImplementedError, lambda: box.__len__()) - self.assertRaises(NotImplementedError, lambda: box.clear()) - self.assertRaises(NotImplementedError, lambda: box.pop('')) - self.assertRaises(NotImplementedError, lambda: box.popitem()) - self.assertRaises(NotImplementedError, lambda: box.update((('', ''),))) - self.assertRaises(NotImplementedError, lambda: box.flush()) - self.assertRaises(NotImplementedError, lambda: box.lock()) - self.assertRaises(NotImplementedError, lambda: box.unlock()) - self.assertRaises(NotImplementedError, lambda: box.close()) - - -class TestMaildir(TestMailbox, unittest.TestCase): - - _factory = lambda self, path, factory=None: mailbox.Maildir(path, factory) - - def setUp(self): - TestMailbox.setUp(self) - if os.name in ('nt', 'os2') or sys.platform == 'cygwin': - self._box.colon = '!' - - def test_add_MM(self): - # Add a MaildirMessage instance - msg = mailbox.MaildirMessage(self._template % 0) - msg.set_subdir('cur') - msg.set_info('foo') - key = self._box.add(msg) - self.assertTrue(os.path.exists(os.path.join(self._path, 'cur', '%s%sfoo' % - (key, self._box.colon)))) - - def test_get_MM(self): - # Get a MaildirMessage instance - msg = mailbox.MaildirMessage(self._template % 0) - msg.set_subdir('cur') - msg.set_flags('RF') - key = self._box.add(msg) - msg_returned = self._box.get_message(key) - self.assertIsInstance(msg_returned, mailbox.MaildirMessage) - self.assertEqual(msg_returned.get_subdir(), 'cur') - self.assertEqual(msg_returned.get_flags(), 'FR') - - def test_set_MM(self): - # Set with a MaildirMessage instance - msg0 = mailbox.MaildirMessage(self._template % 0) - msg0.set_flags('TP') - key = self._box.add(msg0) - msg_returned = self._box.get_message(key) - self.assertEqual(msg_returned.get_subdir(), 'new') - self.assertEqual(msg_returned.get_flags(), 'PT') - msg1 = mailbox.MaildirMessage(self._template % 1) - self._box[key] = msg1 - msg_returned = self._box.get_message(key) - self.assertEqual(msg_returned.get_subdir(), 'new') - self.assertEqual(msg_returned.get_flags(), '') - self.assertEqual(msg_returned.get_payload(), '1\n') - msg2 = mailbox.MaildirMessage(self._template % 2) - msg2.set_info('2,S') - self._box[key] = msg2 - self._box[key] = self._template % 3 - msg_returned = self._box.get_message(key) - self.assertEqual(msg_returned.get_subdir(), 'new') - self.assertEqual(msg_returned.get_flags(), 'S') - self.assertEqual(msg_returned.get_payload(), '3\n') - - def test_consistent_factory(self): - # Add a message. - msg = mailbox.MaildirMessage(self._template % 0) - msg.set_subdir('cur') - msg.set_flags('RF') - key = self._box.add(msg) - - # Create new mailbox with - class FakeMessage(mailbox.MaildirMessage): - pass - box = mailbox.Maildir(self._path, factory=FakeMessage) - box.colon = self._box.colon - msg2 = box.get_message(key) - self.assertIsInstance(msg2, FakeMessage) - - def test_initialize_new(self): - # Initialize a non-existent mailbox - self.tearDown() - self._box = mailbox.Maildir(self._path) - self._check_basics(factory=rfc822.Message) - self._delete_recursively(self._path) - self._box = self._factory(self._path, factory=None) - self._check_basics() - - def test_initialize_existing(self): - # Initialize an existing mailbox - self.tearDown() - for subdir in '', 'tmp', 'new', 'cur': - os.mkdir(os.path.normpath(os.path.join(self._path, subdir))) - self._box = mailbox.Maildir(self._path) - self._check_basics(factory=rfc822.Message) - self._box = mailbox.Maildir(self._path, factory=None) - self._check_basics() - - def _check_basics(self, factory=None): - # (Used by test_open_new() and test_open_existing().) - self.assertEqual(self._box._path, os.path.abspath(self._path)) - self.assertEqual(self._box._factory, factory) - for subdir in '', 'tmp', 'new', 'cur': - path = os.path.join(self._path, subdir) - mode = os.stat(path)[stat.ST_MODE] - self.assertTrue(stat.S_ISDIR(mode), "Not a directory: '%s'" % path) - - def test_list_folders(self): - # List folders - self._box.add_folder('one') - self._box.add_folder('two') - self._box.add_folder('three') - self.assertEqual(len(self._box.list_folders()), 3) - self.assertEqual(set(self._box.list_folders()), - set(('one', 'two', 'three'))) - - def test_get_folder(self): - # Open folders - self._box.add_folder('foo.bar') - folder0 = self._box.get_folder('foo.bar') - folder0.add(self._template % 'bar') - self.assertTrue(os.path.isdir(os.path.join(self._path, '.foo.bar'))) - folder1 = self._box.get_folder('foo.bar') - self.assertEqual(folder1.get_string(folder1.keys()[0]), - self._template % 'bar') - - def test_add_and_remove_folders(self): - # Delete folders - self._box.add_folder('one') - self._box.add_folder('two') - self.assertEqual(len(self._box.list_folders()), 2) - self.assertEqual(set(self._box.list_folders()), set(('one', 'two'))) - self._box.remove_folder('one') - self.assertEqual(len(self._box.list_folders()), 1) - self.assertEqual(set(self._box.list_folders()), set(('two',))) - self._box.add_folder('three') - self.assertEqual(len(self._box.list_folders()), 2) - self.assertEqual(set(self._box.list_folders()), set(('two', 'three'))) - self._box.remove_folder('three') - self.assertEqual(len(self._box.list_folders()), 1) - self.assertEqual(set(self._box.list_folders()), set(('two',))) - self._box.remove_folder('two') - self.assertEqual(len(self._box.list_folders()), 0) - self.assertEqual(self._box.list_folders(), []) - - def test_clean(self): - # Remove old files from 'tmp' - foo_path = os.path.join(self._path, 'tmp', 'foo') - bar_path = os.path.join(self._path, 'tmp', 'bar') - 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)) - foo_stat = os.stat(foo_path) - os.utime(foo_path, (time.time() - 129600 - 2, - foo_stat.st_mtime)) - self._box.clean() - self.assertFalse(os.path.exists(foo_path)) - self.assertTrue(os.path.exists(bar_path)) - - def test_create_tmp(self, repetitions=10): - # Create files in tmp directory - hostname = socket.gethostname() - if '/' in hostname: - hostname = hostname.replace('/', r'\057') - if ':' in hostname: - hostname = hostname.replace(':', r'\072') - pid = os.getpid() - pattern = re.compile(r"(?P<time>\d+)\.M(?P<M>\d{1,6})P(?P<P>\d+)" - r"Q(?P<Q>\d+)\.(?P<host>[^:/]+)") - previous_groups = None - for x in xrange(repetitions): - tmp_file = self._box._create_tmp() - head, tail = os.path.split(tmp_file.name) - self.assertEqual(head, os.path.abspath(os.path.join(self._path, - "tmp")), - "File in wrong location: '%s'" % head) - match = pattern.match(tail) - self.assertTrue(match is not None, "Invalid file name: '%s'" % tail) - groups = match.groups() - if previous_groups is not None: - self.assertGreaterEqual(int(groups[0]), int(previous_groups[0]), - "Non-monotonic seconds: '%s' before '%s'" % - (previous_groups[0], groups[0])) - if int(groups[0]) == int(previous_groups[0]): - self.assertGreaterEqual(int(groups[1]), int(previous_groups[1]), - "Non-monotonic milliseconds: '%s' before '%s'" % - (previous_groups[1], groups[1])) - self.assertTrue(int(groups[2]) == pid, - "Process ID mismatch: '%s' should be '%s'" % - (groups[2], pid)) - self.assertTrue(int(groups[3]) == int(previous_groups[3]) + 1, - "Non-sequential counter: '%s' before '%s'" % - (previous_groups[3], groups[3])) - self.assertTrue(groups[4] == hostname, - "Host name mismatch: '%s' should be '%s'" % - (groups[4], hostname)) - previous_groups = groups - tmp_file.write(_sample_message) - tmp_file.seek(0) - self.assertTrue(tmp_file.read() == _sample_message) - tmp_file.close() - file_count = len(os.listdir(os.path.join(self._path, "tmp"))) - self.assertTrue(file_count == repetitions, - "Wrong file count: '%s' should be '%s'" % - (file_count, repetitions)) - - def test_refresh(self): - # Update the table of contents - self.assertEqual(self._box._toc, {}) - key0 = self._box.add(self._template % 0) - key1 = self._box.add(self._template % 1) - self.assertEqual(self._box._toc, {}) - self._box._refresh() - self.assertEqual(self._box._toc, {key0: os.path.join('new', key0), - key1: os.path.join('new', key1)}) - key2 = self._box.add(self._template % 2) - self.assertEqual(self._box._toc, {key0: os.path.join('new', key0), - key1: os.path.join('new', key1)}) - self._box._refresh() - self.assertEqual(self._box._toc, {key0: os.path.join('new', key0), - key1: os.path.join('new', key1), - key2: os.path.join('new', key2)}) - - def test_refresh_after_safety_period(self): - # Issue #13254: Call _refresh after the "file system safety - # period" of 2 seconds has passed; _toc should still be - # updated because this is the first call to _refresh. - key0 = self._box.add(self._template % 0) - key1 = self._box.add(self._template % 1) - - self._box = self._factory(self._path) - self.assertEqual(self._box._toc, {}) - - # Emulate sleeping. Instead of sleeping for 2 seconds, use the - # skew factor to make _refresh think that the filesystem - # safety period has passed and re-reading the _toc is only - # required if mtimes differ. - self._box._skewfactor = -3 - - self._box._refresh() - self.assertEqual(sorted(self._box._toc.keys()), sorted([key0, key1])) - - def test_lookup(self): - # Look up message subpaths in the TOC - self.assertRaises(KeyError, lambda: self._box._lookup('foo')) - key0 = self._box.add(self._template % 0) - self.assertEqual(self._box._lookup(key0), os.path.join('new', key0)) - os.remove(os.path.join(self._path, 'new', key0)) - self.assertEqual(self._box._toc, {key0: os.path.join('new', key0)}) - # Be sure that the TOC is read back from disk (see issue #6896 - # about bad mtime behaviour on some systems). - self._box.flush() - self.assertRaises(KeyError, lambda: self._box._lookup(key0)) - self.assertEqual(self._box._toc, {}) - - def test_lock_unlock(self): - # Lock and unlock the mailbox. For Maildir, this does nothing. - self._box.lock() - self._box.unlock() - - def test_folder (self): - # Test for bug #1569790: verify that folders returned by .get_folder() - # use the same factory function. - def dummy_factory (s): - return None - box = self._factory(self._path, factory=dummy_factory) - folder = box.add_folder('folder1') - self.assertIs(folder._factory, dummy_factory) - - folder1_alias = box.get_folder('folder1') - self.assertIs(folder1_alias._factory, dummy_factory) - - def test_directory_in_folder (self): - # Test that mailboxes still work if there's a stray extra directory - # in a folder. - for i in range(10): - self._box.add(mailbox.Message(_sample_message)) - - # Create a stray directory - os.mkdir(os.path.join(self._path, 'cur', 'stray-dir')) - - # Check that looping still works with the directory present. - for msg in self._box: - pass - - def test_file_permissions(self): - # Verify that message files are created without execute permissions - if not hasattr(os, "stat") or not hasattr(os, "umask"): - return - msg = mailbox.MaildirMessage(self._template % 0) - orig_umask = os.umask(0) - try: - key = self._box.add(msg) - finally: - os.umask(orig_umask) - path = os.path.join(self._path, self._box._lookup(key)) - mode = os.stat(path).st_mode - self.assertEqual(mode & 0111, 0) - - def test_folder_file_perms(self): - # From bug #3228, we want to verify that the file created inside a Maildir - # subfolder isn't marked as executable. - if not hasattr(os, "stat") or not hasattr(os, "umask"): - return - - orig_umask = os.umask(0) - try: - subfolder = self._box.add_folder('subfolder') - finally: - os.umask(orig_umask) - - path = os.path.join(subfolder._path, 'maildirfolder') - st = os.stat(path) - perms = st.st_mode - self.assertFalse((perms & 0111)) # Execute bits should all be off. - - def test_reread(self): - # Do an initial unconditional refresh - self._box._refresh() - - # Put the last modified times more than two seconds into the past - # (because mtime may have only a two second granularity). - for subdir in ('cur', 'new'): - os.utime(os.path.join(self._box._path, subdir), - (time.time()-5,)*2) - - # Because mtime has a two second granularity in worst case (FAT), a - # refresh is done unconditionally if called for within - # two-second-plus-a-bit of the last one, just in case the mbox has - # changed; so now we have to wait for that interval to expire. - # - # Because this is a test, emulate sleeping. Instead of - # sleeping for 2 seconds, use the skew factor to make _refresh - # think that 2 seconds have passed and re-reading the _toc is - # only required if mtimes differ. - self._box._skewfactor = -3 - - # Re-reading causes the ._toc attribute to be assigned a new dictionary - # object, so we'll check that the ._toc attribute isn't a different - # object. - orig_toc = self._box._toc - def refreshed(): - return self._box._toc is not orig_toc - - self._box._refresh() - self.assertFalse(refreshed()) - - # Now, write something into cur and remove it. This changes - # the mtime and should cause a re-read. Note that "sleep - # emulation" is still in effect, as skewfactor is -3. - filename = os.path.join(self._path, 'cur', 'stray-file') - f = open(filename, 'w') - f.close() - os.unlink(filename) - self._box._refresh() - self.assertTrue(refreshed()) - - -class _TestSingleFile(TestMailbox): - '''Common tests for single-file mailboxes''' - - def test_add_doesnt_rewrite(self): - # When only adding messages, flush() should not rewrite the - # mailbox file. See issue #9559. - - # Inode number changes if the contents are written to another - # file which is then renamed over the original file. So we - # must check that the inode number doesn't change. - inode_before = os.stat(self._path).st_ino - - self._box.add(self._template % 0) - self._box.flush() - - inode_after = os.stat(self._path).st_ino - self.assertEqual(inode_before, inode_after) - - # Make sure the message was really added - self._box.close() - self._box = self._factory(self._path) - self.assertEqual(len(self._box), 1) - - def test_permissions_after_flush(self): - # See issue #5346 - - # Make the mailbox world writable. It's unlikely that the new - # mailbox file would have these permissions after flush(), - # because umask usually prevents it. - mode = os.stat(self._path).st_mode | 0o666 - os.chmod(self._path, mode) - - self._box.add(self._template % 0) - i = self._box.add(self._template % 1) - # Need to remove one message to make flush() create a new file - self._box.remove(i) - self._box.flush() - - self.assertEqual(os.stat(self._path).st_mode, mode) - - -class _TestMboxMMDF(_TestSingleFile): - - def tearDown(self): - self._box.close() - self._delete_recursively(self._path) - for lock_remnant in glob.glob(self._path + '.*'): - test_support.unlink(lock_remnant) - - 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\n') - self.assertEqual(self._box[key].get_from(), 'foo@bar blah') - self.assertEqual(self._box[key].get_payload(), '0\n') - - def test_add_mbox_or_mmdf_message(self): - # Add an mboxMessage or MMDFMessage - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg = class_('From foo@bar blah\nFrom: foo\n\n0\n') - key = self._box.add(msg) - - def test_open_close_open(self): - # Open and inspect previously-created mailbox - values = [self._template % i for i in xrange(3)] - for value in values: - self._box.add(value) - self._box.close() - mtime = os.path.getmtime(self._path) - self._box = self._factory(self._path) - self.assertEqual(len(self._box), 3) - for key in self._box.iterkeys(): - self.assertIn(self._box.get_string(key), values) - self._box.close() - self.assertEqual(mtime, os.path.getmtime(self._path)) - - def test_add_and_close(self): - # Verifying that closing a mailbox doesn't change added items - self._box.add(_sample_message) - for i in xrange(3): - self._box.add(self._template % i) - self._box.add(_sample_message) - self._box._file.flush() - self._box._file.seek(0) - contents = self._box._file.read() - self._box.close() - with open(self._path, 'rb') as f: - self.assertEqual(contents, f.read()) - self._box = self._factory(self._path) - - @unittest.skipUnless(hasattr(os, 'fork'), "Test needs fork().") - @unittest.skipUnless(hasattr(socket, 'socketpair'), "Test needs socketpair().") - def test_lock_conflict(self): - # Fork off a child process that will lock the mailbox temporarily, - # unlock it and exit. - c, p = socket.socketpair() - self.addCleanup(c.close) - self.addCleanup(p.close) - - pid = os.fork() - if pid == 0: - # child - try: - # lock the mailbox, and signal the parent it can proceed - self._box.lock() - c.send(b'c') - - # wait until the parent is done, and unlock the mailbox - c.recv(1) - self._box.unlock() - finally: - os._exit(0) - - # In the parent, wait until the child signals it locked the mailbox. - p.recv(1) - try: - self.assertRaises(mailbox.ExternalClashError, - self._box.lock) - finally: - # Signal the child it can now release the lock and exit. - p.send(b'p') - # Wait for child to exit. Locking should now succeed. - exited_pid, status = os.waitpid(pid, 0) - - self._box.lock() - self._box.unlock() - - def test_relock(self): - # Test case for bug #1575506: the mailbox class was locking the - # wrong file object in its flush() method. - msg = "Subject: sub\n\nbody\n" - key1 = self._box.add(msg) - self._box.flush() - self._box.close() - - self._box = self._factory(self._path) - self._box.lock() - key2 = self._box.add(msg) - self._box.flush() - self.assertTrue(self._box._locked) - self._box.close() - - -class TestMbox(_TestMboxMMDF, unittest.TestCase): - - _factory = lambda self, path, factory=None: mailbox.mbox(path, factory) - - def test_file_perms(self): - # From bug #3228, we want to verify that the mailbox file isn't executable, - # even if the umask is set to something that would leave executable bits set. - # We only run this test on platforms that support umask. - if hasattr(os, 'umask') and hasattr(os, 'stat'): - try: - old_umask = os.umask(0077) - self._box.close() - os.unlink(self._path) - self._box = mailbox.mbox(self._path, create=True) - self._box.add('') - self._box.close() - finally: - os.umask(old_umask) - - st = os.stat(self._path) - perms = st.st_mode - self.assertFalse((perms & 0111)) # Execute bits should all be off. - - def test_terminating_newline(self): - message = email.message.Message() - message['From'] = 'john@example.com' - message.set_payload('No newline at the end') - i = self._box.add(message) - - # A newline should have been appended to the payload - message = self._box.get(i) - self.assertEqual(message.get_payload(), 'No newline at the end\n') - - def test_message_separator(self): - # Check there's always a single blank line after each message - self._box.add('From: foo\n\n0') # No newline at the end - with open(self._path) as f: - data = f.read() - self.assertEqual(data[-3:], '0\n\n') - - self._box.add('From: foo\n\n0\n') # Newline at the end - with open(self._path) as f: - data = f.read() - self.assertEqual(data[-3:], '0\n\n') - - -class TestMMDF(_TestMboxMMDF, unittest.TestCase): - - _factory = lambda self, path, factory=None: mailbox.MMDF(path, factory) - - -class TestMH(TestMailbox, unittest.TestCase): - - _factory = lambda self, path, factory=None: mailbox.MH(path, factory) - - def test_list_folders(self): - # List folders - self._box.add_folder('one') - self._box.add_folder('two') - self._box.add_folder('three') - self.assertEqual(len(self._box.list_folders()), 3) - self.assertEqual(set(self._box.list_folders()), - set(('one', 'two', 'three'))) - - def test_get_folder(self): - # Open folders - def dummy_factory (s): - return None - self._box = self._factory(self._path, dummy_factory) - - new_folder = self._box.add_folder('foo.bar') - folder0 = self._box.get_folder('foo.bar') - folder0.add(self._template % 'bar') - self.assertTrue(os.path.isdir(os.path.join(self._path, 'foo.bar'))) - folder1 = self._box.get_folder('foo.bar') - self.assertEqual(folder1.get_string(folder1.keys()[0]), - self._template % 'bar') - - # Test for bug #1569790: verify that folders returned by .get_folder() - # use the same factory function. - self.assertIs(new_folder._factory, self._box._factory) - self.assertIs(folder0._factory, self._box._factory) - - def test_add_and_remove_folders(self): - # Delete folders - self._box.add_folder('one') - self._box.add_folder('two') - self.assertEqual(len(self._box.list_folders()), 2) - self.assertEqual(set(self._box.list_folders()), set(('one', 'two'))) - self._box.remove_folder('one') - self.assertEqual(len(self._box.list_folders()), 1) - self.assertEqual(set(self._box.list_folders()), set(('two', ))) - self._box.add_folder('three') - self.assertEqual(len(self._box.list_folders()), 2) - self.assertEqual(set(self._box.list_folders()), set(('two', 'three'))) - self._box.remove_folder('three') - self.assertEqual(len(self._box.list_folders()), 1) - self.assertEqual(set(self._box.list_folders()), set(('two', ))) - self._box.remove_folder('two') - self.assertEqual(len(self._box.list_folders()), 0) - self.assertEqual(self._box.list_folders(), []) - - def test_sequences(self): - # Get and set sequences - self.assertEqual(self._box.get_sequences(), {}) - msg0 = mailbox.MHMessage(self._template % 0) - msg0.add_sequence('foo') - key0 = self._box.add(msg0) - self.assertEqual(self._box.get_sequences(), {'foo':[key0]}) - msg1 = mailbox.MHMessage(self._template % 1) - msg1.set_sequences(['bar', 'replied', 'foo']) - key1 = self._box.add(msg1) - self.assertEqual(self._box.get_sequences(), - {'foo':[key0, key1], 'bar':[key1], 'replied':[key1]}) - msg0.set_sequences(['flagged']) - self._box[key0] = msg0 - self.assertEqual(self._box.get_sequences(), - {'foo':[key1], 'bar':[key1], 'replied':[key1], - 'flagged':[key0]}) - self._box.remove(key1) - self.assertEqual(self._box.get_sequences(), {'flagged':[key0]}) - - def test_issue2625(self): - msg0 = mailbox.MHMessage(self._template % 0) - msg0.add_sequence('foo') - 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) - msg1 = mailbox.MHMessage(self._template % 1) - msg2 = mailbox.MHMessage(self._template % 2) - msg3 = mailbox.MHMessage(self._template % 3) - msg0.set_sequences(['foo', 'unseen']) - msg1.set_sequences(['foo']) - msg2.set_sequences(['foo', 'flagged']) - msg3.set_sequences(['foo', 'bar', 'replied']) - key0 = self._box.add(msg0) - key1 = self._box.add(msg1) - key2 = self._box.add(msg2) - key3 = self._box.add(msg3) - self.assertEqual(self._box.get_sequences(), - {'foo':[key0,key1,key2,key3], 'unseen':[key0], - 'flagged':[key2], 'bar':[key3], 'replied':[key3]}) - self._box.remove(key2) - self.assertEqual(self._box.get_sequences(), - {'foo':[key0,key1,key3], 'unseen':[key0], 'bar':[key3], - 'replied':[key3]}) - self._box.pack() - self.assertEqual(self._box.keys(), [1, 2, 3]) - key0 = key0 - key1 = key0 + 1 - key2 = key1 + 1 - self.assertEqual(self._box.get_sequences(), - {'foo':[1, 2, 3], 'unseen':[1], 'bar':[3], 'replied':[3]}) - - # Test case for packing while holding the mailbox locked. - key0 = self._box.add(msg1) - key1 = self._box.add(msg1) - key2 = self._box.add(msg1) - key3 = self._box.add(msg1) - - self._box.remove(key0) - self._box.remove(key2) - self._box.lock() - self._box.pack() - self._box.unlock() - self.assertEqual(self._box.get_sequences(), - {'foo':[1, 2, 3, 4, 5], - 'unseen':[1], 'bar':[3], 'replied':[3]}) - - def _get_lock_path(self): - return os.path.join(self._path, '.mh_sequences.lock') - - -class TestBabyl(_TestSingleFile, unittest.TestCase): - - _factory = lambda self, path, factory=None: mailbox.Babyl(path, factory) - - def tearDown(self): - self._box.close() - self._delete_recursively(self._path) - for lock_remnant in glob.glob(self._path + '.*'): - test_support.unlink(lock_remnant) - - def test_labels(self): - # Get labels from the mailbox - self.assertEqual(self._box.get_labels(), []) - msg0 = mailbox.BabylMessage(self._template % 0) - msg0.add_label('foo') - key0 = self._box.add(msg0) - self.assertEqual(self._box.get_labels(), ['foo']) - msg1 = mailbox.BabylMessage(self._template % 1) - msg1.set_labels(['bar', 'answered', 'foo']) - key1 = self._box.add(msg1) - self.assertEqual(set(self._box.get_labels()), set(['foo', 'bar'])) - msg0.set_labels(['blah', 'filed']) - self._box[key0] = msg0 - self.assertEqual(set(self._box.get_labels()), - set(['foo', 'bar', 'blah'])) - self._box.remove(key1) - self.assertEqual(set(self._box.get_labels()), set(['blah'])) - - -class TestMessage(TestBase, unittest.TestCase): - - _factory = mailbox.Message # Overridden by subclasses to reuse tests - - def setUp(self): - self._path = test_support.TESTFN - - def tearDown(self): - self._delete_recursively(self._path) - - def test_initialize_with_eMM(self): - # Initialize based on email.message.Message instance - eMM = email.message_from_string(_sample_message) - msg = self._factory(eMM) - self._post_initialize_hook(msg) - self._check_sample(msg) - - def test_initialize_with_string(self): - # Initialize based on string - msg = self._factory(_sample_message) - self._post_initialize_hook(msg) - self._check_sample(msg) - - def test_initialize_with_file(self): - # Initialize based on contents of file - 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_nothing(self): - # Initialize without arguments - msg = self._factory() - self._post_initialize_hook(msg) - 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) - - def test_initialize_incorrectly(self): - # Initialize with invalid argument - self.assertRaises(TypeError, lambda: self._factory(object())) - - def test_become_message(self): - # Take on the state of another message - eMM = email.message_from_string(_sample_message) - msg = self._factory() - msg._become_message(eMM) - self._check_sample(msg) - - def test_explain_to(self): - # Copy self's format-specific data to other message formats. - # This test is superficial; better ones are in TestMessageConversion. - msg = self._factory() - for class_ in (mailbox.Message, mailbox.MaildirMessage, - mailbox.mboxMessage, mailbox.MHMessage, - mailbox.BabylMessage, mailbox.MMDFMessage): - other_msg = class_() - msg._explain_to(other_msg) - other_msg = email.message.Message() - self.assertRaises(TypeError, lambda: msg._explain_to(other_msg)) - - def _post_initialize_hook(self, msg): - # Overridden by subclasses to check extra things after initialization - pass - - -class TestMaildirMessage(TestMessage, unittest.TestCase): - - _factory = mailbox.MaildirMessage - - def _post_initialize_hook(self, msg): - self.assertEqual(msg._subdir, 'new') - self.assertEqual(msg._info,'') - - def test_subdir(self): - # Use get_subdir() and set_subdir() - msg = mailbox.MaildirMessage(_sample_message) - self.assertEqual(msg.get_subdir(), 'new') - msg.set_subdir('cur') - self.assertEqual(msg.get_subdir(), 'cur') - msg.set_subdir('new') - self.assertEqual(msg.get_subdir(), 'new') - self.assertRaises(ValueError, lambda: msg.set_subdir('tmp')) - self.assertEqual(msg.get_subdir(), 'new') - msg.set_subdir('new') - self.assertEqual(msg.get_subdir(), 'new') - self._check_sample(msg) - - def test_flags(self): - # Use get_flags(), set_flags(), add_flag(), remove_flag() - msg = mailbox.MaildirMessage(_sample_message) - self.assertEqual(msg.get_flags(), '') - self.assertEqual(msg.get_subdir(), 'new') - msg.set_flags('F') - self.assertEqual(msg.get_subdir(), 'new') - self.assertEqual(msg.get_flags(), 'F') - msg.set_flags('SDTP') - self.assertEqual(msg.get_flags(), 'DPST') - msg.add_flag('FT') - self.assertEqual(msg.get_flags(), 'DFPST') - msg.remove_flag('TDRP') - self.assertEqual(msg.get_flags(), 'FS') - self.assertEqual(msg.get_subdir(), 'new') - self._check_sample(msg) - - def test_date(self): - # Use get_date() and set_date() - msg = mailbox.MaildirMessage(_sample_message) - diff = msg.get_date() - time.time() - self.assertTrue(abs(diff) < 60, diff) - msg.set_date(0.0) - self.assertEqual(msg.get_date(), 0.0) - - def test_info(self): - # Use get_info() and set_info() - msg = mailbox.MaildirMessage(_sample_message) - self.assertEqual(msg.get_info(), '') - msg.set_info('1,foo=bar') - self.assertEqual(msg.get_info(), '1,foo=bar') - self.assertRaises(TypeError, lambda: msg.set_info(None)) - self._check_sample(msg) - - def test_info_and_flags(self): - # Test interaction of info and flag methods - msg = mailbox.MaildirMessage(_sample_message) - self.assertEqual(msg.get_info(), '') - msg.set_flags('SF') - self.assertEqual(msg.get_flags(), 'FS') - self.assertEqual(msg.get_info(), '2,FS') - msg.set_info('1,') - self.assertEqual(msg.get_flags(), '') - self.assertEqual(msg.get_info(), '1,') - msg.remove_flag('RPT') - self.assertEqual(msg.get_flags(), '') - self.assertEqual(msg.get_info(), '1,') - msg.add_flag('D') - self.assertEqual(msg.get_flags(), 'D') - self.assertEqual(msg.get_info(), '2,D') - self._check_sample(msg) - - -class _TestMboxMMDFMessage: - - _factory = mailbox._mboxMMDFMessage - - def _post_initialize_hook(self, msg): - self._check_from(msg) - - def test_initialize_with_unixfrom(self): - # Initialize with a message that already has a _unixfrom attribute - msg = mailbox.Message(_sample_message) - msg.set_unixfrom('From foo@bar blah') - msg = mailbox.mboxMessage(msg) - self.assertEqual(msg.get_from(), 'foo@bar blah') - - def test_from(self): - # Get and set "From " line - msg = mailbox.mboxMessage(_sample_message) - self._check_from(msg) - msg.set_from('foo bar') - self.assertEqual(msg.get_from(), 'foo bar') - msg.set_from('foo@bar', True) - self._check_from(msg, 'foo@bar') - msg.set_from('blah@temp', time.localtime()) - self._check_from(msg, 'blah@temp') - - def test_flags(self): - # Use get_flags(), set_flags(), add_flag(), remove_flag() - msg = mailbox.mboxMessage(_sample_message) - self.assertEqual(msg.get_flags(), '') - msg.set_flags('F') - self.assertEqual(msg.get_flags(), 'F') - msg.set_flags('XODR') - self.assertEqual(msg.get_flags(), 'RODX') - msg.add_flag('FA') - self.assertEqual(msg.get_flags(), 'RODFAX') - msg.remove_flag('FDXA') - self.assertEqual(msg.get_flags(), 'RO') - self._check_sample(msg) - - def _check_from(self, msg, sender=None): - # Check contents of "From " line - if sender is None: - sender = "MAILER-DAEMON" - self.assertTrue(re.match(sender + r" \w{3} \w{3} [\d ]\d [\d ]\d:\d{2}:" - r"\d{2} \d{4}", msg.get_from())) - - -class TestMboxMessage(_TestMboxMMDFMessage, TestMessage): - - _factory = mailbox.mboxMessage - - -class TestMHMessage(TestMessage, unittest.TestCase): - - _factory = mailbox.MHMessage - - def _post_initialize_hook(self, msg): - self.assertEqual(msg._sequences, []) - - def test_sequences(self): - # Get, set, join, and leave sequences - msg = mailbox.MHMessage(_sample_message) - self.assertEqual(msg.get_sequences(), []) - msg.set_sequences(['foobar']) - self.assertEqual(msg.get_sequences(), ['foobar']) - msg.set_sequences([]) - self.assertEqual(msg.get_sequences(), []) - msg.add_sequence('unseen') - self.assertEqual(msg.get_sequences(), ['unseen']) - msg.add_sequence('flagged') - self.assertEqual(msg.get_sequences(), ['unseen', 'flagged']) - msg.add_sequence('flagged') - self.assertEqual(msg.get_sequences(), ['unseen', 'flagged']) - msg.remove_sequence('unseen') - self.assertEqual(msg.get_sequences(), ['flagged']) - msg.add_sequence('foobar') - self.assertEqual(msg.get_sequences(), ['flagged', 'foobar']) - msg.remove_sequence('replied') - self.assertEqual(msg.get_sequences(), ['flagged', 'foobar']) - msg.set_sequences(['foobar', 'replied']) - self.assertEqual(msg.get_sequences(), ['foobar', 'replied']) - - -class TestBabylMessage(TestMessage, unittest.TestCase): - - _factory = mailbox.BabylMessage - - def _post_initialize_hook(self, msg): - self.assertEqual(msg._labels, []) - - def test_labels(self): - # Get, set, join, and leave labels - msg = mailbox.BabylMessage(_sample_message) - self.assertEqual(msg.get_labels(), []) - msg.set_labels(['foobar']) - self.assertEqual(msg.get_labels(), ['foobar']) - msg.set_labels([]) - self.assertEqual(msg.get_labels(), []) - msg.add_label('filed') - self.assertEqual(msg.get_labels(), ['filed']) - msg.add_label('resent') - self.assertEqual(msg.get_labels(), ['filed', 'resent']) - msg.add_label('resent') - self.assertEqual(msg.get_labels(), ['filed', 'resent']) - msg.remove_label('filed') - self.assertEqual(msg.get_labels(), ['resent']) - msg.add_label('foobar') - self.assertEqual(msg.get_labels(), ['resent', 'foobar']) - msg.remove_label('unseen') - self.assertEqual(msg.get_labels(), ['resent', 'foobar']) - msg.set_labels(['foobar', 'answered']) - self.assertEqual(msg.get_labels(), ['foobar', 'answered']) - - def test_visible(self): - # Get, set, and update visible headers - msg = mailbox.BabylMessage(_sample_message) - visible = msg.get_visible() - self.assertEqual(visible.keys(), []) - self.assertIs(visible.get_payload(), None) - visible['User-Agent'] = 'FooBar 1.0' - visible['X-Whatever'] = 'Blah' - self.assertEqual(msg.get_visible().keys(), []) - msg.set_visible(visible) - visible = msg.get_visible() - self.assertEqual(visible.keys(), ['User-Agent', 'X-Whatever']) - self.assertEqual(visible['User-Agent'], 'FooBar 1.0') - self.assertEqual(visible['X-Whatever'], 'Blah') - self.assertIs(visible.get_payload(), None) - msg.update_visible() - self.assertEqual(visible.keys(), ['User-Agent', 'X-Whatever']) - self.assertIs(visible.get_payload(), None) - visible = msg.get_visible() - self.assertEqual(visible.keys(), ['User-Agent', 'Date', 'From', 'To', - 'Subject']) - for header in ('User-Agent', 'Date', 'From', 'To', 'Subject'): - self.assertEqual(visible[header], msg[header]) - - -class TestMMDFMessage(_TestMboxMMDFMessage, TestMessage): - - _factory = mailbox.MMDFMessage - - -class TestMessageConversion(TestBase, unittest.TestCase): - - def test_plain_to_x(self): - # Convert Message to all formats - for class_ in (mailbox.Message, mailbox.MaildirMessage, - mailbox.mboxMessage, mailbox.MHMessage, - mailbox.BabylMessage, mailbox.MMDFMessage): - msg_plain = mailbox.Message(_sample_message) - msg = class_(msg_plain) - self._check_sample(msg) - - def test_x_to_plain(self): - # Convert all formats to Message - for class_ in (mailbox.Message, mailbox.MaildirMessage, - mailbox.mboxMessage, mailbox.MHMessage, - mailbox.BabylMessage, mailbox.MMDFMessage): - msg = class_(_sample_message) - msg_plain = mailbox.Message(msg) - self._check_sample(msg_plain) - - def test_x_to_invalid(self): - # Convert all formats to an invalid format - for class_ in (mailbox.Message, mailbox.MaildirMessage, - mailbox.mboxMessage, mailbox.MHMessage, - mailbox.BabylMessage, mailbox.MMDFMessage): - self.assertRaises(TypeError, lambda: class_(False)) - - def test_maildir_to_maildir(self): - # Convert MaildirMessage to MaildirMessage - msg_maildir = mailbox.MaildirMessage(_sample_message) - msg_maildir.set_flags('DFPRST') - msg_maildir.set_subdir('cur') - date = msg_maildir.get_date() - msg = mailbox.MaildirMessage(msg_maildir) - self._check_sample(msg) - self.assertEqual(msg.get_flags(), 'DFPRST') - self.assertEqual(msg.get_subdir(), 'cur') - self.assertEqual(msg.get_date(), date) - - def test_maildir_to_mboxmmdf(self): - # Convert MaildirMessage to mboxmessage and MMDFMessage - pairs = (('D', ''), ('F', 'F'), ('P', ''), ('R', 'A'), ('S', 'R'), - ('T', 'D'), ('DFPRST', 'RDFA')) - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg_maildir = mailbox.MaildirMessage(_sample_message) - msg_maildir.set_date(0.0) - for setting, result in pairs: - msg_maildir.set_flags(setting) - msg = class_(msg_maildir) - self.assertEqual(msg.get_flags(), result) - self.assertEqual(msg.get_from(), 'MAILER-DAEMON %s' % - time.asctime(time.gmtime(0.0))) - msg_maildir.set_subdir('cur') - self.assertEqual(class_(msg_maildir).get_flags(), 'RODFA') - - def test_maildir_to_mh(self): - # Convert MaildirMessage to MHMessage - msg_maildir = mailbox.MaildirMessage(_sample_message) - pairs = (('D', ['unseen']), ('F', ['unseen', 'flagged']), - ('P', ['unseen']), ('R', ['unseen', 'replied']), ('S', []), - ('T', ['unseen']), ('DFPRST', ['replied', 'flagged'])) - for setting, result in pairs: - msg_maildir.set_flags(setting) - self.assertEqual(mailbox.MHMessage(msg_maildir).get_sequences(), - result) - - def test_maildir_to_babyl(self): - # Convert MaildirMessage to Babyl - msg_maildir = mailbox.MaildirMessage(_sample_message) - pairs = (('D', ['unseen']), ('F', ['unseen']), - ('P', ['unseen', 'forwarded']), ('R', ['unseen', 'answered']), - ('S', []), ('T', ['unseen', 'deleted']), - ('DFPRST', ['deleted', 'answered', 'forwarded'])) - for setting, result in pairs: - msg_maildir.set_flags(setting) - self.assertEqual(mailbox.BabylMessage(msg_maildir).get_labels(), - result) - - def test_mboxmmdf_to_maildir(self): - # Convert mboxMessage and MMDFMessage to MaildirMessage - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg_mboxMMDF = class_(_sample_message) - msg_mboxMMDF.set_from('foo@bar', time.gmtime(0.0)) - pairs = (('R', 'S'), ('O', ''), ('D', 'T'), ('F', 'F'), ('A', 'R'), - ('RODFA', 'FRST')) - for setting, result in pairs: - msg_mboxMMDF.set_flags(setting) - msg = mailbox.MaildirMessage(msg_mboxMMDF) - self.assertEqual(msg.get_flags(), result) - self.assertEqual(msg.get_date(), 0.0) - msg_mboxMMDF.set_flags('O') - self.assertEqual(mailbox.MaildirMessage(msg_mboxMMDF).get_subdir(), - 'cur') - - def test_mboxmmdf_to_mboxmmdf(self): - # Convert mboxMessage and MMDFMessage to mboxMessage and MMDFMessage - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg_mboxMMDF = class_(_sample_message) - msg_mboxMMDF.set_flags('RODFA') - msg_mboxMMDF.set_from('foo@bar') - for class2_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg2 = class2_(msg_mboxMMDF) - self.assertEqual(msg2.get_flags(), 'RODFA') - self.assertEqual(msg2.get_from(), 'foo@bar') - - def test_mboxmmdf_to_mh(self): - # Convert mboxMessage and MMDFMessage to MHMessage - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg_mboxMMDF = class_(_sample_message) - pairs = (('R', []), ('O', ['unseen']), ('D', ['unseen']), - ('F', ['unseen', 'flagged']), - ('A', ['unseen', 'replied']), - ('RODFA', ['replied', 'flagged'])) - for setting, result in pairs: - msg_mboxMMDF.set_flags(setting) - self.assertEqual(mailbox.MHMessage(msg_mboxMMDF).get_sequences(), - result) - - def test_mboxmmdf_to_babyl(self): - # Convert mboxMessage and MMDFMessage to BabylMessage - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg = class_(_sample_message) - pairs = (('R', []), ('O', ['unseen']), - ('D', ['unseen', 'deleted']), ('F', ['unseen']), - ('A', ['unseen', 'answered']), - ('RODFA', ['deleted', 'answered'])) - for setting, result in pairs: - msg.set_flags(setting) - self.assertEqual(mailbox.BabylMessage(msg).get_labels(), result) - - def test_mh_to_maildir(self): - # Convert MHMessage to MaildirMessage - pairs = (('unseen', ''), ('replied', 'RS'), ('flagged', 'FS')) - for setting, result in pairs: - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence(setting) - self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), result) - self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur') - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence('unseen') - msg.add_sequence('replied') - msg.add_sequence('flagged') - self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), 'FR') - self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur') - - def test_mh_to_mboxmmdf(self): - # Convert MHMessage to mboxMessage and MMDFMessage - pairs = (('unseen', 'O'), ('replied', 'ROA'), ('flagged', 'ROF')) - for setting, result in pairs: - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence(setting) - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - self.assertEqual(class_(msg).get_flags(), result) - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence('unseen') - msg.add_sequence('replied') - msg.add_sequence('flagged') - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - self.assertEqual(class_(msg).get_flags(), 'OFA') - - def test_mh_to_mh(self): - # Convert MHMessage to MHMessage - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence('unseen') - msg.add_sequence('replied') - msg.add_sequence('flagged') - self.assertEqual(mailbox.MHMessage(msg).get_sequences(), - ['unseen', 'replied', 'flagged']) - - def test_mh_to_babyl(self): - # Convert MHMessage to BabylMessage - pairs = (('unseen', ['unseen']), ('replied', ['answered']), - ('flagged', [])) - for setting, result in pairs: - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence(setting) - self.assertEqual(mailbox.BabylMessage(msg).get_labels(), result) - msg = mailbox.MHMessage(_sample_message) - msg.add_sequence('unseen') - msg.add_sequence('replied') - msg.add_sequence('flagged') - self.assertEqual(mailbox.BabylMessage(msg).get_labels(), - ['unseen', 'answered']) - - def test_babyl_to_maildir(self): - # Convert BabylMessage to MaildirMessage - pairs = (('unseen', ''), ('deleted', 'ST'), ('filed', 'S'), - ('answered', 'RS'), ('forwarded', 'PS'), ('edited', 'S'), - ('resent', 'PS')) - for setting, result in pairs: - msg = mailbox.BabylMessage(_sample_message) - msg.add_label(setting) - self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), result) - self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur') - msg = mailbox.BabylMessage(_sample_message) - for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded', - 'edited', 'resent'): - msg.add_label(label) - self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), 'PRT') - self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur') - - def test_babyl_to_mboxmmdf(self): - # Convert BabylMessage to mboxMessage and MMDFMessage - pairs = (('unseen', 'O'), ('deleted', 'ROD'), ('filed', 'RO'), - ('answered', 'ROA'), ('forwarded', 'RO'), ('edited', 'RO'), - ('resent', 'RO')) - for setting, result in pairs: - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - msg = mailbox.BabylMessage(_sample_message) - msg.add_label(setting) - self.assertEqual(class_(msg).get_flags(), result) - msg = mailbox.BabylMessage(_sample_message) - for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded', - 'edited', 'resent'): - msg.add_label(label) - for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage): - self.assertEqual(class_(msg).get_flags(), 'ODA') - - def test_babyl_to_mh(self): - # Convert BabylMessage to MHMessage - pairs = (('unseen', ['unseen']), ('deleted', []), ('filed', []), - ('answered', ['replied']), ('forwarded', []), ('edited', []), - ('resent', [])) - for setting, result in pairs: - msg = mailbox.BabylMessage(_sample_message) - msg.add_label(setting) - self.assertEqual(mailbox.MHMessage(msg).get_sequences(), result) - msg = mailbox.BabylMessage(_sample_message) - for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded', - 'edited', 'resent'): - msg.add_label(label) - self.assertEqual(mailbox.MHMessage(msg).get_sequences(), - ['unseen', 'replied']) - - def test_babyl_to_babyl(self): - # Convert BabylMessage to BabylMessage - msg = mailbox.BabylMessage(_sample_message) - msg.update_visible() - for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded', - 'edited', 'resent'): - msg.add_label(label) - msg2 = mailbox.BabylMessage(msg) - self.assertEqual(msg2.get_labels(), ['unseen', 'deleted', 'filed', - 'answered', 'forwarded', 'edited', - 'resent']) - self.assertEqual(msg.get_visible().keys(), msg2.get_visible().keys()) - for key in msg.get_visible().keys(): - self.assertEqual(msg.get_visible()[key], msg2.get_visible()[key]) - - -class TestProxyFileBase(TestBase): - - def _test_read(self, proxy): - # Read by byte - proxy.seek(0) - self.assertEqual(proxy.read(), 'bar') - proxy.seek(1) - self.assertEqual(proxy.read(), 'ar') - proxy.seek(0) - self.assertEqual(proxy.read(2), 'ba') - proxy.seek(1) - self.assertEqual(proxy.read(-1), 'ar') - proxy.seek(2) - self.assertEqual(proxy.read(1000), 'r') - - def _test_readline(self, proxy): - # Read by line - proxy.seek(0) - self.assertEqual(proxy.readline(), 'foo' + os.linesep) - self.assertEqual(proxy.readline(), 'bar' + os.linesep) - self.assertEqual(proxy.readline(), 'fred' + os.linesep) - self.assertEqual(proxy.readline(), 'bob') - proxy.seek(2) - self.assertEqual(proxy.readline(), 'o' + os.linesep) - proxy.seek(6 + 2 * len(os.linesep)) - self.assertEqual(proxy.readline(), 'fred' + os.linesep) - proxy.seek(6 + 2 * len(os.linesep)) - self.assertEqual(proxy.readline(2), 'fr') - self.assertEqual(proxy.readline(-10), 'ed' + os.linesep) - - def _test_readlines(self, proxy): - # Read multiple lines - proxy.seek(0) - self.assertEqual(proxy.readlines(), ['foo' + os.linesep, - 'bar' + os.linesep, - 'fred' + os.linesep, 'bob']) - proxy.seek(0) - self.assertEqual(proxy.readlines(2), ['foo' + os.linesep]) - proxy.seek(3 + len(os.linesep)) - self.assertEqual(proxy.readlines(4 + len(os.linesep)), - ['bar' + os.linesep, 'fred' + os.linesep]) - proxy.seek(3) - self.assertEqual(proxy.readlines(1000), [os.linesep, 'bar' + os.linesep, - 'fred' + os.linesep, 'bob']) - - def _test_iteration(self, proxy): - # Iterate by line - proxy.seek(0) - iterator = iter(proxy) - self.assertEqual(list(iterator), - ['foo' + os.linesep, 'bar' + os.linesep, 'fred' + os.linesep, 'bob']) - - def _test_seek_and_tell(self, proxy): - # Seek and use tell to check position - proxy.seek(3) - self.assertEqual(proxy.tell(), 3) - self.assertEqual(proxy.read(len(os.linesep)), os.linesep) - proxy.seek(2, 1) - self.assertEqual(proxy.read(1 + len(os.linesep)), 'r' + os.linesep) - proxy.seek(-3 - len(os.linesep), 2) - self.assertEqual(proxy.read(3), 'bar') - proxy.seek(2, 0) - self.assertEqual(proxy.read(), 'o' + os.linesep + 'bar' + os.linesep) - proxy.seek(100) - self.assertEqual(proxy.read(), '') - - def _test_close(self, proxy): - # Close a file - proxy.close() - # Issue 11700 subsequent closes should be a no-op, not an error. - proxy.close() - - -class TestProxyFile(TestProxyFileBase, unittest.TestCase): - - def setUp(self): - self._path = test_support.TESTFN - self._file = open(self._path, 'wb+') - - def tearDown(self): - self._file.close() - self._delete_recursively(self._path) - - def test_initialize(self): - # Initialize and check position - self._file.write('foo') - pos = self._file.tell() - proxy0 = mailbox._ProxyFile(self._file) - self.assertEqual(proxy0.tell(), pos) - self.assertEqual(self._file.tell(), pos) - proxy1 = mailbox._ProxyFile(self._file, 0) - self.assertEqual(proxy1.tell(), 0) - self.assertEqual(self._file.tell(), pos) - - def test_read(self): - self._file.write('bar') - self._test_read(mailbox._ProxyFile(self._file)) - - def test_readline(self): - self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep, - os.linesep)) - self._test_readline(mailbox._ProxyFile(self._file)) - - def test_readlines(self): - self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep, - os.linesep)) - self._test_readlines(mailbox._ProxyFile(self._file)) - - def test_iteration(self): - self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep, - os.linesep)) - self._test_iteration(mailbox._ProxyFile(self._file)) - - def test_seek_and_tell(self): - self._file.write('foo%sbar%s' % (os.linesep, os.linesep)) - self._test_seek_and_tell(mailbox._ProxyFile(self._file)) - - def test_close(self): - self._file.write('foo%sbar%s' % (os.linesep, os.linesep)) - self._test_close(mailbox._ProxyFile(self._file)) - - -class TestPartialFile(TestProxyFileBase, unittest.TestCase): - - def setUp(self): - self._path = test_support.TESTFN - self._file = open(self._path, 'wb+') - - def tearDown(self): - self._file.close() - self._delete_recursively(self._path) - - def test_initialize(self): - # Initialize and check position - self._file.write('foo' + os.linesep + 'bar') - pos = self._file.tell() - proxy = mailbox._PartialFile(self._file, 2, 5) - self.assertEqual(proxy.tell(), 0) - self.assertEqual(self._file.tell(), pos) - - def test_read(self): - self._file.write('***bar***') - self._test_read(mailbox._PartialFile(self._file, 3, 6)) - - def test_readline(self): - self._file.write('!!!!!foo%sbar%sfred%sbob!!!!!' % - (os.linesep, os.linesep, os.linesep)) - self._test_readline(mailbox._PartialFile(self._file, 5, - 18 + 3 * len(os.linesep))) - - def test_readlines(self): - self._file.write('foo%sbar%sfred%sbob?????' % - (os.linesep, os.linesep, os.linesep)) - self._test_readlines(mailbox._PartialFile(self._file, 0, - 13 + 3 * len(os.linesep))) - - def test_iteration(self): - self._file.write('____foo%sbar%sfred%sbob####' % - (os.linesep, os.linesep, os.linesep)) - self._test_iteration(mailbox._PartialFile(self._file, 4, - 17 + 3 * len(os.linesep))) - - def test_seek_and_tell(self): - self._file.write('(((foo%sbar%s$$$' % (os.linesep, os.linesep)) - self._test_seek_and_tell(mailbox._PartialFile(self._file, 3, - 9 + 2 * len(os.linesep))) - - def test_close(self): - self._file.write('&foo%sbar%s^' % (os.linesep, os.linesep)) - self._test_close(mailbox._PartialFile(self._file, 1, - 6 + 3 * len(os.linesep))) - - -## Start: tests from the original module (for backward compatibility). - -FROM_ = "From some.body@dummy.domain Sat Jul 24 13:43:35 2004\n" -DUMMY_MESSAGE = """\ -From: some.body@dummy.domain -To: me@my.domain -Subject: Simple Test - -This is a dummy message. -""" - -class MaildirTestCase(unittest.TestCase): - - def setUp(self): - # create a new maildir mailbox to work with: - self._dir = test_support.TESTFN - if os.path.isdir(self._dir): - test_support.rmtree(self._dir) - if os.path.isfile(self._dir): - test_support.unlink(self._dir) - os.mkdir(self._dir) - os.mkdir(os.path.join(self._dir, "cur")) - os.mkdir(os.path.join(self._dir, "tmp")) - os.mkdir(os.path.join(self._dir, "new")) - self._counter = 1 - self._msgfiles = [] - - def tearDown(self): - map(os.unlink, self._msgfiles) - test_support.rmdir(os.path.join(self._dir, "cur")) - test_support.rmdir(os.path.join(self._dir, "tmp")) - test_support.rmdir(os.path.join(self._dir, "new")) - test_support.rmdir(self._dir) - - def createMessage(self, dir, mbox=False): - t = int(time.time() % 1000000) - pid = self._counter - self._counter += 1 - filename = os.extsep.join((str(t), str(pid), "myhostname", "mydomain")) - tmpname = os.path.join(self._dir, "tmp", filename) - newname = os.path.join(self._dir, dir, filename) - 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: - with open(newname, "w") as fp: - fp.write(DUMMY_MESSAGE) - self._msgfiles.append(newname) - return tmpname - - def test_empty_maildir(self): - """Test an empty maildir mailbox""" - # Test for regression on bug #117490: - # Make sure the boxes attribute actually gets set. - self.mbox = mailbox.Maildir(test_support.TESTFN) - #self.assertTrue(hasattr(self.mbox, "boxes")) - #self.assertTrue(len(self.mbox.boxes) == 0) - self.assertIs(self.mbox.next(), None) - self.assertIs(self.mbox.next(), None) - - def test_nonempty_maildir_cur(self): - self.createMessage("cur") - self.mbox = mailbox.Maildir(test_support.TESTFN) - #self.assertTrue(len(self.mbox.boxes) == 1) - msg = self.mbox.next() - self.assertIsNot(msg, None) - msg.fp.close() - self.assertIs(self.mbox.next(), None) - self.assertIs(self.mbox.next(), None) - - def test_nonempty_maildir_new(self): - self.createMessage("new") - self.mbox = mailbox.Maildir(test_support.TESTFN) - #self.assertTrue(len(self.mbox.boxes) == 1) - msg = self.mbox.next() - self.assertIsNot(msg, None) - msg.fp.close() - self.assertIs(self.mbox.next(), None) - self.assertIs(self.mbox.next(), None) - - def test_nonempty_maildir_both(self): - self.createMessage("cur") - self.createMessage("new") - self.mbox = mailbox.Maildir(test_support.TESTFN) - #self.assertTrue(len(self.mbox.boxes) == 2) - msg = self.mbox.next() - self.assertIsNot(msg, None) - msg.fp.close() - msg = self.mbox.next() - self.assertIsNot(msg, None) - msg.fp.close() - self.assertIs(self.mbox.next(), None) - self.assertIs(self.mbox.next(), None) - - def test_unix_mbox(self): - ### should be better! - import email.parser - fname = self.createMessage("cur", True) - n = 0 - fid = open(fname) - for msg in mailbox.PortableUnixMailbox(fid, - email.parser.Parser().parse): - n += 1 - self.assertEqual(msg["subject"], "Simple Test") - self.assertEqual(len(str(msg)), len(FROM_)+len(DUMMY_MESSAGE)) - fid.close() - self.assertEqual(n, 1) - -## End: classes from the original module (for backward compatibility). - - -_sample_message = """\ -Return-Path: <gkj@gregorykjohnson.com> -X-Original-To: gkj+person@localhost -Delivered-To: gkj+person@localhost -Received: from localhost (localhost [127.0.0.1]) - by andy.gregorykjohnson.com (Postfix) with ESMTP id 356ED9DD17 - for <gkj+person@localhost>; Wed, 13 Jul 2005 17:23:16 -0400 (EDT) -Delivered-To: gkj@sundance.gregorykjohnson.com -Received: from localhost [127.0.0.1] - by localhost with POP3 (fetchmail-6.2.5) - for gkj+person@localhost (single-drop); Wed, 13 Jul 2005 17:23:16 -0400 (EDT) -Received: from andy.gregorykjohnson.com (andy.gregorykjohnson.com [64.32.235.228]) - by sundance.gregorykjohnson.com (Postfix) with ESMTP id 5B056316746 - for <gkj@gregorykjohnson.com>; Wed, 13 Jul 2005 17:23:11 -0400 (EDT) -Received: by andy.gregorykjohnson.com (Postfix, from userid 1000) - id 490CD9DD17; Wed, 13 Jul 2005 17:23:11 -0400 (EDT) -Date: Wed, 13 Jul 2005 17:23:11 -0400 -From: "Gregory K. Johnson" <gkj@gregorykjohnson.com> -To: gkj@gregorykjohnson.com -Subject: Sample message -Message-ID: <20050713212311.GC4701@andy.gregorykjohnson.com> -Mime-Version: 1.0 -Content-Type: multipart/mixed; boundary="NMuMz9nt05w80d4+" -Content-Disposition: inline -User-Agent: Mutt/1.5.9i - - ---NMuMz9nt05w80d4+ -Content-Type: text/plain; charset=us-ascii -Content-Disposition: inline - -This is a sample message. - --- -Gregory K. Johnson - ---NMuMz9nt05w80d4+ -Content-Type: application/octet-stream -Content-Disposition: attachment; filename="text.gz" -Content-Transfer-Encoding: base64 - -H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs -3FYlAAAA - ---NMuMz9nt05w80d4+-- -""" - -_sample_headers = { - "Return-Path":"<gkj@gregorykjohnson.com>", - "X-Original-To":"gkj+person@localhost", - "Delivered-To":"gkj+person@localhost", - "Received":"""from localhost (localhost [127.0.0.1]) - by andy.gregorykjohnson.com (Postfix) with ESMTP id 356ED9DD17 - for <gkj+person@localhost>; Wed, 13 Jul 2005 17:23:16 -0400 (EDT)""", - "Delivered-To":"gkj@sundance.gregorykjohnson.com", - "Received":"""from localhost [127.0.0.1] - by localhost with POP3 (fetchmail-6.2.5) - for gkj+person@localhost (single-drop); Wed, 13 Jul 2005 17:23:16 -0400 (EDT)""", - "Received":"""from andy.gregorykjohnson.com (andy.gregorykjohnson.com [64.32.235.228]) - by sundance.gregorykjohnson.com (Postfix) with ESMTP id 5B056316746 - for <gkj@gregorykjohnson.com>; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)""", - "Received":"""by andy.gregorykjohnson.com (Postfix, from userid 1000) - id 490CD9DD17; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)""", - "Date":"Wed, 13 Jul 2005 17:23:11 -0400", - "From":""""Gregory K. Johnson" <gkj@gregorykjohnson.com>""", - "To":"gkj@gregorykjohnson.com", - "Subject":"Sample message", - "Mime-Version":"1.0", - "Content-Type":"""multipart/mixed; boundary="NMuMz9nt05w80d4+\"""", - "Content-Disposition":"inline", - "User-Agent": "Mutt/1.5.9i" } - -_sample_payloads = ("""This is a sample message. - --- -Gregory K. Johnson -""", -"""H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs -3FYlAAAA -""") - - -def test_main(): - tests = (TestMailboxSuperclass, TestMaildir, TestMbox, TestMMDF, TestMH, - TestBabyl, TestMessage, TestMaildirMessage, TestMboxMessage, - TestMHMessage, TestBabylMessage, TestMMDFMessage, - TestMessageConversion, TestProxyFile, TestPartialFile, - MaildirTestCase) - test_support.run_unittest(*tests) - test_support.reap_children() - - -if __name__ == '__main__': - test_main() |