Python code coverage for Lib/test/test_mailbox.py

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