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