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_initialize_new(self):
 512         # Initialize a non-existent mailbox
 513         self.tearDown()
 514         self._box = mailbox.Maildir(self._path)
 515         self._check_basics(factory=rfc822.Message)
 516         self._delete_recursively(self._path)
 517         self._box = self._factory(self._path, factory=None)
 518         self._check_basics()
 519
 520     def test_initialize_existing(self):
 521         # Initialize an existing mailbox
 522         self.tearDown()
 523         for subdir in '', 'tmp', 'new', 'cur':
 524             os.mkdir(os.path.normpath(os.path.join(self._path, subdir)))
 525         self._box = mailbox.Maildir(self._path)
 526         self._check_basics(factory=rfc822.Message)
 527         self._box = mailbox.Maildir(self._path, factory=None)
 528         self._check_basics()
 529
 530     def _check_basics(self, factory=None):
 531         # (Used by test_open_new() and test_open_existing().)
 532         self.assertEqual(self._box._path, os.path.abspath(self._path))
 533         self.assertEqual(self._box._factory, factory)
 534         for subdir in '', 'tmp', 'new', 'cur':
 535             path = os.path.join(self._path, subdir)
 536             mode = os.stat(path)[stat.ST_MODE]
 537             self.assert_(stat.S_ISDIR(mode), "Not a directory: '%s'" % path)
 538
 539     def test_list_folders(self):
 540         # List folders
 541         self._box.add_folder('one')
 542         self._box.add_folder('two')
 543         self._box.add_folder('three')
 544         self.assert_(len(self._box.list_folders()) == 3)
 545         self.assert_(set(self._box.list_folders()) ==
 546                      set(('one', 'two', 'three')))
 547
 548     def test_get_folder(self):
 549         # Open folders
 550         self._box.add_folder('foo.bar')
 551         folder0 = self._box.get_folder('foo.bar')
 552         folder0.add(self._template % 'bar')
 553         self.assert_(os.path.isdir(os.path.join(self._path, '.foo.bar')))
 554         folder1 = self._box.get_folder('foo.bar')
 555         self.assert_(folder1.get_string(folder1.keys()[0]) == \
 556                      self._template % 'bar')
 557
 558     def test_add_and_remove_folders(self):
 559         # Delete folders
 560         self._box.add_folder('one')
 561         self._box.add_folder('two')
 562         self.assert_(len(self._box.list_folders()) == 2)
 563         self.assert_(set(self._box.list_folders()) == set(('one', 'two')))
 564         self._box.remove_folder('one')
 565         self.assert_(len(self._box.list_folders()) == 1)
 566         self.assert_(set(self._box.list_folders()) == set(('two',)))
 567         self._box.add_folder('three')
 568         self.assert_(len(self._box.list_folders()) == 2)
 569         self.assert_(set(self._box.list_folders()) == set(('two', 'three')))
 570         self._box.remove_folder('three')
 571         self.assert_(len(self._box.list_folders()) == 1)
 572         self.assert_(set(self._box.list_folders()) == set(('two',)))
 573         self._box.remove_folder('two')
 574         self.assert_(len(self._box.list_folders()) == 0)
 575         self.assert_(self._box.list_folders() == [])
 576
 577     def test_clean(self):
 578         # Remove old files from 'tmp'
 579         foo_path = os.path.join(self._path, 'tmp', 'foo')
 580         bar_path = os.path.join(self._path, 'tmp', 'bar')
 581         f = open(foo_path, 'w')
 582         f.write("@")
 583         f.close()
 584         f = open(bar_path, 'w')
 585         f.write("@")
 586         f.close()
 587         self._box.clean()
 588         self.assert_(os.path.exists(foo_path))
 589         self.assert_(os.path.exists(bar_path))
 590         foo_stat = os.stat(foo_path)
 591         os.utime(foo_path, (time.time() - 129600 - 2,
 592                             foo_stat.st_mtime))
 593         self._box.clean()
 594         self.assert_(not os.path.exists(foo_path))
 595         self.assert_(os.path.exists(bar_path))
 596
 597     def test_create_tmp(self, repetitions=10):
 598         # Create files in tmp directory
 599         hostname = socket.gethostname()
 600         if '/' in hostname:
 601             hostname = hostname.replace('/', r'\057')
 602         if ':' in hostname:
 603             hostname = hostname.replace(':', r'\072')
 604         pid = os.getpid()
 605         pattern = re.compile(r"(?P<time>\d+)\.M(?P<M>\d{1,6})P(?P<P>\d+)"
 606                              r"Q(?P<Q>\d+)\.(?P<host>[^:/]+)")
 607         previous_groups = None
 608         for x in xrange(repetitions):
 609             tmp_file = self._box._create_tmp()
 610             head, tail = os.path.split(tmp_file.name)
 611             self.assertEqual(head, os.path.abspath(os.path.join(self._path,
 612                                                                 "tmp")),
 613                              "File in wrong location: '%s'" % head)
 614             match = pattern.match(tail)
 615             self.assert_(match != None, "Invalid file name: '%s'" % tail)
 616             groups = match.groups()
 617             if previous_groups != None:
 618                 self.assert_(int(groups[0] >= previous_groups[0]),
 619                              "Non-monotonic seconds: '%s' before '%s'" %
 620                              (previous_groups[0], groups[0]))
 621                 self.assert_(int(groups[1] >= previous_groups[1]) or
 622                              groups[0] != groups[1],
 623                              "Non-monotonic milliseconds: '%s' before '%s'" %
 624                              (previous_groups[1], groups[1]))
 625                 self.assert_(int(groups[2]) == pid,
 626                              "Process ID mismatch: '%s' should be '%s'" %
 627                              (groups[2], pid))
 628                 self.assert_(int(groups[3]) == int(previous_groups[3]) + 1,
 629                              "Non-sequential counter: '%s' before '%s'" %
 630                              (previous_groups[3], groups[3]))
 631                 self.assert_(groups[4] == hostname,
 632                              "Host name mismatch: '%s' should be '%s'" %
 633                              (groups[4], hostname))
 634             previous_groups = groups
 635             tmp_file.write(_sample_message)
 636             tmp_file.seek(0)
 637             self.assert_(tmp_file.read() == _sample_message)
 638             tmp_file.close()
 639         file_count = len(os.listdir(os.path.join(self._path, "tmp")))
 640         self.assert_(file_count == repetitions,
 641                      "Wrong file count: '%s' should be '%s'" %
 642                      (file_count, repetitions))
 643
 644     def test_refresh(self):
 645         # Update the table of contents
 646         self.assert_(self._box._toc == {})
 647         key0 = self._box.add(self._template % 0)
 648         key1 = self._box.add(self._template % 1)
 649         self.assert_(self._box._toc == {})
 650         self._box._refresh()
 651         self.assert_(self._box._toc == {key0: os.path.join('new', key0),
 652                                         key1: os.path.join('new', key1)})
 653         key2 = self._box.add(self._template % 2)
 654         self.assert_(self._box._toc == {key0: os.path.join('new', key0),
 655                                         key1: os.path.join('new', key1)})
 656         self._box._refresh()
 657         self.assert_(self._box._toc == {key0: os.path.join('new', key0),
 658                                         key1: os.path.join('new', key1),
 659                                         key2: os.path.join('new', key2)})
 660
 661     def test_lookup(self):
 662         # Look up message subpaths in the TOC
 663         self.assertRaises(KeyError, lambda: self._box._lookup('foo'))
 664         key0 = self._box.add(self._template % 0)
 665         self.assert_(self._box._lookup(key0) == os.path.join('new', key0))
 666         os.remove(os.path.join(self._path, 'new', key0))
 667         self.assert_(self._box._toc == {key0: os.path.join('new', key0)})
 668         self.assertRaises(KeyError, lambda: self._box._lookup(key0))
 669         self.assert_(self._box._toc == {})
 670
 671     def test_lock_unlock(self):
 672         # Lock and unlock the mailbox. For Maildir, this does nothing.
 673         self._box.lock()
 674         self._box.unlock()
 675
 676
 677 class _TestMboxMMDF(TestMailbox):
 678
 679     def tearDown(self):
 680         self._box.close()
 681         self._delete_recursively(self._path)
 682         for lock_remnant in glob.glob(self._path + '.*'):
 683             os.remove(lock_remnant)
 684
 685     def test_add_from_string(self):
 686         # Add a string starting with 'From ' to the mailbox
 687         key = self._box.add('From foo@bar blah\nFrom: foo\n\n0')
 688         self.assert_(self._box[key].get_from() == 'foo@bar blah')
 689         self.assert_(self._box[key].get_payload() == '0')
 690
 691     def test_add_mbox_or_mmdf_message(self):
 692         # Add an mboxMessage or MMDFMessage
 693         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
 694             msg = class_('From foo@bar blah\nFrom: foo\n\n0')
 695             key = self._box.add(msg)
 696
 697     def test_open_close_open(self):
 698         # Open and inspect previously-created mailbox
 699         values = [self._template % i for i in xrange(3)]
 700         for value in values:
 701             self._box.add(value)
 702         self._box.close()
 703         mtime = os.path.getmtime(self._path)
 704         self._box = self._factory(self._path)
 705         self.assert_(len(self._box) == 3)
 706         for key in self._box.iterkeys():
 707             self.assert_(self._box.get_string(key) in values)
 708         self._box.close()
 709         self.assert_(mtime == os.path.getmtime(self._path))
 710
 711     def test_add_and_close(self):
 712         # Verifying that closing a mailbox doesn't change added items
 713         self._box.add(_sample_message)
 714         for i in xrange(3):
 715             self._box.add(self._template % i)
 716         self._box.add(_sample_message)
 717         self._box._file.flush()
 718         self._box._file.seek(0)
 719         contents = self._box._file.read()
 720         self._box.close()
 721         self.assert_(contents == open(self._path, 'rb').read())
 722         self._box = self._factory(self._path)
 723
 724     def test_lock_conflict(self):
 725         # Fork off a subprocess that will lock the file for 2 seconds,
 726         # unlock it, and then exit.
 727         if not hasattr(os, 'fork'):
 728             return
 729         pid = os.fork()
 730         if pid == 0:
 731             # In the child, lock the mailbox.
 732             self._box.lock()
 733             time.sleep(2)
 734             self._box.unlock()
 735             os._exit(0)
 736
 737         # In the parent, sleep a bit to give the child time to acquire
 738         # the lock.
 739         time.sleep(0.5)
 740         try:
 741             self.assertRaises(mailbox.ExternalClashError,
 742                               self._box.lock)
 743         finally:
 744             # Wait for child to exit.  Locking should now succeed.
 745             exited_pid, status = os.waitpid(pid, 0)
 746
 747         self._box.lock()
 748         self._box.unlock()
 749
 750     def test_relock(self):
 751         # Test case for bug #1575506: the mailbox class was locking the
 752         # wrong file object in its flush() method.
 753         msg = "Subject: sub\n\nbody\n"
 754         key1 = self._box.add(msg)
 755         self._box.flush()
 756         self._box.close()
 757
 758         self._box = self._factory(self._path)
 759         self._box.lock()
 760         key2 = self._box.add(msg)
 761         self._box.flush()
 762         self.assert_(self._box._locked)
 763         self._box.close()
 764
 765
 766
 767 class TestMbox(_TestMboxMMDF):
 768
 769     _factory = lambda self, path, factory=None: mailbox.mbox(path, factory)
 770
 771
 772 class TestMMDF(_TestMboxMMDF):
 773
 774     _factory = lambda self, path, factory=None: mailbox.MMDF(path, factory)
 775
 776
 777 class TestMH(TestMailbox):
 778
 779     _factory = lambda self, path, factory=None: mailbox.MH(path, factory)
 780
 781     def test_list_folders(self):
 782         # List folders
 783         self._box.add_folder('one')
 784         self._box.add_folder('two')
 785         self._box.add_folder('three')
 786         self.assert_(len(self._box.list_folders()) == 3)
 787         self.assert_(set(self._box.list_folders()) ==
 788                      set(('one', 'two', 'three')))
 789
 790     def test_get_folder(self):
 791         # Open folders
 792         self._box.add_folder('foo.bar')
 793         folder0 = self._box.get_folder('foo.bar')
 794         folder0.add(self._template % 'bar')
 795         self.assert_(os.path.isdir(os.path.join(self._path, 'foo.bar')))
 796         folder1 = self._box.get_folder('foo.bar')
 797         self.assert_(folder1.get_string(folder1.keys()[0]) == \
 798                      self._template % 'bar')
 799
 800     def test_add_and_remove_folders(self):
 801         # Delete folders
 802         self._box.add_folder('one')
 803         self._box.add_folder('two')
 804         self.assert_(len(self._box.list_folders()) == 2)
 805         self.assert_(set(self._box.list_folders()) == set(('one', 'two')))
 806         self._box.remove_folder('one')
 807         self.assert_(len(self._box.list_folders()) == 1)
 808         self.assert_(set(self._box.list_folders()) == set(('two',)))
 809         self._box.add_folder('three')
 810         self.assert_(len(self._box.list_folders()) == 2)
 811         self.assert_(set(self._box.list_folders()) == set(('two', 'three')))
 812         self._box.remove_folder('three')
 813         self.assert_(len(self._box.list_folders()) == 1)
 814         self.assert_(set(self._box.list_folders()) == set(('two',)))
 815         self._box.remove_folder('two')
 816         self.assert_(len(self._box.list_folders()) == 0)
 817         self.assert_(self._box.list_folders() == [])
 818
 819     def test_sequences(self):
 820         # Get and set sequences
 821         self.assert_(self._box.get_sequences() == {})
 822         msg0 = mailbox.MHMessage(self._template % 0)
 823         msg0.add_sequence('foo')
 824         key0 = self._box.add(msg0)
 825         self.assert_(self._box.get_sequences() == {'foo':[key0]})
 826         msg1 = mailbox.MHMessage(self._template % 1)
 827         msg1.set_sequences(['bar', 'replied', 'foo'])
 828         key1 = self._box.add(msg1)
 829         self.assert_(self._box.get_sequences() ==
 830                      {'foo':[key0, key1], 'bar':[key1], 'replied':[key1]})
 831         msg0.set_sequences(['flagged'])
 832         self._box[key0] = msg0
 833         self.assert_(self._box.get_sequences() ==
 834                      {'foo':[key1], 'bar':[key1], 'replied':[key1],
 835                       'flagged':[key0]})
 836         self._box.remove(key1)
 837         self.assert_(self._box.get_sequences() == {'flagged':[key0]})
 838
 839     def test_pack(self):
 840         # Pack the contents of the mailbox
 841         msg0 = mailbox.MHMessage(self._template % 0)
 842         msg1 = mailbox.MHMessage(self._template % 1)
 843         msg2 = mailbox.MHMessage(self._template % 2)
 844         msg3 = mailbox.MHMessage(self._template % 3)
 845         msg0.set_sequences(['foo', 'unseen'])
 846         msg1.set_sequences(['foo'])
 847         msg2.set_sequences(['foo', 'flagged'])
 848         msg3.set_sequences(['foo', 'bar', 'replied'])
 849         key0 = self._box.add(msg0)
 850         key1 = self._box.add(msg1)
 851         key2 = self._box.add(msg2)
 852         key3 = self._box.add(msg3)
 853         self.assert_(self._box.get_sequences() ==
 854                      {'foo':[key0,key1,key2,key3], 'unseen':[key0],
 855                       'flagged':[key2], 'bar':[key3], 'replied':[key3]})
 856         self._box.remove(key2)
 857         self.assert_(self._box.get_sequences() ==
 858                      {'foo':[key0,key1,key3], 'unseen':[key0], 'bar':[key3],
 859                       'replied':[key3]})
 860         self._box.pack()
 861         self.assert_(self._box.keys() == [1, 2, 3])
 862         key0 = key0
 863         key1 = key0 + 1
 864         key2 = key1 + 1
 865         self.assert_(self._box.get_sequences() ==
 866                      {'foo':[1, 2, 3], 'unseen':[1], 'bar':[3], 'replied':[3]})
 867
 868     def _get_lock_path(self):
 869         return os.path.join(self._path, '.mh_sequences.lock')
 870
 871
 872 class TestBabyl(TestMailbox):
 873
 874     _factory = lambda self, path, factory=None: mailbox.Babyl(path, factory)
 875
 876     def tearDown(self):
 877         self._box.close()
 878         self._delete_recursively(self._path)
 879         for lock_remnant in glob.glob(self._path + '.*'):
 880             os.remove(lock_remnant)
 881
 882     def test_labels(self):
 883         # Get labels from the mailbox
 884         self.assert_(self._box.get_labels() == [])
 885         msg0 = mailbox.BabylMessage(self._template % 0)
 886         msg0.add_label('foo')
 887         key0 = self._box.add(msg0)
 888         self.assert_(self._box.get_labels() == ['foo'])
 889         msg1 = mailbox.BabylMessage(self._template % 1)
 890         msg1.set_labels(['bar', 'answered', 'foo'])
 891         key1 = self._box.add(msg1)
 892         self.assert_(set(self._box.get_labels()) == set(['foo', 'bar']))
 893         msg0.set_labels(['blah', 'filed'])
 894         self._box[key0] = msg0
 895         self.assert_(set(self._box.get_labels()) ==
 896                      set(['foo', 'bar', 'blah']))
 897         self._box.remove(key1)
 898         self.assert_(set(self._box.get_labels()) == set(['blah']))
 899
 900
 901 class TestMessage(TestBase):
 902
 903     _factory = mailbox.Message      # Overridden by subclasses to reuse tests
 904
 905     def setUp(self):
 906         self._path = test_support.TESTFN
 907
 908     def tearDown(self):
 909         self._delete_recursively(self._path)
 910
 911     def test_initialize_with_eMM(self):
 912         # Initialize based on email.Message.Message instance
 913         eMM = email.message_from_string(_sample_message)
 914         msg = self._factory(eMM)
 915         self._post_initialize_hook(msg)
 916         self._check_sample(msg)
 917
 918     def test_initialize_with_string(self):
 919         # Initialize based on string
 920         msg = self._factory(_sample_message)
 921         self._post_initialize_hook(msg)
 922         self._check_sample(msg)
 923
 924     def test_initialize_with_file(self):
 925         # Initialize based on contents of file
 926         f = open(self._path, 'w+')
 927         f.write(_sample_message)
 928         f.seek(0)
 929         msg = self._factory(f)
 930         self._post_initialize_hook(msg)
 931         self._check_sample(msg)
 932         f.close()
 933
 934     def test_initialize_with_nothing(self):
 935         # Initialize without arguments
 936         msg = self._factory()
 937         self._post_initialize_hook(msg)
 938         self.assert_(isinstance(msg, email.Message.Message))
 939         self.assert_(isinstance(msg, mailbox.Message))
 940         self.assert_(isinstance(msg, self._factory))
 941         self.assert_(msg.keys() == [])
 942         self.assert_(not msg.is_multipart())
 943         self.assert_(msg.get_payload() == None)
 944
 945     def test_initialize_incorrectly(self):
 946         # Initialize with invalid argument
 947         self.assertRaises(TypeError, lambda: self._factory(object()))
 948
 949     def test_become_message(self):
 950         # Take on the state of another message
 951         eMM = email.message_from_string(_sample_message)
 952         msg = self._factory()
 953         msg._become_message(eMM)
 954         self._check_sample(msg)
 955
 956     def test_explain_to(self):
 957         # Copy self's format-specific data to other message formats.
 958         # This test is superficial; better ones are in TestMessageConversion.
 959         msg = self._factory()
 960         for class_ in (mailbox.Message, mailbox.MaildirMessage,
 961                        mailbox.mboxMessage, mailbox.MHMessage,
 962                        mailbox.BabylMessage, mailbox.MMDFMessage):
 963             other_msg = class_()
 964             msg._explain_to(other_msg)
 965         other_msg = email.Message.Message()
 966         self.assertRaises(TypeError, lambda: msg._explain_to(other_msg))
 967
 968     def _post_initialize_hook(self, msg):
 969         # Overridden by subclasses to check extra things after initialization
 970         pass
 971
 972
 973 class TestMaildirMessage(TestMessage):
 974
 975     _factory = mailbox.MaildirMessage
 976
 977     def _post_initialize_hook(self, msg):
 978         self.assert_(msg._subdir == 'new')
 979         self.assert_(msg._info == '')
 980
 981     def test_subdir(self):
 982         # Use get_subdir() and set_subdir()
 983         msg = mailbox.MaildirMessage(_sample_message)
 984         self.assert_(msg.get_subdir() == 'new')
 985         msg.set_subdir('cur')
 986         self.assert_(msg.get_subdir() == 'cur')
 987         msg.set_subdir('new')
 988         self.assert_(msg.get_subdir() == 'new')
 989         self.assertRaises(ValueError, lambda: msg.set_subdir('tmp'))
 990         self.assert_(msg.get_subdir() == 'new')
 991         msg.set_subdir('new')
 992         self.assert_(msg.get_subdir() == 'new')
 993         self._check_sample(msg)
 994
 995     def test_flags(self):
 996         # Use get_flags(), set_flags(), add_flag(), remove_flag()
 997         msg = mailbox.MaildirMessage(_sample_message)
 998         self.assert_(msg.get_flags() == '')
 999         self.assert_(msg.get_subdir() == 'new')
1000         msg.set_flags('F')
1001         self.assert_(msg.get_subdir() == 'new')
1002         self.assert_(msg.get_flags() == 'F')
1003         msg.set_flags('SDTP')
1004         self.assert_(msg.get_flags() == 'DPST')
1005         msg.add_flag('FT')
1006         self.assert_(msg.get_flags() == 'DFPST')
1007         msg.remove_flag('TDRP')
1008         self.assert_(msg.get_flags() == 'FS')
1009         self.assert_(msg.get_subdir() == 'new')
1010         self._check_sample(msg)
1011
1012     def test_date(self):
1013         # Use get_date() and set_date()
1014         msg = mailbox.MaildirMessage(_sample_message)
1015         self.assert_(abs(msg.get_date() - time.time()) < 60)
1016         msg.set_date(0.0)
1017         self.assert_(msg.get_date() == 0.0)
1018
1019     def test_info(self):
1020         # Use get_info() and set_info()
1021         msg = mailbox.MaildirMessage(_sample_message)
1022         self.assert_(msg.get_info() == '')
1023         msg.set_info('1,foo=bar')
1024         self.assert_(msg.get_info() == '1,foo=bar')
1025         self.assertRaises(TypeError, lambda: msg.set_info(None))
1026         self._check_sample(msg)
1027
1028     def test_info_and_flags(self):
1029         # Test interaction of info and flag methods
1030         msg = mailbox.MaildirMessage(_sample_message)
1031         self.assert_(msg.get_info() == '')
1032         msg.set_flags('SF')
1033         self.assert_(msg.get_flags() == 'FS')
1034         self.assert_(msg.get_info() == '2,FS')
1035         msg.set_info('1,')
1036         self.assert_(msg.get_flags() == '')
1037         self.assert_(msg.get_info() == '1,')
1038         msg.remove_flag('RPT')
1039         self.assert_(msg.get_flags() == '')
1040         self.assert_(msg.get_info() == '1,')
1041         msg.add_flag('D')
1042         self.assert_(msg.get_flags() == 'D')
1043         self.assert_(msg.get_info() == '2,D')
1044         self._check_sample(msg)
1045
1046
1047 class _TestMboxMMDFMessage(TestMessage):
1048
1049     _factory = mailbox._mboxMMDFMessage
1050
1051     def _post_initialize_hook(self, msg):
1052         self._check_from(msg)
1053
1054     def test_initialize_with_unixfrom(self):
1055         # Initialize with a message that already has a _unixfrom attribute
1056         msg = mailbox.Message(_sample_message)
1057         msg.set_unixfrom('From foo@bar blah')
1058         msg = mailbox.mboxMessage(msg)
1059         self.assert_(msg.get_from() == 'foo@bar blah', msg.get_from())
1060
1061     def test_from(self):
1062         # Get and set "From " line
1063         msg = mailbox.mboxMessage(_sample_message)
1064         self._check_from(msg)
1065         msg.set_from('foo bar')
1066         self.assert_(msg.get_from() == 'foo bar')
1067         msg.set_from('foo@bar', True)
1068         self._check_from(msg, 'foo@bar')
1069         msg.set_from('blah@temp', time.localtime())
1070         self._check_from(msg, 'blah@temp')
1071
1072     def test_flags(self):
1073         # Use get_flags(), set_flags(), add_flag(), remove_flag()
1074         msg = mailbox.mboxMessage(_sample_message)
1075         self.assert_(msg.get_flags() == '')
1076         msg.set_flags('F')
1077         self.assert_(msg.get_flags() == 'F')
1078         msg.set_flags('XODR')
1079         self.assert_(msg.get_flags() == 'RODX')
1080         msg.add_flag('FA')
1081         self.assert_(msg.get_flags() == 'RODFAX')
1082         msg.remove_flag('FDXA')
1083         self.assert_(msg.get_flags() == 'RO')
1084         self._check_sample(msg)
1085
1086     def _check_from(self, msg, sender=None):
1087         # Check contents of "From " line
1088         if sender is None:
1089             sender = "MAILER-DAEMON"
1090         self.assert_(re.match(sender + r" \w{3} \w{3} [\d ]\d [\d ]\d:\d{2}:"
1091                               r"\d{2} \d{4}", msg.get_from()) is not None)
1092
1093
1094 class TestMboxMessage(_TestMboxMMDFMessage):
1095
1096     _factory = mailbox.mboxMessage
1097
1098
1099 class TestMHMessage(TestMessage):
1100
1101     _factory = mailbox.MHMessage
1102
1103     def _post_initialize_hook(self, msg):
1104         self.assert_(msg._sequences == [])
1105
1106     def test_sequences(self):
1107         # Get, set, join, and leave sequences
1108         msg = mailbox.MHMessage(_sample_message)
1109         self.assert_(msg.get_sequences() == [])
1110         msg.set_sequences(['foobar'])
1111         self.assert_(msg.get_sequences() == ['foobar'])
1112         msg.set_sequences([])
1113         self.assert_(msg.get_sequences() == [])
1114         msg.add_sequence('unseen')
1115         self.assert_(msg.get_sequences() == ['unseen'])
1116         msg.add_sequence('flagged')
1117         self.assert_(msg.get_sequences() == ['unseen', 'flagged'])
1118         msg.add_sequence('flagged')
1119         self.assert_(msg.get_sequences() == ['unseen', 'flagged'])
1120         msg.remove_sequence('unseen')
1121         self.assert_(msg.get_sequences() == ['flagged'])
1122         msg.add_sequence('foobar')
1123         self.assert_(msg.get_sequences() == ['flagged', 'foobar'])
1124         msg.remove_sequence('replied')
1125         self.assert_(msg.get_sequences() == ['flagged', 'foobar'])
1126         msg.set_sequences(['foobar', 'replied'])
1127         self.assert_(msg.get_sequences() == ['foobar', 'replied'])
1128
1129
1130 class TestBabylMessage(TestMessage):
1131
1132     _factory = mailbox.BabylMessage
1133
1134     def _post_initialize_hook(self, msg):
1135         self.assert_(msg._labels == [])
1136
1137     def test_labels(self):
1138         # Get, set, join, and leave labels
1139         msg = mailbox.BabylMessage(_sample_message)
1140         self.assert_(msg.get_labels() == [])
1141         msg.set_labels(['foobar'])
1142         self.assert_(msg.get_labels() == ['foobar'])
1143         msg.set_labels([])
1144         self.assert_(msg.get_labels() == [])
1145         msg.add_label('filed')
1146         self.assert_(msg.get_labels() == ['filed'])
1147         msg.add_label('resent')
1148         self.assert_(msg.get_labels() == ['filed', 'resent'])
1149         msg.add_label('resent')
1150         self.assert_(msg.get_labels() == ['filed', 'resent'])
1151         msg.remove_label('filed')
1152         self.assert_(msg.get_labels() == ['resent'])
1153         msg.add_label('foobar')
1154         self.assert_(msg.get_labels() == ['resent', 'foobar'])
1155         msg.remove_label('unseen')
1156         self.assert_(msg.get_labels() == ['resent', 'foobar'])
1157         msg.set_labels(['foobar', 'answered'])
1158         self.assert_(msg.get_labels() == ['foobar', 'answered'])
1159
1160     def test_visible(self):
1161         # Get, set, and update visible headers
1162         msg = mailbox.BabylMessage(_sample_message)
1163         visible = msg.get_visible()
1164         self.assert_(visible.keys() == [])
1165         self.assert_(visible.get_payload() is None)
1166         visible['User-Agent'] = 'FooBar 1.0'
1167         visible['X-Whatever'] = 'Blah'
1168         self.assert_(msg.get_visible().keys() == [])
1169         msg.set_visible(visible)
1170         visible = msg.get_visible()
1171         self.assert_(visible.keys() == ['User-Agent', 'X-Whatever'])
1172         self.assert_(visible['User-Agent'] == 'FooBar 1.0')
1173         self.assert_(visible['X-Whatever'] == 'Blah')
1174         self.assert_(visible.get_payload() is None)
1175         msg.update_visible()
1176         self.assert_(visible.keys() == ['User-Agent', 'X-Whatever'])
1177         self.assert_(visible.get_payload() is None)
1178         visible = msg.get_visible()
1179         self.assert_(visible.keys() == ['User-Agent', 'Date', 'From', 'To',
1180                                         'Subject'])
1181         for header in ('User-Agent', 'Date', 'From', 'To', 'Subject'):
1182             self.assert_(visible[header] == msg[header])
1183
1184
1185 class TestMMDFMessage(_TestMboxMMDFMessage):
1186
1187     _factory = mailbox.MMDFMessage
1188
1189
1190 class TestMessageConversion(TestBase):
1191
1192     def test_plain_to_x(self):
1193         # Convert Message to all formats
1194         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1195                        mailbox.mboxMessage, mailbox.MHMessage,
1196                        mailbox.BabylMessage, mailbox.MMDFMessage):
1197             msg_plain = mailbox.Message(_sample_message)
1198             msg = class_(msg_plain)
1199             self._check_sample(msg)
1200
1201     def test_x_to_plain(self):
1202         # Convert all formats to Message
1203         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1204                        mailbox.mboxMessage, mailbox.MHMessage,
1205                        mailbox.BabylMessage, mailbox.MMDFMessage):
1206             msg = class_(_sample_message)
1207             msg_plain = mailbox.Message(msg)
1208             self._check_sample(msg_plain)
1209
1210     def test_x_to_invalid(self):
1211         # Convert all formats to an invalid format
1212         for class_ in (mailbox.Message, mailbox.MaildirMessage,
1213                        mailbox.mboxMessage, mailbox.MHMessage,
1214                        mailbox.BabylMessage, mailbox.MMDFMessage):
1215             self.assertRaises(TypeError, lambda: class_(False))
1216
1217     def test_maildir_to_maildir(self):
1218         # Convert MaildirMessage to MaildirMessage
1219         msg_maildir = mailbox.MaildirMessage(_sample_message)
1220         msg_maildir.set_flags('DFPRST')
1221         msg_maildir.set_subdir('cur')
1222         date = msg_maildir.get_date()
1223         msg = mailbox.MaildirMessage(msg_maildir)
1224         self._check_sample(msg)
1225         self.assert_(msg.get_flags() == 'DFPRST')
1226         self.assert_(msg.get_subdir() == 'cur')
1227         self.assert_(msg.get_date() == date)
1228
1229     def test_maildir_to_mboxmmdf(self):
1230         # Convert MaildirMessage to mboxmessage and MMDFMessage
1231         pairs = (('D', ''), ('F', 'F'), ('P', ''), ('R', 'A'), ('S', 'R'),
1232                  ('T', 'D'), ('DFPRST', 'RDFA'))
1233         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1234             msg_maildir = mailbox.MaildirMessage(_sample_message)
1235             msg_maildir.set_date(0.0)
1236             for setting, result in pairs:
1237                 msg_maildir.set_flags(setting)
1238                 msg = class_(msg_maildir)
1239                 self.assert_(msg.get_flags() == result)
1240                 self.assert_(msg.get_from() == 'MAILER-DAEMON %s' %
1241                              time.asctime(time.gmtime(0.0)))
1242             msg_maildir.set_subdir('cur')
1243             self.assert_(class_(msg_maildir).get_flags() == 'RODFA')
1244
1245     def test_maildir_to_mh(self):
1246         # Convert MaildirMessage to MHMessage
1247         msg_maildir = mailbox.MaildirMessage(_sample_message)
1248         pairs = (('D', ['unseen']), ('F', ['unseen', 'flagged']),
1249                  ('P', ['unseen']), ('R', ['unseen', 'replied']), ('S', []),
1250                  ('T', ['unseen']), ('DFPRST', ['replied', 'flagged']))
1251         for setting, result in pairs:
1252             msg_maildir.set_flags(setting)
1253             self.assert_(mailbox.MHMessage(msg_maildir).get_sequences() == \
1254                          result)
1255
1256     def test_maildir_to_babyl(self):
1257         # Convert MaildirMessage to Babyl
1258         msg_maildir = mailbox.MaildirMessage(_sample_message)
1259         pairs = (('D', ['unseen']), ('F', ['unseen']),
1260                  ('P', ['unseen', 'forwarded']), ('R', ['unseen', 'answered']),
1261                  ('S', []), ('T', ['unseen', 'deleted']),
1262                  ('DFPRST', ['deleted', 'answered', 'forwarded']))
1263         for setting, result in pairs:
1264             msg_maildir.set_flags(setting)
1265             self.assert_(mailbox.BabylMessage(msg_maildir).get_labels() == \
1266                          result)
1267
1268     def test_mboxmmdf_to_maildir(self):
1269         # Convert mboxMessage and MMDFMessage to MaildirMessage
1270         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1271             msg_mboxMMDF = class_(_sample_message)
1272             msg_mboxMMDF.set_from('foo@bar', time.gmtime(0.0))
1273             pairs = (('R', 'S'), ('O', ''), ('D', 'T'), ('F', 'F'), ('A', 'R'),
1274                      ('RODFA', 'FRST'))
1275             for setting, result in pairs:
1276                 msg_mboxMMDF.set_flags(setting)
1277                 msg = mailbox.MaildirMessage(msg_mboxMMDF)
1278                 self.assert_(msg.get_flags() == result)
1279                 self.assert_(msg.get_date() == 0.0, msg.get_date())
1280             msg_mboxMMDF.set_flags('O')
1281             self.assert_(mailbox.MaildirMessage(msg_mboxMMDF).get_subdir() == \
1282                          'cur')
1283
1284     def test_mboxmmdf_to_mboxmmdf(self):
1285         # Convert mboxMessage and MMDFMessage to mboxMessage and MMDFMessage
1286         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1287             msg_mboxMMDF = class_(_sample_message)
1288             msg_mboxMMDF.set_flags('RODFA')
1289             msg_mboxMMDF.set_from('foo@bar')
1290             for class2_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1291                 msg2 = class2_(msg_mboxMMDF)
1292                 self.assert_(msg2.get_flags() == 'RODFA')
1293                 self.assert_(msg2.get_from() == 'foo@bar')
1294
1295     def test_mboxmmdf_to_mh(self):
1296         # Convert mboxMessage and MMDFMessage to MHMessage
1297         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1298             msg_mboxMMDF = class_(_sample_message)
1299             pairs = (('R', []), ('O', ['unseen']), ('D', ['unseen']),
1300                      ('F', ['unseen', 'flagged']),
1301                      ('A', ['unseen', 'replied']),
1302                      ('RODFA', ['replied', 'flagged']))
1303             for setting, result in pairs:
1304                 msg_mboxMMDF.set_flags(setting)
1305                 self.assert_(mailbox.MHMessage(msg_mboxMMDF).get_sequences() \
1306                              == result)
1307
1308     def test_mboxmmdf_to_babyl(self):
1309         # Convert mboxMessage and MMDFMessage to BabylMessage
1310         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1311             msg = class_(_sample_message)
1312             pairs = (('R', []), ('O', ['unseen']),
1313                      ('D', ['unseen', 'deleted']), ('F', ['unseen']),
1314                      ('A', ['unseen', 'answered']),
1315                      ('RODFA', ['deleted', 'answered']))
1316             for setting, result in pairs:
1317                 msg.set_flags(setting)
1318                 self.assert_(mailbox.BabylMessage(msg).get_labels() == result)
1319
1320     def test_mh_to_maildir(self):
1321         # Convert MHMessage to MaildirMessage
1322         pairs = (('unseen', ''), ('replied', 'RS'), ('flagged', 'FS'))
1323         for setting, result in pairs:
1324             msg = mailbox.MHMessage(_sample_message)
1325             msg.add_sequence(setting)
1326             self.assert_(mailbox.MaildirMessage(msg).get_flags() == result)
1327             self.assert_(mailbox.MaildirMessage(msg).get_subdir() == 'cur')
1328         msg = mailbox.MHMessage(_sample_message)
1329         msg.add_sequence('unseen')
1330         msg.add_sequence('replied')
1331         msg.add_sequence('flagged')
1332         self.assert_(mailbox.MaildirMessage(msg).get_flags() == 'FR')
1333         self.assert_(mailbox.MaildirMessage(msg).get_subdir() == 'cur')
1334
1335     def test_mh_to_mboxmmdf(self):
1336         # Convert MHMessage to mboxMessage and MMDFMessage
1337         pairs = (('unseen', 'O'), ('replied', 'ROA'), ('flagged', 'ROF'))
1338         for setting, result in pairs:
1339             msg = mailbox.MHMessage(_sample_message)
1340             msg.add_sequence(setting)
1341             for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1342                 self.assert_(class_(msg).get_flags() == result)
1343         msg = mailbox.MHMessage(_sample_message)
1344         msg.add_sequence('unseen')
1345         msg.add_sequence('replied')
1346         msg.add_sequence('flagged')
1347         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1348             self.assert_(class_(msg).get_flags() == 'OFA')
1349
1350     def test_mh_to_mh(self):
1351         # Convert MHMessage to MHMessage
1352         msg = mailbox.MHMessage(_sample_message)
1353         msg.add_sequence('unseen')
1354         msg.add_sequence('replied')
1355         msg.add_sequence('flagged')
1356         self.assert_(mailbox.MHMessage(msg).get_sequences() == \
1357                      ['unseen', 'replied', 'flagged'])
1358
1359     def test_mh_to_babyl(self):
1360         # Convert MHMessage to BabylMessage
1361         pairs = (('unseen', ['unseen']), ('replied', ['answered']),
1362                  ('flagged', []))
1363         for setting, result in pairs:
1364             msg = mailbox.MHMessage(_sample_message)
1365             msg.add_sequence(setting)
1366             self.assert_(mailbox.BabylMessage(msg).get_labels() == result)
1367         msg = mailbox.MHMessage(_sample_message)
1368         msg.add_sequence('unseen')
1369         msg.add_sequence('replied')
1370         msg.add_sequence('flagged')
1371         self.assert_(mailbox.BabylMessage(msg).get_labels() == \
1372                      ['unseen', 'answered'])
1373
1374     def test_babyl_to_maildir(self):
1375         # Convert BabylMessage to MaildirMessage
1376         pairs = (('unseen', ''), ('deleted', 'ST'), ('filed', 'S'),
1377                  ('answered', 'RS'), ('forwarded', 'PS'), ('edited', 'S'),
1378                  ('resent', 'PS'))
1379         for setting, result in pairs:
1380             msg = mailbox.BabylMessage(_sample_message)
1381             msg.add_label(setting)
1382             self.assert_(mailbox.MaildirMessage(msg).get_flags() == result)
1383             self.assert_(mailbox.MaildirMessage(msg).get_subdir() == 'cur')
1384         msg = mailbox.BabylMessage(_sample_message)
1385         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1386                       'edited', 'resent'):
1387             msg.add_label(label)
1388         self.assert_(mailbox.MaildirMessage(msg).get_flags() == 'PRT')
1389         self.assert_(mailbox.MaildirMessage(msg).get_subdir() == 'cur')
1390
1391     def test_babyl_to_mboxmmdf(self):
1392         # Convert BabylMessage to mboxMessage and MMDFMessage
1393         pairs = (('unseen', 'O'), ('deleted', 'ROD'), ('filed', 'RO'),
1394                  ('answered', 'ROA'), ('forwarded', 'RO'), ('edited', 'RO'),
1395                  ('resent', 'RO'))
1396         for setting, result in pairs:
1397             for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1398                 msg = mailbox.BabylMessage(_sample_message)
1399                 msg.add_label(setting)
1400                 self.assert_(class_(msg).get_flags() == result)
1401         msg = mailbox.BabylMessage(_sample_message)
1402         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1403                       'edited', 'resent'):
1404             msg.add_label(label)
1405         for class_ in (mailbox.mboxMessage, mailbox.MMDFMessage):
1406             self.assert_(class_(msg).get_flags() == 'ODA')
1407
1408     def test_babyl_to_mh(self):
1409         # Convert BabylMessage to MHMessage
1410         pairs = (('unseen', ['unseen']), ('deleted', []), ('filed', []),
1411                  ('answered', ['replied']), ('forwarded', []), ('edited', []),
1412                  ('resent', []))
1413         for setting, result in pairs:
1414             msg = mailbox.BabylMessage(_sample_message)
1415             msg.add_label(setting)
1416             self.assert_(mailbox.MHMessage(msg).get_sequences() == result)
1417         msg = mailbox.BabylMessage(_sample_message)
1418         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1419                       'edited', 'resent'):
1420             msg.add_label(label)
1421         self.assert_(mailbox.MHMessage(msg).get_sequences() == \
1422                      ['unseen', 'replied'])
1423
1424     def test_babyl_to_babyl(self):
1425         # Convert BabylMessage to BabylMessage
1426         msg = mailbox.BabylMessage(_sample_message)
1427         msg.update_visible()
1428         for label in ('unseen', 'deleted', 'filed', 'answered', 'forwarded',
1429                       'edited', 'resent'):
1430             msg.add_label(label)
1431         msg2 = mailbox.BabylMessage(msg)
1432         self.assert_(msg2.get_labels() == ['unseen', 'deleted', 'filed',
1433                                            'answered', 'forwarded', 'edited',
1434                                            'resent'])
1435         self.assert_(msg.get_visible().keys() == msg2.get_visible().keys())
1436         for key in msg.get_visible().keys():
1437             self.assert_(msg.get_visible()[key] == msg2.get_visible()[key])
1438
1439
1440 class TestProxyFileBase(TestBase):
1441
1442     def _test_read(self, proxy):
1443         # Read by byte
1444         proxy.seek(0)
1445         self.assert_(proxy.read() == 'bar')
1446         proxy.seek(1)
1447         self.assert_(proxy.read() == 'ar')
1448         proxy.seek(0)
1449         self.assert_(proxy.read(2) == 'ba')
1450         proxy.seek(1)
1451         self.assert_(proxy.read(-1) == 'ar')
1452         proxy.seek(2)
1453         self.assert_(proxy.read(1000) == 'r')
1454
1455     def _test_readline(self, proxy):
1456         # Read by line
1457         proxy.seek(0)
1458         self.assert_(proxy.readline() == 'foo' + os.linesep)
1459         self.assert_(proxy.readline() == 'bar' + os.linesep)
1460         self.assert_(proxy.readline() == 'fred' + os.linesep)
1461         self.assert_(proxy.readline() == 'bob')
1462         proxy.seek(2)
1463         self.assert_(proxy.readline() == 'o' + os.linesep)
1464         proxy.seek(6 + 2 * len(os.linesep))
1465         self.assert_(proxy.readline() == 'fred' + os.linesep)
1466         proxy.seek(6 + 2 * len(os.linesep))
1467         self.assert_(proxy.readline(2) == 'fr')
1468         self.assert_(proxy.readline(-10) == 'ed' + os.linesep)
1469
1470     def _test_readlines(self, proxy):
1471         # Read multiple lines
1472         proxy.seek(0)
1473         self.assert_(proxy.readlines() == ['foo' + os.linesep,
1474                                            'bar' + os.linesep,
1475                                            'fred' + os.linesep, 'bob'])
1476         proxy.seek(0)
1477         self.assert_(proxy.readlines(2) == ['foo' + os.linesep])
1478         proxy.seek(3 + len(os.linesep))
1479         self.assert_(proxy.readlines(4 + len(os.linesep)) ==
1480                      ['bar' + os.linesep, 'fred' + os.linesep])
1481         proxy.seek(3)
1482         self.assert_(proxy.readlines(1000) == [os.linesep, 'bar' + os.linesep,
1483                                                'fred' + os.linesep, 'bob'])
1484
1485     def _test_iteration(self, proxy):
1486         # Iterate by line
1487         proxy.seek(0)
1488         iterator = iter(proxy)
1489         self.assert_(iterator.next() == 'foo' + os.linesep)
1490         self.assert_(iterator.next() == 'bar' + os.linesep)
1491         self.assert_(iterator.next() == 'fred' + os.linesep)
1492         self.assert_(iterator.next() == 'bob')
1493         self.assertRaises(StopIteration, lambda: iterator.next())
1494
1495     def _test_seek_and_tell(self, proxy):
1496         # Seek and use tell to check position
1497         proxy.seek(3)
1498         self.assert_(proxy.tell() == 3)
1499         self.assert_(proxy.read(len(os.linesep)) == os.linesep)
1500         proxy.seek(2, 1)
1501         self.assert_(proxy.read(1 + len(os.linesep)) == 'r' + os.linesep)
1502         proxy.seek(-3 - len(os.linesep), 2)
1503         self.assert_(proxy.read(3) == 'bar')
1504         proxy.seek(2, 0)
1505         self.assert_(proxy.read() == 'o' + os.linesep + 'bar' + os.linesep)
1506         proxy.seek(100)
1507         self.assert_(proxy.read() == '')
1508
1509     def _test_close(self, proxy):
1510         # Close a file
1511         proxy.close()
1512         self.assertRaises(AttributeError, lambda: proxy.close())
1513
1514
1515 class TestProxyFile(TestProxyFileBase):
1516
1517     def setUp(self):
1518         self._path = test_support.TESTFN
1519         self._file = open(self._path, 'wb+')
1520
1521     def tearDown(self):
1522         self._file.close()
1523         self._delete_recursively(self._path)
1524
1525     def test_initialize(self):
1526         # Initialize and check position
1527         self._file.write('foo')
1528         pos = self._file.tell()
1529         proxy0 = mailbox._ProxyFile(self._file)
1530         self.assert_(proxy0.tell() == pos)
1531         self.assert_(self._file.tell() == pos)
1532         proxy1 = mailbox._ProxyFile(self._file, 0)
1533         self.assert_(proxy1.tell() == 0)
1534         self.assert_(self._file.tell() == pos)
1535
1536     def test_read(self):
1537         self._file.write('bar')
1538         self._test_read(mailbox._ProxyFile(self._file))
1539
1540     def test_readline(self):
1541         self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep,
1542                                                   os.linesep))
1543         self._test_readline(mailbox._ProxyFile(self._file))
1544
1545     def test_readlines(self):
1546         self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep,
1547                                                   os.linesep))
1548         self._test_readlines(mailbox._ProxyFile(self._file))
1549
1550     def test_iteration(self):
1551         self._file.write('foo%sbar%sfred%sbob' % (os.linesep, os.linesep,
1552                                                   os.linesep))
1553         self._test_iteration(mailbox._ProxyFile(self._file))
1554
1555     def test_seek_and_tell(self):
1556         self._file.write('foo%sbar%s' % (os.linesep, os.linesep))
1557         self._test_seek_and_tell(mailbox._ProxyFile(self._file))
1558
1559     def test_close(self):
1560         self._file.write('foo%sbar%s' % (os.linesep, os.linesep))
1561         self._test_close(mailbox._ProxyFile(self._file))
1562
1563
1564 class TestPartialFile(TestProxyFileBase):
1565
1566     def setUp(self):
1567         self._path = test_support.TESTFN
1568         self._file = open(self._path, 'wb+')
1569
1570     def tearDown(self):
1571         self._file.close()
1572         self._delete_recursively(self._path)
1573
1574     def test_initialize(self):
1575         # Initialize and check position
1576         self._file.write('foo' + os.linesep + 'bar')
1577         pos = self._file.tell()
1578         proxy = mailbox._PartialFile(self._file, 2, 5)
1579         self.assert_(proxy.tell() == 0)
1580         self.assert_(self._file.tell() == pos)
1581
1582     def test_read(self):
1583         self._file.write('***bar***')
1584         self._test_read(mailbox._PartialFile(self._file, 3, 6))
1585
1586     def test_readline(self):
1587         self._file.write('!!!!!foo%sbar%sfred%sbob!!!!!' %
1588                          (os.linesep, os.linesep, os.linesep))
1589         self._test_readline(mailbox._PartialFile(self._file, 5,
1590                                                  18 + 3 * len(os.linesep)))
1591
1592     def test_readlines(self):
1593         self._file.write('foo%sbar%sfred%sbob?????' %
1594                          (os.linesep, os.linesep, os.linesep))
1595         self._test_readlines(mailbox._PartialFile(self._file, 0,
1596                                                   13 + 3 * len(os.linesep)))
1597
1598     def test_iteration(self):
1599         self._file.write('____foo%sbar%sfred%sbob####' %
1600                          (os.linesep, os.linesep, os.linesep))
1601         self._test_iteration(mailbox._PartialFile(self._file, 4,
1602                                                   17 + 3 * len(os.linesep)))
1603
1604     def test_seek_and_tell(self):
1605         self._file.write('(((foo%sbar%s$$$' % (os.linesep, os.linesep))
1606         self._test_seek_and_tell(mailbox._PartialFile(self._file, 3,
1607                                                       9 + 2 * len(os.linesep)))
1608
1609     def test_close(self):
1610         self._file.write('&foo%sbar%s^' % (os.linesep, os.linesep))
1611         self._test_close(mailbox._PartialFile(self._file, 1,
1612                                               6 + 3 * len(os.linesep)))
1613
1614
1615 ## Start: tests from the original module (for backward compatibility).
1616
1617 FROM_ = "From some.body@dummy.domain  Sat Jul 24 13:43:35 2004\n"
1618 DUMMY_MESSAGE = """\
1619 From: some.body@dummy.domain
1620 To: me@my.domain
1621 Subject: Simple Test
1622
1623 This is a dummy message.
1624 """
1625
1626 class MaildirTestCase(unittest.TestCase):
1627
1628     def setUp(self):
1629         # create a new maildir mailbox to work with:
1630         self._dir = test_support.TESTFN
1631         os.mkdir(self._dir)
1632         os.mkdir(os.path.join(self._dir, "cur"))
1633         os.mkdir(os.path.join(self._dir, "tmp"))
1634         os.mkdir(os.path.join(self._dir, "new"))
1635         self._counter = 1
1636         self._msgfiles = []
1637
1638     def tearDown(self):
1639         map(os.unlink, self._msgfiles)
1640         os.rmdir(os.path.join(self._dir, "cur"))
1641         os.rmdir(os.path.join(self._dir, "tmp"))
1642         os.rmdir(os.path.join(self._dir, "new"))
1643         os.rmdir(self._dir)
1644
1645     def createMessage(self, dir, mbox=False):
1646         t = int(time.time() % 1000000)
1647         pid = self._counter
1648         self._counter += 1
1649         filename = os.extsep.join((str(t), str(pid), "myhostname", "mydomain"))
1650         tmpname = os.path.join(self._dir, "tmp", filename)
1651         newname = os.path.join(self._dir, dir, filename)
1652         fp = open(tmpname, "w")
1653         self._msgfiles.append(tmpname)
1654         if mbox:
1655             fp.write(FROM_)
1656         fp.write(DUMMY_MESSAGE)
1657         fp.close()
1658         if hasattr(os, "link"):
1659             os.link(tmpname, newname)
1660         else:
1661             fp = open(newname, "w")
1662             fp.write(DUMMY_MESSAGE)
1663             fp.close()
1664         self._msgfiles.append(newname)
1665         return tmpname
1666
1667     def test_empty_maildir(self):
1668         """Test an empty maildir mailbox"""
1669         # Test for regression on bug #117490:
1670         # Make sure the boxes attribute actually gets set.
1671         self.mbox = mailbox.Maildir(test_support.TESTFN)
1672         #self.assert_(hasattr(self.mbox, "boxes"))
1673         #self.assert_(len(self.mbox.boxes) == 0)
1674         self.assert_(self.mbox.next() is None)
1675         self.assert_(self.mbox.next() is None)
1676
1677     def test_nonempty_maildir_cur(self):
1678         self.createMessage("cur")
1679         self.mbox = mailbox.Maildir(test_support.TESTFN)
1680         #self.assert_(len(self.mbox.boxes) == 1)
1681         self.assert_(self.mbox.next() is not None)
1682         self.assert_(self.mbox.next() is None)
1683         self.assert_(self.mbox.next() is None)
1684
1685     def test_nonempty_maildir_new(self):
1686         self.createMessage("new")
1687         self.mbox = mailbox.Maildir(test_support.TESTFN)
1688         #self.assert_(len(self.mbox.boxes) == 1)
1689         self.assert_(self.mbox.next() is not None)
1690         self.assert_(self.mbox.next() is None)
1691         self.assert_(self.mbox.next() is None)
1692
1693     def test_nonempty_maildir_both(self):
1694         self.createMessage("cur")
1695         self.createMessage("new")
1696         self.mbox = mailbox.Maildir(test_support.TESTFN)
1697         #self.assert_(len(self.mbox.boxes) == 2)
1698         self.assert_(self.mbox.next() is not None)
1699         self.assert_(self.mbox.next() is not None)
1700         self.assert_(self.mbox.next() is None)
1701         self.assert_(self.mbox.next() is None)
1702
1703     def test_unix_mbox(self):
1704         ### should be better!
1705         import email.Parser
1706         fname = self.createMessage("cur", True)
1707         n = 0
1708         for msg in mailbox.PortableUnixMailbox(open(fname),
1709                                                email.Parser.Parser().parse):
1710             n += 1
1711             self.assertEqual(msg["subject"], "Simple Test")
1712             self.assertEqual(len(str(msg)), len(FROM_)+len(DUMMY_MESSAGE))
1713         self.assertEqual(n, 1)
1714
1715 ## End: classes from the original module (for backward compatibility).
1716
1717
1718 _sample_message = """\
1719 Return-Path: <gkj@gregorykjohnson.com>
1720 X-Original-To: gkj+person@localhost
1721 Delivered-To: gkj+person@localhost
1722 Received: from localhost (localhost [127.0.0.1])
1723         by andy.gregorykjohnson.com (Postfix) with ESMTP id 356ED9DD17
1724         for <gkj+person@localhost>; Wed, 13 Jul 2005 17:23:16 -0400 (EDT)
1725 Delivered-To: gkj@sundance.gregorykjohnson.com
1726 Received: from localhost [127.0.0.1]
1727         by localhost with POP3 (fetchmail-6.2.5)
1728         for gkj+person@localhost (single-drop); Wed, 13 Jul 2005 17:23:16 -0400 (EDT)
1729 Received: from andy.gregorykjohnson.com (andy.gregorykjohnson.com [64.32.235.228])
1730         by sundance.gregorykjohnson.com (Postfix) with ESMTP id 5B056316746
1731         for <gkj@gregorykjohnson.com>; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)
1732 Received: by andy.gregorykjohnson.com (Postfix, from userid 1000)
1733         id 490CD9DD17; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)
1734 Date: Wed, 13 Jul 2005 17:23:11 -0400
1735 From: "Gregory K. Johnson" <gkj@gregorykjohnson.com>
1736 To: gkj@gregorykjohnson.com
1737 Subject: Sample message
1738 Message-ID: <20050713212311.GC4701@andy.gregorykjohnson.com>
1739 Mime-Version: 1.0
1740 Content-Type: multipart/mixed; boundary="NMuMz9nt05w80d4+"
1741 Content-Disposition: inline
1742 User-Agent: Mutt/1.5.9i
1743
1744
1745 --NMuMz9nt05w80d4+
1746 Content-Type: text/plain; charset=us-ascii
1747 Content-Disposition: inline
1748
1749 This is a sample message.
1750
1751 --
1752 Gregory K. Johnson
1753
1754 --NMuMz9nt05w80d4+
1755 Content-Type: application/octet-stream
1756 Content-Disposition: attachment; filename="text.gz"
1757 Content-Transfer-Encoding: base64
1758
1759 H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs
1760 3FYlAAAA
1761
1762 --NMuMz9nt05w80d4+--
1763 """
1764
1765 _sample_headers = {
1766     "Return-Path":"<gkj@gregorykjohnson.com>",
1767     "X-Original-To":"gkj+person@localhost",
1768     "Delivered-To":"gkj+person@localhost",
1769     "Received":"""from localhost (localhost [127.0.0.1])
1770         by andy.gregorykjohnson.com (Postfix) with ESMTP id 356ED9DD17
1771         for <gkj+person@localhost>; Wed, 13 Jul 2005 17:23:16 -0400 (EDT)""",
1772     "Delivered-To":"gkj@sundance.gregorykjohnson.com",
1773     "Received":"""from localhost [127.0.0.1]
1774         by localhost with POP3 (fetchmail-6.2.5)
1775         for gkj+person@localhost (single-drop); Wed, 13 Jul 2005 17:23:16 -0400 (EDT)""",
1776     "Received":"""from andy.gregorykjohnson.com (andy.gregorykjohnson.com [64.32.235.228])
1777         by sundance.gregorykjohnson.com (Postfix) with ESMTP id 5B056316746
1778         for <gkj@gregorykjohnson.com>; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)""",
1779     "Received":"""by andy.gregorykjohnson.com (Postfix, from userid 1000)
1780         id 490CD9DD17; Wed, 13 Jul 2005 17:23:11 -0400 (EDT)""",
1781     "Date":"Wed, 13 Jul 2005 17:23:11 -0400",
1782     "From":""""Gregory K. Johnson" <gkj@gregorykjohnson.com>""",
1783     "To":"gkj@gregorykjohnson.com",
1784     "Subject":"Sample message",
1785     "Mime-Version":"1.0",
1786     "Content-Type":"""multipart/mixed; boundary="NMuMz9nt05w80d4+\"""",
1787     "Content-Disposition":"inline",
1788     "User-Agent": "Mutt/1.5.9i" }
1789
1790 _sample_payloads = ("""This is a sample message.
1791
1792 --
1793 Gregory K. Johnson
1794 """,
1795 """H4sICM2D1UIAA3RleHQAC8nILFYAokSFktSKEoW0zJxUPa7wzJIMhZLyfIWczLzUYj0uAHTs
1796 3FYlAAAA
1797 """)
1798
1799
1800 def test_main():
1801     tests = (TestMailboxSuperclass, TestMaildir, TestMbox, TestMMDF, TestMH,
1802              TestBabyl, TestMessage, TestMaildirMessage, TestMboxMessage,
1803              TestMHMessage, TestBabylMessage, TestMMDFMessage,
1804              TestMessageConversion, TestProxyFile, TestPartialFile,
1805              MaildirTestCase)
1806     test_support.run_unittest(*tests)
1807     test_support.reap_children()
1808
1809
1810 if __name__ == '__main__':
1811     test_main()