Lib/test/test_mailbox.py

   1 import os
   2 import sys
   3 import time
   4 import stat
   5 import socket
   6 import email
   7 import email.message
   8 import re
   9 import StringIO
  10 from test import test_support
  11 import unittest
  12 import mailbox
  13 import glob
  14 try:
  15     import fcntl
  16 except ImportError:
  17     pass
  18
  19 # Silence Py3k warning
  20 rfc822 = test_support.import_module('rfc822', deprecated=True)
  21
  22 class TestBase(unittest.TestCase):
  23
  24     def _check_sample(self, msg):
  25         # Inspect a mailbox.Message representation of the sample message
  26         self.assertIsInstance(msg, email.message.Message)
  27         self.assertIsInstance(msg, mailbox.Message)
  28         for key, value in _sample_headers.iteritems():
  29             self.assertIn(value, msg.get_all(key))
  30         self.assertTrue(msg.is_multipart())
  31         self.assertEqual(len(msg.get_payload()), len(_sample_payloads))
  32         for i, payload in enumerate(_sample_payloads):
  33             part = msg.get_payload(i)
  34             self.assertIsInstance(part, email.message.Message)
  35             self.assertNotIsInstance(part, mailbox.Message)
  36             self.assertEqual(part.get_payload(), payload)
  37
  38     def _delete_recursively(self, target):
  39         # Delete a file or delete a directory recursively
  40         if os.path.isdir(target):
  41             for path, dirs, files in os.walk(target, topdown=False):
  42                 for name in files:
  43                     os.remove(os.path.join(path, name))
  44                 for name in dirs:
  45                     os.rmdir(os.path.join(path, name))
  46             os.rmdir(target)
  47         elif os.path.exists(target):
  48             os.remove(target)
  49
  50
  51 class TestMailbox(TestBase):
  52
  53     _factory = None     # Overridden by subclasses to reuse tests
  54     _template = 'From: foo\n\n%s'
  55
  56     def setUp(self):
  57         self._path = test_support.TESTFN
  58         self._delete_recursively(self._path)
  59         self._box = self._factory(self._path)
  60
  61     def tearDown(self):
  62         self._box.close()
  63         self._delete_recursively(self._path)
  64
  65     def test_add(self):
  66         # Add copies of a sample message
  67         keys = []
  68         keys.append(self._box.add(self._template % 0))
  69         self.assertEqual(len(self._box), 1)
  70         keys.append(self._box.add(mailbox.Message(_sample_message)))
  71         self.assertEqual(len(self._box), 2)
  72         keys.append(self._box.add(email.message_from_string(_sample_message)))
  73         self.assertEqual(len(self._box), 3)
  74         keys.append(self._box.add(StringIO.StringIO(_sample_message)))
  75         self.assertEqual(len(self._box), 4)
  76         keys.append(self._box.add(_sample_message))
  77         self.assertEqual(len(self._box), 5)
  78         self.assertEqual(self._box.get_string(keys[0]), self._template % 0)
  79         for i in (1, 2, 3, 4):
  80             self._check_sample(self._box[keys[i]])
  81
  82     def test_remove(self):
  83         # Remove messages using remove()
  84         self._test_remove_or_delitem(self._box.remove)
  85
  86     def test_delitem(self):
  87         # Remove messages using __delitem__()
  88         self._test_remove_or_delitem(self._box.__delitem__)
  89
  90     def _test_remove_or_delitem(self, method):
  91         # (Used by test_remove() and test_delitem().)
  92         key0 = self._box.add(self._template % 0)
  93         key1 = self._box.add(self._template % 1)
  94         self.assertEqual(len(self._box), 2)
  95         method(key0)
  96         l = len(self._box)
  97         self.assertEqual(l, 1)
  98         self.assertRaises(KeyError, lambda: self._box[key0])
  99         self.assertRaises(KeyError, lambda: method(key0))
 100         self.assertEqual(self._box.get_string(key1), self._template % 1)
 101         key2 = self._box.add(self._template % 2)
 102         self.assertEqual(len(self._box), 2)
 103         method(key2)
 104         l = len(self._box)
 105         self.assertEqual(l, 1)
 106         self.assertRaises(KeyError, lambda: self._box[key2])
 107         self.assertRaises(KeyError, lambda: method(key2))
 108         self.assertEqual(self._box.get_string(key1), self._template % 1)
 109         method(key1)
 110         self.assertEqual(len(self._box), 0)
 111         self.assertRaises(KeyError, lambda: self._box[key1])
 112         self.assertRaises(KeyError, lambda: method(key1))
 113
 114     def test_discard(self, repetitions=10):
 115         # Discard messages
 116         key0 = self._box.add(self._template % 0)
 117         key1 = self._box.add(self._template % 1)
 118         self.assertEqual(len(self._box), 2)
 119         self._box.discard(key0)
 120         self.assertEqual(len(self._box), 1)
 121         self.assertRaises(KeyError, lambda: self._box[key0])
 122         self._box.discard(key0)
 123         self.assertEqual(len(self._box), 1)
 124         self.assertRaises(KeyError, lambda: self._box[key0])
 125
 126     def test_get(self):
 127         # Retrieve messages using get()
 128         key0 = self._box.add(self._template % 0)
 129         msg = self._box.get(key0)
 130         self.assertEqual(msg['from'], 'foo')
 131         self.assertEqual(msg.get_payload(), '0')
 132         self.assertIs(self._box.get('foo'), None)
 133         self.assertFalse(self._box.get('foo', False))
 134         self._box.close()
 135         self._box = self._factory(self._path, factory=rfc822.Message)
 136         key1 = self._box.add(self._template % 1)
 137         msg = self._box.get(key1)
 138         self.assertEqual(msg['from'], 'foo')
 139         self.assertEqual(msg.fp.read(), '1')
 140
 141     def test_getitem(self):
 142         # Retrieve message using __getitem__()
 143         key0 = self._box.add(self._template % 0)
 144         msg = self._box[key0]
 145         self.assertEqual(msg['from'], 'foo')
 146         self.assertEqual(msg.get_payload(), '0')
 147         self.assertRaises(KeyError, lambda: self._box['foo'])
 148         self._box.discard(key0)
 149         self.assertRaises(KeyError, lambda: self._box[key0])
 150
 151     def test_get_message(self):
 152         # Get Message representations of messages
 153         key0 = self._box.add(self._template % 0)
 154         key1 = self._box.add(_sample_message)
 155         msg0 = self._box.get_message(key0)
 156         self.assertIsInstance(msg0, mailbox.Message)
 157         self.assertEqual(msg0['from'], 'foo')
 158         self.assertEqual(msg0.get_payload(), '0')
 159         self._check_sample(self._box.get_message(key1))
 160
 161     def test_get_string(self):
 162         # Get string representations of messages
 163         key0 = self._box.add(self._template % 0)
 164         key1 = self._box.add(_sample_message)
 165         self.assertEqual(self._box.get_string(key0), self._template % 0)
 166         self.assertEqual(self._box.get_string(key1), _sample_message)
 167
 168     def test_get_file(self):
 169         # Get file representations of messages
 170         key0 = self._box.add(self._template % 0)
 171         key1 = self._box.add(_sample_message)
 172         self.assertEqual(self._box.get_file(key0).read().replace(os.linesep, '\n'),
 173                          self._template % 0)
 174         self.assertEqual(self._box.get_file(key1).read().replace(os.linesep, '\n'),
 175                          _sample_message)
 176
 177     def test_iterkeys(self):
 178         # Get keys using iterkeys()
 179         self._check_iteration(self._box.iterkeys, do_keys=True, do_values=False)
 180
 181     def test_keys(self):
 182         # Get keys using keys()
 183         self._check_iteration(self._box.keys, do_keys=True, do_values=False)
 184
 185     def test_itervalues(self):
 186         # Get values using itervalues()
 187         self._check_iteration(self._box.itervalues, do_keys=False,
 188                               do_values=True)
 189
 190     def test_iter(self):
 191         # Get values using __iter__()
 192         self._check_iteration(self._box.__iter__, do_keys=False,
 193                               do_values=True)
 194
 195     def test_values(self):
 196         # Get values using values()
 197         self._check_iteration(self._box.values, do_keys=False, do_values=True)
 198
 199     def test_iteritems(self):
 200         # Get keys and values using iteritems()
 201         self._check_iteration(self._box.iteritems, do_keys=True,
 202                               do_values=True)
 203
 204     def test_items(self):
 205         # Get keys and values using items()
 206         self._check_iteration(self._box.items, do_keys=True, do_values=True)
 207
 208     def _check_iteration(self, method, do_keys, do_values, repetitions=10):
 209         for value in method():
 210             self.fail("Not empty")
 211         keys, values = [], []
 212         for i in xrange(repetitions):
 213             keys.append(self._box.add(self._template % i))
 214             values.append(self._template % i)
 215         if do_keys and not do_values:
 216             returned_keys = list(method())
 217         elif do_values and not do_keys:
 218             returned_values = list(method())
 219         else:
 220             returned_keys, returned_values = [], []
 221             for key, value in method():
 222                 returned_keys.append(key)
 223                 returned_values.append(value)
 224         if do_keys:
 225             self.assertEqual(len(keys), len(returned_keys))
 226             self.assertEqual(set(keys), set(returned_keys))
 227         if do_values:
 228             count = 0
 229             for value in returned_values:
 230                 self.assertEqual(value['from'], 'foo')
 231                 self.assertTrue(int(value.get_payload()) < repetitions,
 232                                 (value.get_payload(), repetitions))
 233                 count += 1
 234             self.assertEqual(len(values), count)
 235
 236     def test_has_key(self):
 237         # Check existence of keys using has_key()
 238         self._test_has_key_or_contains(self._box.has_key)
 239
 240     def test_contains(self):
 241         # Check existence of keys using __contains__()
 242         self._test_has_key_or_contains(self._box.__contains__)
 243
 244     def _test_has_key_or_contains(self, method):
 245         # (Used by test_has_key() and test_contains().)
 246         self.assertFalse(method('foo'))
 247         key0 = self._box.add(self._template % 0)
 248         self.assertTrue(method(key0))
 249         self.assertFalse(method('foo'))
 250         key1 = self._box.add(self._template % 1)
 251         self.assertTrue(method(key1))
 252         self.assertTrue(method(key0))
 253         self.assertFalse(method('foo'))
 254         self._box.remove(key0)
 255         self.assertFalse(method(key0))
 256         self.assertTrue(method(key1))
 257         self.assertFalse(method('foo'))
 258         self._box.remove(key1)
 259         self.assertFalse(method(key1))
 260         self.assertFalse(method(key0))
 261         self.assertFalse(method('foo'))
 262
 263     def test_len(self, repetitions=10):
 264         # Get message count
 265         keys = []
 266         for i in xrange(repetitions):
 267             self.assertEqual(len(self._box), i)
 268             keys.append(self._box.add(self._template % i))
 269             self.assertEqual(len(self._box),  i + 1)
 270         for i in xrange(repetitions):
 271             self.assertEqual(len(self._box), repetitions - i)
 272             self._box.remove(keys[i])
 273             self.assertEqual(len(self._box), repetitions - i - 1)
 274
 275     def test_set_item(self):
 276         # Modify messages using __setitem__()
 277         key0 = self._box.add(self._template % 'original 0')
 278         self.assertEqual(self._box.get_string(key0),
 279                          self._template % 'original 0')
 280         key1 = self._box.add(self._template % 'original 1')
 281         self.assertEqual(self._box.get_string(key1),
 282                          self._template % 'original 1')
 283         self._box[key0] = self._template % 'changed 0'
 284         self.assertEqual(self._box.get_string(key0),
 285                          self._template % 'changed 0')
 286         self._box[key1] = self._template % 'changed 1'
 287         self.assertEqual(self._box.get_string(key1),
 288                          self._template % 'changed 1')
 289         self._box[key0] = _sample_message
 290         self._check_sample(self._box[key0])
 291         self._box[key1] = self._box[key0]
 292         self._check_sample(self._box[key1])
 293         self._box[key0] = self._template % 'original 0'
 294         self.assertEqual(self._box.get_string(key0),
 295                          self._template % 'original 0')
 296         self._check_sample(self._box[key1])
 297         self.assertRaises(KeyError,
 298                           lambda: self._box.__setitem__('foo', 'bar'))
 299         self.assertRaises(KeyError, lambda: self._box['foo'])
 300         self.assertEqual(len(self._box), 2)
 301
 302     def test_clear(self, iterations=10):
 303         # Remove all messages using clear()
 304         keys = []
 305         for i in xrange(iterations):
 306             self._box.add(self._template % i)
 307         for i, key in enumerate(keys):
 308             self.assertEqual(self._box.get_string(key), self._template % i)
 309         self._box.clear()
 310         self.assertEqual(len(self._box), 0)
 311         for i, key in enumerate(keys):
 312             self.assertRaises(KeyError, lambda: self._box.get_string(key))
 313
 314     def test_pop(self):
 315         # Get and remove a message using pop()
 316         key0 = self._box.add(self._template % 0)
 317         self.assertIn(key0, self._box)
 318         key1 = self._box.add(self._template % 1)
 319         self.assertIn(key1, self._box)
 320         self.assertEqual(self._box.pop(key0).get_payload(), '0')
 321         self.assertNotIn(key0, self._box)
 322         self.assertIn(key1, self._box)
 323         key2 = self._box.add(self._template % 2)
 324         self.assertIn(key2, self._box)
 325         self.assertEqual(self._box.pop(key2).get_payload(), '2')
 326         self.assertNotIn(key2, self._box)
 327         self.assertIn(key1, self._box)
 328         self.assertEqual(self._box.pop(key1).get_payload(), '1')
 329         self.assertNotIn(key1, self._box)
 330         self.assertEqual(len(self._box), 0)
 331
 332     def test_popitem(self, iterations=10):
 333         # Get and remove an arbitrary (key, message) using popitem()
 334         keys = []
 335         for i in xrange(10):
 336             keys.append(self._box.add(self._template % i))
 337         seen = []
 338         for i in xrange(10):
 339             key, msg = self._box.popitem()
 340             self.assertIn(key, keys)
 341             self.assertNotIn(key, seen)
 342             seen.append(key)
 343             self.assertEqual(int(msg.get_payload()), keys.index(key))
 344         self.assertEqual(len(self._box), 0)
 345         for key in keys:
 346             self.assertRaises(KeyError, lambda: self._box[key])
 347
 348     def test_update(self):
 349         # Modify multiple messages using update()
 350         key0 = self._box.add(self._template % 'original 0')
 351         key1 = self._box.add(self._template % 'original 1')
 352         key2 = self._box.add(self._template % 'original 2')
 353         self._box.update({key0: self._template % 'changed 0',
 354                           key2: _sample_message})
 355         self.assertEqual(len(self._box), 3)
 356         self.assertEqual(self._box.get_string(key0),
 357                          self._template % 'changed 0')
 358         self.assertEqual(self._box.get_string(key1),
 359                          self._template % 'original 1')
 360         self._check_sample(self._box[key2])
 361         self._box.update([(key2, self._template % 'changed 2'),
 362                     (key1, self._template % 'changed 1'),
 363                     (key0, self._template % 'original 0')])
 364         self.assertEqual(len(self._box), 3)
 365         self.assertEqual(self._box.get_string(key0),
 366                          self._template % 'original 0')
 367         self.assertEqual(self._box.get_string(key1),
 368                          self._template % 'changed 1')
 369         self.assertEqual(self._box.get_string(key2),
 370                          self._template % 'changed 2')
 371         self.assertRaises(KeyError,
 372                           lambda: self._box.update({'foo': 'bar',
 373                                           key0: self._template % "changed 0"}))
 374         self.assertEqual(len(self._box), 3)
 375         self.assertEqual(self._box.get_string(key0),
 376                          self._template % "changed 0")
 377         self.assertEqual(self._box.get_string(key1),
 378                          self._template % "changed 1")
 379         self.assertEqual(self._box.get_string(key2),
 380                          self._template % "changed 2")
 381
 382     def test_flush(self):
 383         # Write changes to disk
 384         self._test_flush_or_close(self._box.flush, True)
 385
 386     def test_lock_unlock(self):
 387         # Lock and unlock the mailbox
 388         self.assertFalse(os.path.exists(self._get_lock_path()))
 389         self._box.lock()
 390         self.assertTrue(os.path.exists(self._get_lock_path()))
 391         self._box.unlock()
 392         self.assertFalse(os.path.exists(self._get_lock_path()))
 393
 394     def test_close(self):
 395         # Close mailbox and flush changes to disk
 396         self._test_flush_or_close(self._box.close, False)
 397
 398     def _test_flush_or_close(self, method, should_call_close):
 399         contents = [self._template % i for i in xrange(3)]
 400         self._box.add(contents[0])
 401         self._box.add(contents[1])
 402         self._box.add(contents[2])
 403         method()
 404         if should_call_close:
 405             self._box.close()
 406         self._box = self._factory(self._path)
 407         keys = self._box.keys()
 408         self.assertEqual(len(keys), 3)
 409         for key in keys:
 410             self.assertIn(self._box.get_string(key), contents)
 411
 412     def test_dump_message(self):
 413         # Write message representations to disk
 414         for input in (email.message_from_string(_sample_message),
 415                       _sample_message, StringIO.StringIO(_sample_message)):
 416             output = StringIO.StringIO()
 417             self._box._dump_message(input, output)
 418             self.assertEqual(output.getvalue(),
 419                              _sample_message.replace('\n', os.linesep))
 420         output = StringIO.StringIO()
 421         self.assertRaises(TypeError,
 422                           lambda: self._box._dump_message(None, output))
 423
 424     def _get_lock_path(self):
 425         # Return the path of the dot lock file. May be overridden.
 426         return self._path + '.lock'
 427
 428
 429 class TestMailboxSuperclass(TestBase):
 430
 431     def test_notimplemented(self):
 432         # Test that all Mailbox methods raise NotImplementedException.
 433         box = mailbox.Mailbox('path')
 434         self.assertRaises(NotImplementedError, lambda: box.add(''))
 435         self.assertRaises(NotImplementedError, lambda: box.remove(''))
 436         self.assertRaises(NotImplementedError, lambda: box.__delitem__(''))
 437         self.assertRaises(NotImplementedError, lambda: box.discard(''))
 438         self.assertRaises(NotImplementedError, lambda: box.__setitem__('', ''))
 439         self.assertRaises(NotImplementedError, lambda: box.iterkeys())
 440         self.assertRaises(NotImplementedError, lambda: box.keys())
 441         self.assertRaises(NotImplementedError, lambda: box.itervalues().next())
 442         self.assertRaises(NotImplementedError, lambda: box.__iter__().next())
 443         self.assertRaises(NotImplementedError, lambda: box.values())
 444         self.assertRaises(NotImplementedError, lambda: box.iteritems().next())
 445         self.assertRaises(NotImplementedError, lambda: box.items())
 446         self.assertRaises(NotImplementedError, lambda: box.get(''))
 447         self.assertRaises(NotImplementedError, lambda: box.__getitem__(''))
 448         self.assertRaises(NotImplementedError, lambda: box.get_message(''))
 449         self.assertRaises(NotImplementedError, lambda: box.get_string(''))
 450         self.assertRaises(NotImplementedError, lambda: box.get_file(''))
 451         self.assertRaises(NotImplementedError, lambda: box.has_key(''))
 452         self.assertRaises(NotImplementedError, lambda: box.__contains__(''))
 453         self.assertRaises(NotImplementedError, lambda: box.__len__())
 454         self.assertRaises(NotImplementedError, lambda: box.clear())
 455         self.assertRaises(NotImplementedError, lambda: box.pop(''))
 456         self.assertRaises(NotImplementedError, lambda: box.popitem())
 457         self.assertRaises(NotImplementedError, lambda: box.update((('', ''),)))
 458         self.assertRaises(NotImplementedError, lambda: box.flush())
 459         self.assertRaises(NotImplementedError, lambda: box.lock())
 460         self.assertRaises(NotImplementedError, lambda: box.unlock())
 461         self.assertRaises(NotImplementedError, lambda: box.close())
 462
 463
 464 class TestMaildir(TestMailbox):
 465
 466     _factory = lambda self, path, factory=None: mailbox.Maildir(path, factory)
 467
 468     def setUp(self):
 469         TestMailbox.setUp(self)
 470         if os.name in ('nt', 'os2') or sys.platform == 'cygwin':
 471             self._box.colon = '!'
 472
 473     def test_add_MM(self):
 474         # Add a MaildirMessage instance
 475         msg = mailbox.MaildirMessage(self._template % 0)
 476         msg.set_subdir('cur')
 477         msg.set_info('foo')
 478         key = self._box.add(msg)
 479         self.assertTrue(os.path.exists(os.path.join(self._path, 'cur', '%s%sfoo' %
 480                                                  (key, self._box.colon))))
 481
 482     def test_get_MM(self):
 483         # Get a MaildirMessage instance
 484         msg = mailbox.MaildirMessage(self._template % 0)
 485         msg.set_subdir('cur')
 486         msg.set_flags('RF')
 487         key = self._box.add(msg)
 488         msg_returned = self._box.get_message(key)
 489         self.assertIsInstance(msg_returned, mailbox.MaildirMessage)
 490         self.assertEqual(msg_returned.get_subdir(), 'cur')
 491         self.assertEqual(msg_returned.get_flags(), 'FR')
 492
 493     def test_set_MM(self):
 494         # Set with a MaildirMessage instance
 495         msg0 = mailbox.MaildirMessage(self._template % 0)
 496         msg0.set_flags('TP')
 497         key = self._box.add(msg0)
 498         msg_returned = self._box.get_message(key)
 499         self.assertEqual(msg_returned.get_subdir(), 'new')
 500         self.assertEqual(msg_returned.get_flags(), 'PT')
 501         msg1 = mailbox.MaildirMessage(self._template % 1)
 502         self._box[key] = msg1
 503         msg_returned = self._box.get_message(key)
 504         self.assertEqual(msg_returned.get_subdir(), 'new')
 505         self.assertEqual(msg_returned.get_flags(), '')
 506         self.assertEqual(msg_returned.get_payload(), '1')
 507         msg2 = mailbox.MaildirMessage(self._template % 2)
 508         msg2.set_info('2,S')
 509         self._box[key] = msg2
 510         self._box[key] = self._template % 3
 511         msg_returned = self._box.get_message(key)
 512         self.assertEqual(msg_returned.get_subdir(), 'new')
 513         self.assertEqual(msg_returned.get_flags(), 'S')
 514         self.assertEqual(msg_returned.get_payload(), '3')
 515
 516     def test_consistent_factory(self):
 517         # Add a message.
 518         msg = mailbox.MaildirMessage(self._template % 0)
 519         msg.set_subdir('cur')
 520         msg.set_flags('RF')
 521         key = self._box.add(msg)
 522
 523         # Create new mailbox with
 524         class FakeMessage(mailbox.MaildirMessage):
 525             pass
 526         box = mailbox.Maildir(self._path, factory=FakeMessage)
 527         box.colon = self._box.colon
 528         msg2 = box.get_message(key)
 529         self.assertIsInstance(msg2, FakeMessage)
 530
 531     def test_initialize_new(self):
 532         # Initialize a non-existent mailbox
 533         self.tearDown()
 534         self._box = mailbox.Maildir(self._path)
 535         self._check_basics(factory=rfc822.Message)
 536         self._delete_recursively(self._path)
 537         self._box = self._factory(self._path, factory=None)
 538         self._check_basics()
 539
 540     def test_initialize_existing(self):
 541         # Initialize an existing mailbox
 542         self.tearDown()
 543         for subdir in '', 'tmp', 'new', 'cur':
 544             os.mkdir(os.path.normpath(os.path.join(self._path, subdir)))
 545         self._box = mailbox.Maildir(self._path)
 546         self._check_basics(factory=rfc822.Message)
 547         self._box = mailbox.Maildir(self._path, factory=None)
 548         self._check_basics()
 549
 550     def _check_basics(self, factory=None):
 551         # (Used by test_open_new() and test_open_existing().)
 552         self.assertEqual(self._box._path, os.path.abspath(self._path))
 553         self.assertEqual(self._box._factory, factory)
 554         for subdir in '', 'tmp', 'new', 'cur':
 555             path = os.path.join(self._path, subdir)
 556             mode = os.stat(path)[stat.ST_MODE]
 557             self.assertTrue(stat.S_ISDIR(mode), "Not a directory: '%s'" % path)
 558
 559     def test_list_folders(self):
 560         # List folders
 561         self._box.add_folder('one')
 562         self._box.add_folder('two')
 563         self._box.add_folder('three')
 564         self.assertEqual(len(self._box.list_folders()), 3)
 565         self.assertEqual(set(self._box.list_folders()),
 566                          set(('one', 'two', 'three')))
 567
 568     def test_get_folder(self):
 569         # Open folders
 570         self._box.add_folder('foo.bar')
 571         folder0 = self._box.get_folder('foo.bar')
 572         folder0.add(self._template % 'bar')
 573         self.assertTrue(os.path.isdir(os.path.join(self._path, '.foo.bar')))
 574         folder1 = self._box.get_folder('foo.bar')
 575         self.assertEqual(folder1.get_string(folder1.keys()[0]),
 576                          self._template % 'bar')
 577
 578     def test_add_and_remove_folders(self):
 579         # Delete folders
 580         self._box.add_folder('one')
 581         self._box.add_folder('two')
 582         self.assertEqual(len(self._box.list_folders()), 2)
 583         self.assertEqual(set(self._box.list_folders()), set(('one', 'two')))
 584         self._box.remove_folder('one')
 585         self.assertEqual(len(self._box.list_folders()), 1)
 586         self.assertEqual(set(self._box.list_folders()), set(('two',)))
 587         self._box.add_folder('three')
 588         self.assertEqual(len(self._box.list_folders()), 2)
 589         self.assertEqual(set(self._box.list_folders()), set(('two', 'three')))
 590         self._box.remove_folder('three')
 591         self.assertEqual(len(self._box.list_folders()), 1)
 592         self.assertEqual(set(self._box.list_folders()), set(('two',)))
 593         self._box.remove_folder('two')
 594         self.assertEqual(len(self._box.list_folders()), 0)
 595         self.assertEqual(self._box.list_folders(), [])
 596
 597     def test_clean(self):
 598         # Remove old files from 'tmp'
 599         foo_path = os.path.join(self._path, 'tmp', 'foo')
 600         bar_path = os.path.join(self._path, 'tmp', 'bar')
 601         f = open(foo_path, 'w')
 602         f.write("@")
 603         f.close()
 604         f = open(bar_path, 'w')
 605         f.write("@")
 606         f.close()
 607         self._box.clean()
 608         self.assertTrue(os.path.exists(foo_path))
 609         self.assertTrue(os.path.exists(bar_path))
 610         foo_stat = os.stat(foo_path)
 611         os.utime(foo_path, (time.time() - 129600 - 2,
 612                             foo_stat.st_mtime))
 613         self._box.clean()
 614         self.assertFalse(os.path.exists(foo_path))
 615         self.assertTrue(os.path.exists(bar_path))
 616
 617     def test_create_tmp(self, repetitions=10):
 618         # Create files in tmp directory
 619         hostname = socket.gethostname()
 620         if '/' in hostname:
 621             hostname = hostname.replace('/', r'\057')
 622         if ':' in hostname:
 623             hostname = hostname.replace(':', r'\072')
 624         pid = os.getpid()
 625         pattern = re.compile(r"(?P<time>\d+)\.M(?P<M>\d{1,6})P(?P<P>\d+)"
 626                              r"Q(?P<Q>\d+)\.(?P<host>[^:/]+)")
 627         previous_groups = None
 628         for x in xrange(repetitions):
 629             tmp_file = self._box._create_tmp()
 630             head, tail = os.path.split(tmp_file.name)
 631             self.assertEqual(head, os.path.abspath(os.path.join(self._path,
 632                                                                 "tmp")),
 633                              "File in wrong location: '%s'" % head)
 634             match = pattern.match(tail)
 635             self.assertTrue(match is not None, "Invalid file name: '%s'" % tail)
 636             groups = match.groups()
 637             if previous_groups is not None:
 638                 self.assertTrue(int(groups[0] >= previous_groups[0]),
 639                              "Non-monotonic seconds: '%s' before '%s'" %
 640                              (previous_groups[0], groups[0]))
 641                 self.assertTrue(int(groups[1] >= previous_groups[1]) or
 642                              groups[0] != groups[1],
 643                              "Non-monotonic milliseconds: '%s' before '%s'" %
 644                              (previous_groups[1], groups[1]))
 645                 self.assertTrue(int(groups[2]) == pid,
 646                              "Process ID mismatch: '%s' should be '%s'" %
 647                              (groups[2], pid))
 648                 self.assertTrue(int(groups[3]) == int(previous_groups[3]) + 1,
 649                              "Non-sequential counter: '%s' before '%s'" %
 650                              (previous_groups[3], groups[3]))
 651                 self.assertTrue(groups[4] == hostname,
 652                              "Host name mismatch: '%s' should be '%s'" %
 653                              (groups[4], hostname))
 654             previous_groups = groups
 655             tmp_file.write(_sample_message)
 656             tmp_file.seek(0)
 657             self.assertTrue(tmp_file.read() == _sample_message)
 658             tmp_file.close()
 659         file_count = len(os.listdir(os.path.join(self._path, "tmp")))
 660         self.assertTrue(file_count == repetitions,
 661                      "Wrong file count: '%s' should be '%s'" %
 662                      (file_count, repetitions))
 663
 664     def test_refresh(self):
 665         # Update the table of contents
 666         self.assertEqual(self._box._toc, {})
 667         key0 = self._box.add(self._template % 0)
 668         key1 = self._box.add(self._template % 1)
 669         self.assertEqual(self._box._toc, {})
 670         self._box._refresh()
 671         self.assertEqual(self._box._toc, {key0: os.path.join('new', key0),
 672                                           key1: os.path.join('new', key1)})
 673         key2 = self._box.add(self._template % 2)
 674         self.assertEqual(self._box._toc, {key0: os.path.join('new', key0),
 675                                           key1: os.path.join('new', key1)})
 676         self._box._refresh()
 677         self.assertEqual(self._box._toc, {key0: os.path.join('new', key0),
 678                                           key1: os.path.join('new', key1),
 679                                           key2: os.path.join('new', key2)})
 680
 681     def test_lookup(self):
 682         # Look up message subpaths in the TOC
 683         self.assertRaises(KeyError, lambda: self._box._lookup('foo'))
 684         key0 = self._box.add(self._template % 0)
 685         self.assertEqual(self._box._lookup(key0), os.path.join('new', key0))
 686         os.remove(os.path.join(self._path, 'new', key0))
 687         self.assertEqual(self._box._toc, {key0: os.path.join('new', key0)})
 688         # Be sure that the TOC is read back from disk (see issue #6896
 689         # about bad mtime behaviour on some systems).
 690         self._box.flush()
 691         self.assertRaises(KeyError, lambda: self._box._lookup(key0))
 692         self.assertEqual(self._box._toc, {})
 693
 694     def test_lock_unlock(self):
 695         # Lock and unlock the mailbox. For Maildir, this does nothing.
 696         self._box.lock()
 697         self._box.unlock()
 698
 699     def test_folder (self):
 700         # Test for bug #1569790: verify that folders returned by .get_folder()
 701         # use the same factory function.
 702         def dummy_factory (s):
 703             return None
 704         box = self._factory(self._path, factory=dummy_factory)
 705         folder = box.add_folder('folder1')
 706         self.assertIs(folder._factory, dummy_factory)
 707
 708         folder1_alias = box.get_folder('folder1')
 709         self.assertIs(folder1_alias._factory, dummy_factory)
 710
 711     def test_directory_in_folder (self):
 712         # Test that mailboxes still work if there's a stray extra directory
 713         # in a folder.
 714         for i in range(10):
 715             self._box.add(mailbox.Message(_sample_message))
 716
 717         # Create a stray directory
 718         os.mkdir(os.path.join(self._path, 'cur', 'stray-dir'))
 719
 720         # Check that looping still works with the directory present.
 721         for msg in self._box:
 722             pass
 723
 724     def test_file_permissions(self):
 725         # Verify that message files are created without execute permissions
 726         if not hasattr(os, "stat") or not hasattr(os, "umask"):
 727             return
 728         msg = mailbox.MaildirMessage(self._template % 0)
 729         orig_umask = os.umask(0)
 730         try:
 731             key = self._box.add(msg)
 732         finally:
 733             os.umask(orig_umask)
 734         path = os.path.join(self._path, self._box._lookup(key))
 735         mode = os.stat(path).st_mode
 736         self.assertEqual(mode & 0111, 0)
 737
 738     def test_folder_file_perms(self):
 739         # From bug #3228, we want to verify that the file created inside a Maildir
 740         # subfolder isn't marked as executable.
 741         if not hasattr(os, "stat") or not hasattr(os, "umask"):
 742             return
 743
 744         orig_umask = os.umask(0)
 745         try:
 746             subfolder = self._box.add_folder('subfolder')
 747         finally:
 748             os.umask(orig_umask)
 749
 750         path = os.path.join(subfolder._path, 'maildirfolder')
 751         st = os.stat(path)
 752         perms = st.st_mode
 753         self.assertFalse((perms & 0111)) # Execute bits should all be off.
 754
 755     def test_reread(self):
 756         # Wait for 2 seconds
 757         time.sleep(2)
 758
 759         # Initially, the mailbox has not been read and the time is null.
 760         assert getattr(self._box, '_last_read', None) is None
 761
 762         # Refresh mailbox; the times should now be set to something.
 763         self._box._refresh()
 764         assert getattr(self._box, '_last_read', None) is not None
 765
 766         # Try calling _refresh() again; the modification times shouldn't have
 767         # changed, so the mailbox should not be re-reading.  Re-reading causes
 768         # the ._toc attribute to be assigned a new dictionary object, so
 769         # we'll check that the ._toc attribute isn't a different object.
 770         orig_toc = self._box._toc
 771         def refreshed():
 772             return self._box._toc is not orig_toc
 773
 774         time.sleep(1)         # Wait 1sec to ensure time.time()'s value changes
 775         self._box._refresh()
 776         assert not refreshed()
 777
 778         # Now, write something into cur and remove it.  This changes
 779         # the mtime and should cause a re-read.
 780         filename = os.path.join(self._path, 'cur', 'stray-file')
 781         f = open(filename, 'w')
 782         f.close()
 783         os.unlink(filename)
 784         self._box._refresh()
 785         assert refreshed()
 786
 787 class _TestMboxMMDF(TestMailbox):
 788
 789     def tearDown(self):
 790         self._box.close()
 791         self._delete_recursively(self._path)
 792         for lock_remnant in glob.glob(self._path + '.*'):
 793             test_support.unlink(lock_remnant)
 794
 795     def test_add_from_string(self):
 796         # Add a string starting with 'From ' to the mailbox
 797         key = self._box.add('From foo@bar blah\nFrom: foo\n\n0')
 798         self.assertEqual(self._box[key].get_from(), 'foo@bar blah')
 799         self.assertEqual(self._box[key].get_payload(), '0')
 800
 801     def test_add_mbox_or_mmdf_message(self):
 802         # Add an mboxMessage or MMDFMessage
 803         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
 804             msg = class_('From foo@bar blah\nFrom: foo\n\n0')
 805             key = self._box.add(msg)
 806
 807     def test_open_close_open(self):
 808         # Open and inspect previously-created mailbox
 809         values = [self._template % i for i in xrange(3)]
 810         for value in values:
 811             self._box.add(value)
 812         self._box.close()
 813         mtime = os.path.getmtime(self._path)
 814         self._box = self._factory(self._path)
 815         self.assertEqual(len(self._box), 3)
 816         for key in self._box.iterkeys():
 817             self.assertIn(self._box.get_string(key), values)
 818         self._box.close()
 819         self.assertEqual(mtime, os.path.getmtime(self._path))
 820
 821     def test_add_and_close(self):
 822         # Verifying that closing a mailbox doesn't change added items
 823         self._box.add(_sample_message)
 824         for i in xrange(3):
 825             self._box.add(self._template % i)
 826         self._box.add(_sample_message)
 827         self._box._file.flush()
 828         self._box._file.seek(0)
 829         contents = self._box._file.read()
 830         self._box.close()
 831         with open(self._path, 'rb') as f:
 832             self.assertEqual(contents, f.read())
 833         self._box = self._factory(self._path)
 834
 835     def test_lock_conflict(self):
 836         # Fork off a subprocess that will lock the file for 2 seconds,
 837         # unlock it, and then exit.
 838         if not hasattr(os, 'fork'):
 839             return
 840         pid = os.fork()
 841         if pid == 0:
 842             # In the child, lock the mailbox.
 843             self._box.lock()
 844             time.sleep(2)
 845             self._box.unlock()
 846             os._exit(0)
 847
 848         # In the parent, sleep a bit to give the child time to acquire
 849         # the lock.
 850         time.sleep(0.5)
 851         try:
 852             self.assertRaises(mailbox.ExternalClashError,
 853                               self._box.lock)
 854         finally:
 855             # Wait for child to exit.  Locking should now succeed.
 856             exited_pid, status = os.waitpid(pid, 0)
 857
 858         self._box.lock()
 859         self._box.unlock()
 860
 861     def test_relock(self):
 862         # Test case for bug #1575506: the mailbox class was locking the
 863         # wrong file object in its flush() method.
 864         msg = "Subject: sub\n\nbody\n"
 865         key1 = self._box.add(msg)
 866         self._box.flush()
 867         self._box.close()
 868
 869         self._box = self._factory(self._path)
 870         self._box.lock()
 871         key2 = self._box.add(msg)
 872         self._box.flush()
 873         self.assertTrue(self._box._locked)
 874         self._box.close()
 875
 876
 877 class TestMbox(_TestMboxMMDF):
 878
 879     _factory = lambda self, path, factory=None: mailbox.mbox(path, factory)
 880
 881     def test_file_perms(self):
 882         # From bug #3228, we want to verify that the mailbox file isn't executable,
 883         # even if the umask is set to something that would leave executable bits set.
 884         # We only run this test on platforms that support umask.
 885         if hasattr(os, 'umask') and hasattr(os, 'stat'):
 886             try:
 887                 old_umask = os.umask(0077)
 888                 self._box.close()
 889                 os.unlink(self._path)
 890                 self._box = mailbox.mbox(self._path, create=True)
 891                 self._box.add('')
 892                 self._box.close()
 893             finally:
 894                 os.umask(old_umask)
 895
 896             st = os.stat(self._path)
 897             perms = st.st_mode
 898             self.assertFalse((perms & 0111)) # Execute bits should all be off.
 899
 900 class TestMMDF(_TestMboxMMDF):
 901
 902     _factory = lambda self, path, factory=None: mailbox.MMDF(path, factory)
 903
 904
 905 class TestMH(TestMailbox):
 906
 907     _factory = lambda self, path, factory=None: mailbox.MH(path, factory)
 908
 909     def test_list_folders(self):
 910         # List folders
 911         self._box.add_folder('one')
 912         self._box.add_folder('two')
 913         self._box.add_folder('three')
 914         self.assertEqual(len(self._box.list_folders()), 3)
 915         self.assertEqual(set(self._box.list_folders()),
 916                          set(('one', 'two', 'three')))
 917
 918     def test_get_folder(self):
 919         # Open folders
 920         def dummy_factory (s):
 921             return None
 922         self._box = self._factory(self._path, dummy_factory)
 923
 924         new_folder = self._box.add_folder('foo.bar')
 925         folder0 = self._box.get_folder('foo.bar')
 926         folder0.add(self._template % 'bar')
 927         self.assertTrue(os.path.isdir(os.path.join(self._path, 'foo.bar')))
 928         folder1 = self._box.get_folder('foo.bar')
 929         self.assertEqual(folder1.get_string(folder1.keys()[0]),
 930                          self._template % 'bar')
 931
 932         # Test for bug #1569790: verify that folders returned by .get_folder()
 933         # use the same factory function.
 934         self.assertIs(new_folder._factory, self._box._factory)
 935         self.assertIs(folder0._factory, self._box._factory)
 936
 937     def test_add_and_remove_folders(self):
 938         # Delete folders
 939         self._box.add_folder('one')
 940         self._box.add_folder('two')
 941         self.assertEqual(len(self._box.list_folders()), 2)
 942         self.assertEqual(set(self._box.list_folders()), set(('one', 'two')))
 943         self._box.remove_folder('one')
 944         self.assertEqual(len(self._box.list_folders()), 1)
 945         self.assertEqual(set(self._box.list_folders()), set(('two', )))
 946         self._box.add_folder('three')
 947         self.assertEqual(len(self._box.list_folders()), 2)
 948         self.assertEqual(set(self._box.list_folders()), set(('two', 'three')))
 949         self._box.remove_folder('three')
 950         self.assertEqual(len(self._box.list_folders()), 1)
 951         self.assertEqual(set(self._box.list_folders()), set(('two', )))
 952         self._box.remove_folder('two')
 953         self.assertEqual(len(self._box.list_folders()), 0)
 954         self.assertEqual(self._box.list_folders(), [])
 955
 956     def test_sequences(self):
 957         # Get and set sequences
 958         self.assertEqual(self._box.get_sequences(), {})
 959         msg0 = mailbox.MHMessage(self._template % 0)
 960         msg0.add_sequence('foo')
 961         key0 = self._box.add(msg0)
 962         self.assertEqual(self._box.get_sequences(), {'foo':[key0]})
 963         msg1 = mailbox.MHMessage(self._template % 1)
 964         msg1.set_sequences(['bar', 'replied', 'foo'])
 965         key1 = self._box.add(msg1)
 966         self.assertEqual(self._box.get_sequences(),
 967                          {'foo':[key0, key1], 'bar':[key1], 'replied':[key1]})
 968         msg0.set_sequences(['flagged'])
 969         self._box[key0] = msg0
 970         self.assertEqual(self._box.get_sequences(),
 971                          {'foo':[key1], 'bar':[key1], 'replied':[key1],
 972                           'flagged':[key0]})
 973         self._box.remove(key1)
 974         self.assertEqual(self._box.get_sequences(), {'flagged':[key0]})
 975
 976     def test_issue2625(self):
 977         msg0 = mailbox.MHMessage(self._template % 0)
 978         msg0.add_sequence('foo')
 979         key0 = self._box.add(msg0)
 980         refmsg0 = self._box.get_message(key0)
 981
 982     def test_issue7627(self):
 983         msg0 = mailbox.MHMessage(self._template % 0)
 984         key0 = self._box.add(msg0)
 985         self._box.lock()
 986         self._box.remove(key0)
 987         self._box.unlock()
 988
 989     def test_pack(self):
 990         # Pack the contents of the mailbox
 991         msg0 = mailbox.MHMessage(self._template % 0)
 992         msg1 = mailbox.MHMessage(self._template % 1)
 993         msg2 = mailbox.MHMessage(self._template % 2)
 994         msg3 = mailbox.MHMessage(self._template % 3)
 995         msg0.set_sequences(['foo', 'unseen'])
 996         msg1.set_sequences(['foo'])
 997         msg2.set_sequences(['foo', 'flagged'])
 998         msg3.set_sequences(['foo', 'bar', 'replied'])
 999         key0 = self._box.add(msg0)
1000         key1 = self._box.add(msg1)
1001         key2 = self._box.add(msg2)
1002         key3 = self._box.add(msg3)
1003         self.assertEqual(self._box.get_sequences(),
1004                          {'foo':[key0,key1,key2,key3], 'unseen':[key0],
1005                           'flagged':[key2], 'bar':[key3], 'replied':[key3]})
1006         self._box.remove(key2)
1007         self.assertEqual(self._box.get_sequences(),
1008                          {'foo':[key0,key1,key3], 'unseen':[key0], 'bar':[key3],
1009                           'replied':[key3]})
1010         self._box.pack()
1011         self.assertEqual(self._box.keys(), [1, 2, 3])
1012         key0 = key0
1013         key1 = key0 + 1
1014         key2 = key1 + 1
1015         self.assertEqual(self._box.get_sequences(),
1016                      {'foo':[1, 2, 3], 'unseen':[1], 'bar':[3], 'replied':[3]})
1017
1018         # Test case for packing while holding the mailbox locked.
1019         key0 = self._box.add(msg1)
1020         key1 = self._box.add(msg1)
1021         key2 = self._box.add(msg1)
1022         key3 = self._box.add(msg1)
1023
1024         self._box.remove(key0)
1025         self._box.remove(key2)
1026         self._box.lock()
1027         self._box.pack()
1028         self._box.unlock()
1029         self.assertEqual(self._box.get_sequences(),
1030                          {'foo':[1, 2, 3, 4, 5],
1031                           'unseen':[1], 'bar':[3], 'replied':[3]})
1032
1033     def _get_lock_path(self):
1034         return os.path.join(self._path, '.mh_sequences.lock')
1035
1036
1037 class TestBabyl(TestMailbox):
1038
1039     _factory = lambda self, path, factory=None: mailbox.Babyl(path, factory)
1040
1041     def tearDown(self):
1042         self._box.close()
1043         self._delete_recursively(self._path)
1044         for lock_remnant in glob.glob(self._path + '.*'):
1045             test_support.unlink(lock_remnant)
1046
1047     def test_labels(self):
1048         # Get labels from the mailbox
1049         self.assertEqual(self._box.get_labels(), [])
1050         msg0 = mailbox.BabylMessage(self._template % 0)
1051         msg0.add_label('foo')
1052         key0 = self._box.add(msg0)
1053         self.assertEqual(self._box.get_labels(), ['foo'])
1054         msg1 = mailbox.BabylMessage(self._template % 1)
1055         msg1.set_labels(['bar', 'answered', 'foo'])
1056         key1 = self._box.add(msg1)
1057         self.assertEqual(set(self._box.get_labels()), set(['foo', 'bar']))
1058         msg0.set_labels(['blah', 'filed'])
1059         self._box[key0] = msg0
1060         self.assertEqual(set(self._box.get_labels()),
1061                          set(['foo', 'bar', 'blah']))
1062         self._box.remove(key1)
1063         self.assertEqual(set(self._box.get_labels()), set(['blah']))
1064
1065
1066 class TestMessage(TestBase):
1067
1068     _factory = mailbox.Message      # Overridden by subclasses to reuse tests
1069
1070     def setUp(self):
1071         self._path = test_support.TESTFN
1072
1073     def tearDown(self):
1074         self._delete_recursively(self._path)
1075
1076     def test_initialize_with_eMM(self):
1077         # Initialize based on email.message.Message instance
1078         eMM = email.message_from_string(_sample_message)
1079         msg = self._factory(eMM)
1080         self._post_initialize_hook(msg)
1081         self._check_sample(msg)
1082
1083     def test_initialize_with_string(self):
1084         # Initialize based on string
1085         msg = self._factory(_sample_message)
1086         self._post_initialize_hook(msg)
1087         self._check_sample(msg)
1088
1089     def test_initialize_with_file(self):
1090         # Initialize based on contents of file
1091         f = open(self._path, 'w+')
1092         f.write(_sample_message)
1093         f.seek(0)
1094         msg = self._factory(f)
1095         self._post_initialize_hook(msg)
1096         self._check_sample(msg)
1097         f.close()
1098
1099     def test_initialize_with_nothing(self):
1100         # Initialize without arguments
1101         msg = self._factory()
1102         self._post_initialize_hook(msg)
1103         self.assertIsInstance(msg, email.message.Message)
1104         self.assertIsInstance(msg, mailbox.Message)
1105         self.assertIsInstance(msg, self._factory)
1106         self.assertEqual(msg.keys(), [])
1107         self.assertFalse(msg.is_multipart())
1108         self.assertEqual(msg.get_payload(), None)
1109
1110     def test_initialize_incorrectly(self):
1111         # Initialize with invalid argument
1112         self.assertRaises(TypeError, lambda: self._factory(object()))
1113
1114     def test_become_message(self):
1115         # Take on the state of another message
1116         eMM = email.message_from_string(_sample_message)
1117         msg = self._factory()
1118         msg._become_message(eMM)
1119         self._check_sample(msg)
1120
1121     def test_explain_to(self):
1122         # Copy self's format-specific data to other message formats.
1123         # This test is superficial; better ones are in TestMessageConversion.
1124         msg = self._factory()
1125         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1126                        mailbox.mboxMessage, mailbox.MHMessage,
1127                        mailbox.BabylMessage, mailbox.MMDFMessage):
1128             other_msg = class_()
1129             msg._explain_to(other_msg)
1130         other_msg = email.message.Message()
1131         self.assertRaises(TypeError, lambda: msg._explain_to(other_msg))
1132
1133     def _post_initialize_hook(self, msg):
1134         # Overridden by subclasses to check extra things after initialization
1135         pass
1136
1137
1138 class TestMaildirMessage(TestMessage):
1139
1140     _factory = mailbox.MaildirMessage
1141
1142     def _post_initialize_hook(self, msg):
1143         self.assertEqual(msg._subdir, 'new')
1144         self.assertEqual(msg._info,'')
1145
1146     def test_subdir(self):
1147         # Use get_subdir() and set_subdir()
1148         msg = mailbox.MaildirMessage(_sample_message)
1149         self.assertEqual(msg.get_subdir(), 'new')
1150         msg.set_subdir('cur')
1151         self.assertEqual(msg.get_subdir(), 'cur')
1152         msg.set_subdir('new')
1153         self.assertEqual(msg.get_subdir(), 'new')
1154         self.assertRaises(ValueError, lambda: msg.set_subdir('tmp'))
1155         self.assertEqual(msg.get_subdir(), 'new')
1156         msg.set_subdir('new')
1157         self.assertEqual(msg.get_subdir(), 'new')
1158         self._check_sample(msg)
1159
1160     def test_flags(self):
1161         # Use get_flags(), set_flags(), add_flag(), remove_flag()
1162         msg = mailbox.MaildirMessage(_sample_message)
1163         self.assertEqual(msg.get_flags(), '')
1164         self.assertEqual(msg.get_subdir(), 'new')
1165         msg.set_flags('F')
1166         self.assertEqual(msg.get_subdir(), 'new')
1167         self.assertEqual(msg.get_flags(), 'F')
1168         msg.set_flags('SDTP')
1169         self.assertEqual(msg.get_flags(), 'DPST')
1170         msg.add_flag('FT')
1171         self.assertEqual(msg.get_flags(), 'DFPST')
1172         msg.remove_flag('TDRP')
1173         self.assertEqual(msg.get_flags(), 'FS')
1174         self.assertEqual(msg.get_subdir(), 'new')
1175         self._check_sample(msg)
1176
1177     def test_date(self):
1178         # Use get_date() and set_date()
1179         msg = mailbox.MaildirMessage(_sample_message)
1180         diff = msg.get_date() - time.time()
1181         self.assertTrue(abs(diff) < 60, diff)
1182         msg.set_date(0.0)
1183         self.assertEqual(msg.get_date(), 0.0)
1184
1185     def test_info(self):
1186         # Use get_info() and set_info()
1187         msg = mailbox.MaildirMessage(_sample_message)
1188         self.assertEqual(msg.get_info(), '')
1189         msg.set_info('1,foo=bar')
1190         self.assertEqual(msg.get_info(), '1,foo=bar')
1191         self.assertRaises(TypeError, lambda: msg.set_info(None))
1192         self._check_sample(msg)
1193
1194     def test_info_and_flags(self):
1195         # Test interaction of info and flag methods
1196         msg = mailbox.MaildirMessage(_sample_message)
1197         self.assertEqual(msg.get_info(), '')
1198         msg.set_flags('SF')
1199         self.assertEqual(msg.get_flags(), 'FS')
1200         self.assertEqual(msg.get_info(), '2,FS')
1201         msg.set_info('1,')
1202         self.assertEqual(msg.get_flags(), '')
1203         self.assertEqual(msg.get_info(), '1,')
1204         msg.remove_flag('RPT')
1205         self.assertEqual(msg.get_flags(), '')
1206         self.assertEqual(msg.get_info(), '1,')
1207         msg.add_flag('D')
1208         self.assertEqual(msg.get_flags(), 'D')
1209         self.assertEqual(msg.get_info(), '2,D')
1210         self._check_sample(msg)
1211
1212
1213 class _TestMboxMMDFMessage(TestMessage):
1214
1215     _factory = mailbox._mboxMMDFMessage
1216
1217     def _post_initialize_hook(self, msg):
1218         self._check_from(msg)
1219
1220     def test_initialize_with_unixfrom(self):
1221         # Initialize with a message that already has a _unixfrom attribute
1222         msg = mailbox.Message(_sample_message)
1223         msg.set_unixfrom('From foo@bar blah')
1224         msg = mailbox.mboxMessage(msg)
1225         self.assertEqual(msg.get_from(), 'foo@bar blah')
1226
1227     def test_from(self):
1228         # Get and set "From " line
1229         msg = mailbox.mboxMessage(_sample_message)
1230         self._check_from(msg)
1231         msg.set_from('foo bar')
1232         self.assertEqual(msg.get_from(), 'foo bar')
1233         msg.set_from('foo@bar', True)
1234         self._check_from(msg, 'foo@bar')
1235         msg.set_from('blah@temp', time.localtime())
1236         self._check_from(msg, 'blah@temp')
1237
1238     def test_flags(self):
1239         # Use get_flags(), set_flags(), add_flag(), remove_flag()
1240         msg = mailbox.mboxMessage(_sample_message)
1241         self.assertEqual(msg.get_flags(), '')
1242         msg.set_flags('F')
1243         self.assertEqual(msg.get_flags(), 'F')
1244         msg.set_flags('XODR')
1245         self.assertEqual(msg.get_flags(), 'RODX')
1246         msg.add_flag('FA')
1247         self.assertEqual(msg.get_flags(), 'RODFAX')
1248         msg.remove_flag('FDXA')
1249         self.assertEqual(msg.get_flags(), 'RO')
1250         self._check_sample(msg)
1251
1252     def _check_from(self, msg, sender=None):
1253         # Check contents of "From " line
1254         if sender is None:
1255             sender = "MAILER-DAEMON"
1256         self.assertTrue(re.match(sender + r" \w{3} \w{3} [\d ]\d [\d ]\d:\d{2}:"
1257                                  r"\d{2} \d{4}", msg.get_from()))
1258
1259
1260 class TestMboxMessage(_TestMboxMMDFMessage):
1261
1262     _factory = mailbox.mboxMessage
1263
1264
1265 class TestMHMessage(TestMessage):
1266
1267     _factory = mailbox.MHMessage
1268
1269     def _post_initialize_hook(self, msg):
1270         self.assertEqual(msg._sequences, [])
1271
1272     def test_sequences(self):
1273         # Get, set, join, and leave sequences
1274         msg = mailbox.MHMessage(_sample_message)
1275         self.assertEqual(msg.get_sequences(), [])
1276         msg.set_sequences(['foobar'])
1277         self.assertEqual(msg.get_sequences(), ['foobar'])
1278         msg.set_sequences([])
1279         self.assertEqual(msg.get_sequences(), [])
1280         msg.add_sequence('unseen')
1281         self.assertEqual(msg.get_sequences(), ['unseen'])
1282         msg.add_sequence('flagged')
1283         self.assertEqual(msg.get_sequences(), ['unseen', 'flagged'])
1284         msg.add_sequence('flagged')
1285         self.assertEqual(msg.get_sequences(), ['unseen', 'flagged'])
1286         msg.remove_sequence('unseen')
1287         self.assertEqual(msg.get_sequences(), ['flagged'])
1288         msg.add_sequence('foobar')
1289         self.assertEqual(msg.get_sequences(), ['flagged', 'foobar'])
1290         msg.remove_sequence('replied')
1291         self.assertEqual(msg.get_sequences(), ['flagged', 'foobar'])
1292         msg.set_sequences(['foobar', 'replied'])
1293         self.assertEqual(msg.get_sequences(), ['foobar', 'replied'])
1294
1295
1296 class TestBabylMessage(TestMessage):
1297
1298     _factory = mailbox.BabylMessage
1299
1300     def _post_initialize_hook(self, msg):
1301         self.assertEqual(msg._labels, [])
1302
1303     def test_labels(self):
1304         # Get, set, join, and leave labels
1305         msg = mailbox.BabylMessage(_sample_message)
1306         self.assertEqual(msg.get_labels(), [])
1307         msg.set_labels(['foobar'])
1308         self.assertEqual(msg.get_labels(), ['foobar'])
1309         msg.set_labels([])
1310         self.assertEqual(msg.get_labels(), [])
1311         msg.add_label('filed')
1312         self.assertEqual(msg.get_labels(), ['filed'])
1313         msg.add_label('resent')
1314         self.assertEqual(msg.get_labels(), ['filed', 'resent'])
1315         msg.add_label('resent')
1316         self.assertEqual(msg.get_labels(), ['filed', 'resent'])
1317         msg.remove_label('filed')
1318         self.assertEqual(msg.get_labels(), ['resent'])
1319         msg.add_label('foobar')
1320         self.assertEqual(msg.get_labels(), ['resent', 'foobar'])
1321         msg.remove_label('unseen')
1322         self.assertEqual(msg.get_labels(), ['resent', 'foobar'])
1323         msg.set_labels(['foobar', 'answered'])
1324         self.assertEqual(msg.get_labels(), ['foobar', 'answered'])
1325
1326     def test_visible(self):
1327         # Get, set, and update visible headers
1328         msg = mailbox.BabylMessage(_sample_message)
1329         visible = msg.get_visible()
1330         self.assertEqual(visible.keys(), [])
1331         self.assertIs(visible.get_payload(), None)
1332         visible['User-Agent'] = 'FooBar 1.0'
1333         visible['X-Whatever'] = 'Blah'
1334         self.assertEqual(msg.get_visible().keys(), [])
1335         msg.set_visible(visible)
1336         visible = msg.get_visible()
1337         self.assertEqual(visible.keys(), ['User-Agent', 'X-Whatever'])
1338         self.assertEqual(visible['User-Agent'], 'FooBar 1.0')
1339         self.assertEqual(visible['X-Whatever'], 'Blah')
1340         self.assertIs(visible.get_payload(), None)
1341         msg.update_visible()
1342         self.assertEqual(visible.keys(), ['User-Agent', 'X-Whatever'])
1343         self.assertIs(visible.get_payload(), None)
1344         visible = msg.get_visible()
1345         self.assertEqual(visible.keys(), ['User-Agent', 'Date', 'From', 'To',
1346                                           'Subject'])
1347         for header in ('User-Agent', 'Date', 'From', 'To', 'Subject'):
1348             self.assertEqual(visible[header], msg[header])
1349
1350
1351 class TestMMDFMessage(_TestMboxMMDFMessage):
1352
1353     _factory = mailbox.MMDFMessage
1354
1355
1356 class TestMessageConversion(TestBase):
1357
1358     def test_plain_to_x(self):
1359         # Convert Message to all formats
1360         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1361                        mailbox.mboxMessage, mailbox.MHMessage,
1362                        mailbox.BabylMessage, mailbox.MMDFMessage):
1363             msg_plain = mailbox.Message(_sample_message)
1364             msg = class_(msg_plain)
1365             self._check_sample(msg)
1366
1367     def test_x_to_plain(self):
1368         # Convert all formats to Message
1369         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1370                        mailbox.mboxMessage, mailbox.MHMessage,
1371                        mailbox.BabylMessage, mailbox.MMDFMessage):
1372             msg = class_(_sample_message)
1373             msg_plain = mailbox.Message(msg)
1374             self._check_sample(msg_plain)
1375
1376     def test_x_to_invalid(self):
1377         # Convert all formats to an invalid format
1378         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1379                        mailbox.mboxMessage, mailbox.MHMessage,
1380                        mailbox.BabylMessage, mailbox.MMDFMessage):
1381             self.assertRaises(TypeError, lambda: class_(False))
1382
1383     def test_maildir_to_maildir(self):
1384         # Convert MaildirMessage to MaildirMessage
1385         msg_maildir = mailbox.MaildirMessage(_sample_message)
1386         msg_maildir.set_flags('DFPRST')
1387         msg_maildir.set_subdir('cur')
1388         date = msg_maildir.get_date()
1389         msg = mailbox.MaildirMessage(msg_maildir)
1390         self._check_sample(msg)
1391         self.assertEqual(msg.get_flags(), 'DFPRST')
1392         self.assertEqual(msg.get_subdir(), 'cur')
1393         self.assertEqual(msg.get_date(), date)
1394
1395     def test_maildir_to_mboxmmdf(self):
1396         # Convert MaildirMessage to mboxmessage and MMDFMessage
1397         pairs = (('D', ''), ('F', 'F'), ('P', ''), ('R', 'A'), ('S', 'R'),
1398                  ('T', 'D'), ('DFPRST', 'RDFA'))
1399         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1400             msg_maildir = mailbox.MaildirMessage(_sample_message)
1401             msg_maildir.set_date(0.0)
1402             for setting, result in pairs:
1403                 msg_maildir.set_flags(setting)
1404                 msg = class_(msg_maildir)
1405                 self.assertEqual(msg.get_flags(), result)
1406                 self.assertEqual(msg.get_from(), 'MAILER-DAEMON %s' %
1407                                  time.asctime(time.gmtime(0.0)))
1408             msg_maildir.set_subdir('cur')
1409             self.assertEqual(class_(msg_maildir).get_flags(), 'RODFA')
1410
1411     def test_maildir_to_mh(self):
1412         # Convert MaildirMessage to MHMessage
1413         msg_maildir = mailbox.MaildirMessage(_sample_message)
1414         pairs = (('D', ['unseen']), ('F', ['unseen', 'flagged']),
1415                  ('P', ['unseen']), ('R', ['unseen', 'replied']), ('S', []),
1416                  ('T', ['unseen']), ('DFPRST', ['replied', 'flagged']))
1417         for setting, result in pairs:
1418             msg_maildir.set_flags(setting)
1419             self.assertEqual(mailbox.MHMessage(msg_maildir).get_sequences(),
1420                              result)
1421
1422     def test_maildir_to_babyl(self):
1423         # Convert MaildirMessage to Babyl
1424         msg_maildir = mailbox.MaildirMessage(_sample_message)
1425         pairs = (('D', ['unseen']), ('F', ['unseen']),
1426                  ('P', ['unseen', 'forwarded']), ('R', ['unseen', 'answered']),
1427                  ('S', []), ('T', ['unseen', 'deleted']),
1428                  ('DFPRST', ['deleted', 'answered', 'forwarded']))
1429         for setting, result in pairs:
1430             msg_maildir.set_flags(setting)
1431             self.assertEqual(mailbox.BabylMessage(msg_maildir).get_labels(),
1432                              result)
1433
1434     def test_mboxmmdf_to_maildir(self):
1435         # Convert mboxMessage and MMDFMessage to MaildirMessage
1436         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1437             msg_mboxMMDF = class_(_sample_message)
1438             msg_mboxMMDF.set_from('foo@bar', time.gmtime(0.0))
1439             pairs = (('R', 'S'), ('O', ''), ('D', 'T'), ('F', 'F'), ('A', 'R'),
1440                      ('RODFA', 'FRST'))
1441             for setting, result in pairs:
1442                 msg_mboxMMDF.set_flags(setting)
1443                 msg = mailbox.MaildirMessage(msg_mboxMMDF)
1444                 self.assertEqual(msg.get_flags(), result)
1445                 self.assertEqual(msg.get_date(), 0.0)
1446             msg_mboxMMDF.set_flags('O')
1447             self.assertEqual(mailbox.MaildirMessage(msg_mboxMMDF).get_subdir(),
1448                              'cur')
1449
1450     def test_mboxmmdf_to_mboxmmdf(self):
1451         # Convert mboxMessage and MMDFMessage to mboxMessage and MMDFMessage
1452         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1453             msg_mboxMMDF = class_(_sample_message)
1454             msg_mboxMMDF.set_flags('RODFA')
1455             msg_mboxMMDF.set_from('foo@bar')
1456             for class2_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1457                 msg2 = class2_(msg_mboxMMDF)
1458                 self.assertEqual(msg2.get_flags(), 'RODFA')
1459                 self.assertEqual(msg2.get_from(), 'foo@bar')
1460
1461     def test_mboxmmdf_to_mh(self):
1462         # Convert mboxMessage and MMDFMessage to MHMessage
1463         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1464             msg_mboxMMDF = class_(_sample_message)
1465             pairs = (('R', []), ('O', ['unseen']), ('D', ['unseen']),
1466                      ('F', ['unseen', 'flagged']),
1467                      ('A', ['unseen', 'replied']),
1468                      ('RODFA', ['replied', 'flagged']))
1469             for setting, result in pairs:
1470                 msg_mboxMMDF.set_flags(setting)
1471                 self.assertEqual(mailbox.MHMessage(msg_mboxMMDF).get_sequences(),
1472                                  result)
1473
1474     def test_mboxmmdf_to_babyl(self):
1475         # Convert mboxMessage and MMDFMessage to BabylMessage
1476         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1477             msg = class_(_sample_message)
1478             pairs = (('R', []), ('O', ['unseen']),
1479                      ('D', ['unseen', 'deleted']), ('F', ['unseen']),
1480                      ('A', ['unseen', 'answered']),
1481                      ('RODFA', ['deleted', 'answered']))
1482             for setting, result in pairs:
1483                 msg.set_flags(setting)
1484                 self.assertEqual(mailbox.BabylMessage(msg).get_labels(), result)
1485
1486     def test_mh_to_maildir(self):
1487         # Convert MHMessage to MaildirMessage
1488         pairs = (('unseen', ''), ('replied', 'RS'), ('flagged', 'FS'))
1489         for setting, result in pairs:
1490             msg = mailbox.MHMessage(_sample_message)
1491             msg.add_sequence(setting)
1492             self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), result)
1493             self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur')
1494         msg = mailbox.MHMessage(_sample_message)
1495         msg.add_sequence('unseen')
1496         msg.add_sequence('replied')
1497         msg.add_sequence('flagged')
1498         self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), 'FR')
1499         self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur')
1500
1501     def test_mh_to_mboxmmdf(self):
1502         # Convert MHMessage to mboxMessage and MMDFMessage
1503         pairs = (('unseen', 'O'), ('replied', 'ROA'), ('flagged', 'ROF'))
1504         for setting, result in pairs:
1505             msg = mailbox.MHMessage(_sample_message)
1506             msg.add_sequence(setting)
1507             for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1508                 self.assertEqual(class_(msg).get_flags(), result)
1509         msg = mailbox.MHMessage(_sample_message)
1510         msg.add_sequence('unseen')
1511         msg.add_sequence('replied')
1512         msg.add_sequence('flagged')
1513         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1514             self.assertEqual(class_(msg).get_flags(), 'OFA')
1515
1516     def test_mh_to_mh(self):
1517         # Convert MHMessage to MHMessage
1518         msg = mailbox.MHMessage(_sample_message)
1519         msg.add_sequence('unseen')
1520         msg.add_sequence('replied')
1521         msg.add_sequence('flagged')
1522         self.assertEqual(mailbox.MHMessage(msg).get_sequences(),
1523                          ['unseen', 'replied', 'flagged'])
1524
1525     def test_mh_to_babyl(self):
1526         # Convert MHMessage to BabylMessage
1527         pairs = (('unseen', ['unseen']), ('replied', ['answered']),
1528                  ('flagged', []))
1529         for setting, result in pairs:
1530             msg = mailbox.MHMessage(_sample_message)
1531             msg.add_sequence(setting)
1532             self.assertEqual(mailbox.BabylMessage(msg).get_labels(), result)
1533         msg = mailbox.MHMessage(_sample_message)
1534         msg.add_sequence('unseen')
1535         msg.add_sequence('replied')
1536         msg.add_sequence('flagged')
1537         self.assertEqual(mailbox.BabylMessage(msg).get_labels(),
1538                          ['unseen', 'answered'])
1539
1540     def test_babyl_to_maildir(self):
1541         # Convert BabylMessage to MaildirMessage
1542         pairs = (('unseen', ''), ('deleted', 'ST'), ('filed', 'S'),
1543                  ('answered', 'RS'), ('forwarded', 'PS'), ('edited', 'S'),
1544                  ('resent', 'PS'))
1545         for setting, result in pairs:
1546             msg = mailbox.BabylMessage(_sample_message)
1547             msg.add_label(setting)
1548             self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), result)
1549             self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur')
1550         msg = mailbox.BabylMessage(_sample_message)
1551         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1552                       'edited', 'resent'):
1553             msg.add_label(label)
1554         self.assertEqual(mailbox.MaildirMessage(msg).get_flags(), 'PRT')
1555         self.assertEqual(mailbox.MaildirMessage(msg).get_subdir(), 'cur')
1556
1557     def test_babyl_to_mboxmmdf(self):
1558         # Convert BabylMessage to mboxMessage and MMDFMessage
1559         pairs = (('unseen', 'O'), ('deleted', 'ROD'), ('filed', 'RO'),
1560                  ('answered', 'ROA'), ('forwarded', 'RO'), ('edited', 'RO'),
1561                  ('resent', 'RO'))
1562         for setting, result in pairs:
1563             for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1564                 msg = mailbox.BabylMessage(_sample_message)
1565                 msg.add_label(setting)
1566                 self.assertEqual(class_(msg).get_flags(), result)
1567         msg = mailbox.BabylMessage(_sample_message)
1568         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1569                       'edited', 'resent'):
1570             msg.add_label(label)
1571         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1572             self.assertEqual(class_(msg).get_flags(), 'ODA')
1573
1574     def test_babyl_to_mh(self):
1575         # Convert BabylMessage to MHMessage
1576         pairs = (('unseen', ['unseen']), ('deleted', []), ('filed', []),
1577                  ('answered', ['replied']), ('forwarded', []), ('edited', []),
1578                  ('resent', []))
1579         for setting, result in pairs:
1580             msg = mailbox.BabylMessage(_sample_message)
1581             msg.add_label(setting)
1582             self.assertEqual(mailbox.MHMessage(msg).get_sequences(), result)
1583         msg = mailbox.BabylMessage(_sample_message)
1584         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1585                       'edited', 'resent'):
1586             msg.add_label(label)
1587         self.assertEqual(mailbox.MHMessage(msg).get_sequences(),
1588                          ['unseen', 'replied'])
1589
1590     def test_babyl_to_babyl(self):
1591         # Convert BabylMessage to BabylMessage
1592         msg = mailbox.BabylMessage(_sample_message)
1593         msg.update_visible()
1594         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1595                       'edited', 'resent'):
1596             msg.add_label(label)
1597         msg2 = mailbox.BabylMessage(msg)
1598         self.assertEqual(msg2.get_labels(), ['unseen', 'deleted', 'filed',
1599                                              'answered', 'forwarded', 'edited',
1600                                              'resent'])
1601         self.assertEqual(msg.get_visible().keys(), msg2.get_visible().keys())
1602         for key in msg.get_visible().keys():
1603             self.assertEqual(msg.get_visible()[key], msg2.get_visible()[key])
1604
1605
1606 class TestProxyFileBase(TestBase):
1607
1608     def _test_read(self, proxy):
1609         # Read by byte
1610         proxy.seek(0)
1611         self.assertEqual(proxy.read(), 'bar')
1612         proxy.seek(1)
1613         self.assertEqual(proxy.read(), 'ar')
1614         proxy.seek(0)
1615         self.assertEqual(proxy.read(2), 'ba')
1616         proxy.seek(1)
1617         self.assertEqual(proxy.read(-1), 'ar')
1618         proxy.seek(2)
1619         self.assertEqual(proxy.read(1000), 'r')
1620
1621     def _test_readline(self, proxy):
1622         # Read by line
1623         proxy.seek(0)
1624         self.assertEqual(proxy.readline(), 'foo' + os.linesep)
1625         self.assertEqual(proxy.readline(), 'bar' + os.linesep)
1626         self.assertEqual(proxy.readline(), 'fred' + os.linesep)
1627         self.assertEqual(proxy.readline(), 'bob')
1628         proxy.seek(2)
1629         self.assertEqual(proxy.readline(), 'o' + os.linesep)
1630         proxy.seek(6 + 2 * len(os.linesep))
1631         self.assertEqual(proxy.readline(), 'fred' + os.linesep)
1632         proxy.seek(6 + 2 * len(os.linesep))
1633         self.assertEqual(proxy.readline(2), 'fr')
1634         self.assertEqual(proxy.readline(-10), 'ed' + os.linesep)
1635
1636     def _test_readlines(self, proxy):
1637         # Read multiple lines
1638         proxy.seek(0)
1639         self.assertEqual(proxy.readlines(), ['foo' + os.linesep,
1640                                             'bar' + os.linesep,
1641                                             'fred' + os.linesep, 'bob'])
1642         proxy.seek(0)
1643         self.assertEqual(proxy.readlines(2), ['foo' + os.linesep])
1644         proxy.seek(3 + len(os.linesep))
1645         self.assertEqual(proxy.readlines(4 + len(os.linesep)),
1646                          ['bar' + os.linesep, 'fred' + os.linesep])
1647         proxy.seek(3)
1648         self.assertEqual(proxy.readlines(1000), [os.linesep, 'bar' + os.linesep,
1649                                                  'fred' + os.linesep, 'bob'])
1650
1651     def _test_iteration(self, proxy):
1652         # Iterate by line
1653         proxy.seek(0)
1654         iterator = iter(proxy)
1655         self.assertEqual(list(iterator),
1656             ['foo' + os.linesep, 'bar' + os.linesep, 'fred' + os.linesep, 'bob'])
1657
1658     def _test_seek_and_tell(self, proxy):
1659         # Seek and use tell to check position
1660         proxy.seek(3)
1661         self.assertEqual(proxy.tell(), 3)
1662         self.assertEqual(proxy.read(len(os.linesep)), os.linesep)
1663         proxy.seek(2, 1)
1664         self.assertEqual(proxy.read(1 + len(os.linesep)), 'r' + os.linesep)
1665         proxy.seek(-3 - len(os.linesep), 2)
1666         self.assertEqual(proxy.read(3), 'bar')
1667         proxy.seek(2, 0)
1668         self.assertEqual(proxy.read(), 'o' + os.linesep + 'bar' + os.linesep)
1669         proxy.seek(100)
1670         self.assertEqual(proxy.read(), '')
1671
1672     def _test_close(self, proxy):
1673         # Close a file
1674         proxy.close()
1675         self.assertRaises(AttributeError, lambda: proxy.close())
1676
1677
1678 class TestProxyFile(TestProxyFileBase):
1679
1680     def setUp(self):
1681         self._path = test_support.TESTFN
1682         self._file = open(self._path, 'wb+')
1683
1684     def tearDown(self):
1685         self._file.close()
1686         self._delete_recursively(self._path)
1687
1688     def test_initialize(self):
1689         # Initialize and check position
1690         self._file.write('foo')
1691         pos = self._file.tell()
1692         proxy0 = mailbox._ProxyFile(self._file)
1693         self.assertEqual(proxy0.tell(), pos)
1694         self.assertEqual(self._file.tell(), pos)
1695         proxy1 = mailbox._ProxyFile(self._file, 0)
1696         self.assertEqual(proxy1.tell(), 0)
1697         self.assertEqual(self._file.tell(), pos)
1698
1699     def test_read(self):
1700         self._file.write('bar')
1701         self._test_read(mailbox._ProxyFile(self._file))
1702
1703     def test_readline(self):
1704         self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep,
1705                                                   os.linesep))
1706         self._test_readline(mailbox._ProxyFile(self._file))
1707
1708     def test_readlines(self):
1709         self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep,
1710                                                   os.linesep))
1711         self._test_readlines(mailbox._ProxyFile(self._file))
1712
1713     def test_iteration(self):
1714         self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep,
1715                                                   os.linesep))
1716         self._test_iteration(mailbox._ProxyFile(self._file))
1717
1718     def test_seek_and_tell(self):
1719         self._file.write('foo%sbar%s' % (os.linesep, os.linesep))
1720         self._test_seek_and_tell(mailbox._ProxyFile(self._file))
1721
1722     def test_close(self):
1723         self._file.write('foo%sbar%s' % (os.linesep, os.linesep))
1724         self._test_close(mailbox._ProxyFile(self._file))
1725
1726
1727 class TestPartialFile(TestProxyFileBase):
1728
1729     def setUp(self):
1730         self._path = test_support.TESTFN
1731         self._file = open(self._path, 'wb+')
1732
1733     def tearDown(self):
1734         self._file.close()
1735         self._delete_recursively(self._path)
1736
1737     def test_initialize(self):
1738         # Initialize and check position
1739         self._file.write('foo' + os.linesep + 'bar')
1740         pos = self._file.tell()
1741         proxy = mailbox._PartialFile(self._file, 2, 5)
1742         self.assertEqual(proxy.tell(), 0)
1743         self.assertEqual(self._file.tell(), pos)
1744
1745     def test_read(self):
1746         self._file.write('***bar***')
1747         self._test_read(mailbox._PartialFile(self._file, 3, 6))
1748
1749     def test_readline(self):
1750         self._file.write('!!!!!foo%sbar%sfred%sbob!!!!!' %
1751                          (os.linesep, os.linesep, os.linesep))
1752         self._test_readline(mailbox._PartialFile(self._file, 5,
1753                                                  18 + 3 * len(os.linesep)))
1754
1755     def test_readlines(self):
1756         self._file.write('foo%sbar%sfred%sbob?????' %
1757                          (os.linesep, os.linesep, os.linesep))
1758         self._test_readlines(mailbox._PartialFile(self._file, 0,
1759                                                   13 + 3 * len(os.linesep)))
1760
1761     def test_iteration(self):
1762         self._file.write('____foo%sbar%sfred%sbob####' %
1763                          (os.linesep, os.linesep, os.linesep))
1764         self._test_iteration(mailbox._PartialFile(self._file, 4,
1765                                                   17 + 3 * len(os.linesep)))
1766
1767     def test_seek_and_tell(self):
1768         self._file.write('(((foo%sbar%s$$$' % (os.linesep, os.linesep))
1769         self._test_seek_and_tell(mailbox._PartialFile(self._file, 3,
1770                                                       9 + 2 * len(os.linesep)))
1771
1772     def test_close(self):
1773         self._file.write('&foo%sbar%s^' % (os.linesep, os.linesep))
1774         self._test_close(mailbox._PartialFile(self._file, 1,
1775                                               6 + 3 * len(os.linesep)))
1776
1777
1778 ## Start: tests from the original module (for backward compatibility).
1779
1780 FROM_ = "From some.body@dummy.domain  Sat Jul 24 13:43:35 2004\n"
1781 DUMMY_MESSAGE = """\
1782 From: some.body@dummy.domain
1783 To: me@my.domain
1784 Subject: Simple Test
1785
1786 This is a dummy message.
1787 """
1788
1789 class MaildirTestCase(unittest.TestCase):
1790
1791     def setUp(self):
1792         # create a new maildir mailbox to work with:
1793         self._dir = test_support.TESTFN
1794         os.mkdir(self._dir)
1795         os.mkdir(os.path.join(self._dir, "cur"))
1796         os.mkdir(os.path.join(self._dir, "tmp"))
1797         os.mkdir(os.path.join(self._dir, "new"))
1798         self._counter = 1
1799         self._msgfiles = []
1800
1801     def tearDown(self):
1802         map(os.unlink, self._msgfiles)
1803         os.rmdir(os.path.join(self._dir, "cur"))
1804         os.rmdir(os.path.join(self._dir, "tmp"))
1805         os.rmdir(os.path.join(self._dir, "new"))
1806         os.rmdir(self._dir)
1807
1808     def createMessage(self, dir, mbox=False):
1809         t = int(time.time() % 1000000)
1810         pid = self._counter
1811         self._counter += 1
1812         filename = os.extsep.join((str(t), str(pid), "myhostname", "mydomain"))
1813         tmpname = os.path.join(self._dir, "tmp", filename)
1814         newname = os.path.join(self._dir, dir, filename)
1815         fp = open(tmpname, "w")
1816         self._msgfiles.append(tmpname)
1817         if mbox:
1818             fp.write(FROM_)
1819         fp.write(DUMMY_MESSAGE)
1820         fp.close()
1821         if hasattr(os, "link"):
1822             os.link(tmpname, newname)
1823         else:
1824             fp = open(newname, "w")
1825             fp.write(DUMMY_MESSAGE)
1826             fp.close()
1827         self._msgfiles.append(newname)
1828         return tmpname
1829
1830     def test_empty_maildir(self):
1831         """Test an empty maildir mailbox"""
1832         # Test for regression on bug #117490:
1833         # Make sure the boxes attribute actually gets set.
1834         self.mbox = mailbox.Maildir(test_support.TESTFN)
1835         #self.assertTrue(hasattr(self.mbox, "boxes"))
1836         #self.assertTrue(len(self.mbox.boxes) == 0)
1837         self.assertIs(self.mbox.next(), None)
1838         self.assertIs(self.mbox.next(), None)
1839
1840     def test_nonempty_maildir_cur(self):
1841         self.createMessage("cur")
1842         self.mbox = mailbox.Maildir(test_support.TESTFN)
1843         #self.assertTrue(len(self.mbox.boxes) == 1)
1844         self.assertIsNot(self.mbox.next(), None)
1845         self.assertIs(self.mbox.next(), None)
1846         self.assertIs(self.mbox.next(), None)
1847
1848     def test_nonempty_maildir_new(self):
1849         self.createMessage("new")
1850         self.mbox = mailbox.Maildir(test_support.TESTFN)
1851         #self.assertTrue(len(self.mbox.boxes) == 1)
1852         self.assertIsNot(self.mbox.next(), None)
1853         self.assertIs(self.mbox.next(), None)
1854         self.assertIs(self.mbox.next(), None)
1855
1856     def test_nonempty_maildir_both(self):
1857         self.createMessage("cur")
1858         self.createMessage("new")
1859         self.mbox = mailbox.Maildir(test_support.TESTFN)
1860         #self.assertTrue(len(self.mbox.boxes) == 2)
1861         self.assertIsNot(self.mbox.next(), None)
1862         self.assertIsNot(self.mbox.next(), None)
1863         self.assertIs(self.mbox.next(), None)
1864         self.assertIs(self.mbox.next(), None)
1865
1866     def test_unix_mbox(self):
1867         ### should be better!
1868         import email.parser
1869         fname = self.createMessage("cur", True)
1870         n = 0
1871         for msg in mailbox.PortableUnixMailbox(open(fname),
1872                                                email.parser.Parser().parse):
1873             n += 1
1874             self.assertEqual(msg["subject"], "Simple Test")
1875             self.assertEqual(len(str(msg)), len(FROM_)+len(DUMMY_MESSAGE))
1876         self.assertEqual(n, 1)
1877
1878 ## End: classes from the original module (for backward compatibility).
1879
1880
1881 _sample_message = """\
1882 Return-Path: <gkj@gregorykjohnson.com>
1883 X-Original-To: gkj+person@localhost
1884 Delivered-To: gkj+person@localhost
1885 Received: from localhost (localhost [127.0.0.1])
1886         by andy.gregorykjohnson.com (Postfix) with ESMTP id 356ED9DD17
1887         for <gkj+person@localhost>; Wed, 13 Jul 2005 17:23:16 -0400 (EDT)
1888 Delivered-To: gkj@sundance.gregorykjohnson.com
1889 Received: from localhost [127.0.0.1]
1890         by localhost with POP3 (fetchmail-6.2.5)
1891         for gkj+person@localhost (single-drop); Wed, 13 Jul 2005 17:23:16 -0400 (EDT)
1892 Received: from andy.gregorykjohnson.com (andy.gregorykjohnson.com [64.32.235.228])
1893         by sundance.gregorykjohnson.com (Postfix) with ESMTP id 5B056316746
1894         for <gkj@gregorykjohnson.com>; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)
1895 Received: by andy.gregorykjohnson.com (Postfix, from userid 1000)
1896         id 490CD9DD17; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)
1897 Date: Wed, 13 Jul 2005 17:23:11 -0400
1898 From: "Gregory K. Johnson" <gkj@gregorykjohnson.com>
1899 To: gkj@gregorykjohnson.com
1900 Subject: Sample message
1901 Message-ID: <20050713212311.GC4701@andy.gregorykjohnson.com>
1902 Mime-Version: 1.0
1903 Content-Type: multipart/mixed; boundary="NMuMz9nt05w80d4+"
1904 Content-Disposition: inline
1905 User-Agent: Mutt/1.5.9i
1906
1907
1908 --NMuMz9nt05w80d4+
1909 Content-Type: text/plain; charset=us-ascii
1910 Content-Disposition: inline
1911
1912 This is a sample message.
1913
1914 --
1915 Gregory K. Johnson
1916
1917 --NMuMz9nt05w80d4+
1918 Content-Type: application/octet-stream
1919 Content-Disposition: attachment; filename="text.gz"
1920 Content-Transfer-Encoding: base64
1921
1922 H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs
1923 3FYlAAAA
1924
1925 --NMuMz9nt05w80d4+--
1926 """
1927
1928 _sample_headers = {
1929     "Return-Path":"<gkj@gregorykjohnson.com>",
1930     "X-Original-To":"gkj+person@localhost",
1931     "Delivered-To":"gkj+person@localhost",
1932     "Received":"""from localhost (localhost [127.0.0.1])
1933         by andy.gregorykjohnson.com (Postfix) with ESMTP id 356ED9DD17
1934         for <gkj+person@localhost>; Wed, 13 Jul 2005 17:23:16 -0400 (EDT)""",
1935     "Delivered-To":"gkj@sundance.gregorykjohnson.com",
1936     "Received":"""from localhost [127.0.0.1]
1937         by localhost with POP3 (fetchmail-6.2.5)
1938         for gkj+person@localhost (single-drop); Wed, 13 Jul 2005 17:23:16 -0400 (EDT)""",
1939     "Received":"""from andy.gregorykjohnson.com (andy.gregorykjohnson.com [64.32.235.228])
1940         by sundance.gregorykjohnson.com (Postfix) with ESMTP id 5B056316746
1941         for <gkj@gregorykjohnson.com>; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)""",
1942     "Received":"""by andy.gregorykjohnson.com (Postfix, from userid 1000)
1943         id 490CD9DD17; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)""",
1944     "Date":"Wed, 13 Jul 2005 17:23:11 -0400",
1945     "From":""""Gregory K. Johnson" <gkj@gregorykjohnson.com>""",
1946     "To":"gkj@gregorykjohnson.com",
1947     "Subject":"Sample message",
1948     "Mime-Version":"1.0",
1949     "Content-Type":"""multipart/mixed; boundary="NMuMz9nt05w80d4+\"""",
1950     "Content-Disposition":"inline",
1951     "User-Agent": "Mutt/1.5.9i" }
1952
1953 _sample_payloads = ("""This is a sample message.
1954
1955 --
1956 Gregory K. Johnson
1957 """,
1958 """H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs
1959 3FYlAAAA
1960 """)
1961
1962
1963 def test_main():
1964     tests = (TestMailboxSuperclass, TestMaildir, TestMbox, TestMMDF, TestMH,
1965              TestBabyl, TestMessage, TestMaildirMessage, TestMboxMessage,
1966              TestMHMessage, TestBabylMessage, TestMMDFMessage,
1967              TestMessageConversion, TestProxyFile, TestPartialFile,
1968              MaildirTestCase)
1969     test_support.run_unittest(*tests)
1970     test_support.reap_children()
1971
1972
1973 if __name__ == '__main__':
1974     test_main()