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