diff --git a/Doc/library/mailbox.rst b/Doc/library/mailbox.rst index f82a3b200deb7c4..64fc8d3148ab9a3 100644 --- a/Doc/library/mailbox.rst +++ b/Doc/library/mailbox.rst @@ -65,6 +65,10 @@ Supported mailbox formats are Maildir, mbox, MH, Babyl, and MMDF. :exc:`KeyError` exception if the corresponding message is subsequently removed. + .. versionchanged:: 3.10 + :class:`Mailbox` (and subclasses) supports a :term:`path-like object` for + the mailbox/maildir. + .. warning:: Be very cautious when modifying mailboxes that might be simultaneously diff --git a/Lib/mailbox.py b/Lib/mailbox.py index 70da07ed2e9e8bc..76e256db32c396d 100644 --- a/Lib/mailbox.py +++ b/Lib/mailbox.py @@ -17,6 +17,8 @@ import email.message import email.generator import io +import itertools +from pathlib import Path import contextlib from types import GenericAlias try: @@ -31,12 +33,13 @@ linesep = os.linesep.encode('ascii') + class Mailbox: """A group of messages in a particular place.""" def __init__(self, path, factory=None, create=True): """Initialize a Mailbox instance.""" - self._path = os.path.abspath(os.path.expanduser(path)) + self._path = Path(path).expanduser().resolve() self._factory = factory def add(self, message): @@ -273,15 +276,15 @@ def __init__(self, dirname, factory=None, create=True): """Initialize a Maildir instance.""" Mailbox.__init__(self, dirname, factory, create) self._paths = { - 'tmp': os.path.join(self._path, 'tmp'), - 'new': os.path.join(self._path, 'new'), - 'cur': os.path.join(self._path, 'cur'), + 'tmp': self._path / 'tmp', + 'new': self._path / 'new', + 'cur': self._path / 'cur', } - if not os.path.exists(self._path): + if not self._path.exists(): if create: - os.mkdir(self._path, 0o700) + self._path.mkdir(0o700) for path in self._paths.values(): - os.mkdir(path, 0o700) + path.mkdir(0o700) else: raise NoSuchMailboxError(self._path) self._toc = {} @@ -296,7 +299,7 @@ def add(self, message): self._dump_message(message, tmp_file) except BaseException: tmp_file.close() - os.remove(tmp_file.name) + Path(tmp_file.name).unlink() raise _sync_close(tmp_file) if isinstance(message, MaildirMessage): @@ -307,8 +310,8 @@ def add(self, message): else: subdir = 'new' suffix = '' - uniq = os.path.basename(tmp_file.name).split(self.colon)[0] - dest = os.path.join(self._path, subdir, uniq + suffix) + uniq = Path(tmp_file.name).name.split(self.colon)[0] + dest = self._path / subdir / (uniq + suffix) if isinstance(message, MaildirMessage): os.utime(tmp_file.name, (os.path.getatime(tmp_file.name), message.get_date())) @@ -317,13 +320,13 @@ def add(self, message): # from other programs try: try: - os.link(tmp_file.name, dest) + Path(tmp_file.name).link_to(dest) except (AttributeError, PermissionError): - os.rename(tmp_file.name, dest) + Path(tmp_file.name).rename(dest) else: - os.remove(tmp_file.name) + Path(tmp_file.name).unlink() except OSError as e: - os.remove(tmp_file.name) + Path(tmp_file.name).unlink() if e.errno == errno.EEXIST: raise ExternalClashError('Name clash with existing message: %s' % dest) @@ -333,7 +336,7 @@ def add(self, message): def remove(self, key): """Remove the keyed message; raise KeyError if it doesn't exist.""" - os.remove(os.path.join(self._path, self._lookup(key))) + (self._path / self._lookup(key)).unlink() def discard(self, key): """If the keyed message exists, remove it.""" @@ -354,26 +357,26 @@ def __setitem__(self, key, message): else: # temp's subdir and suffix were defaults from add(). dominant_subpath = old_subpath - subdir = os.path.dirname(dominant_subpath) + subdir = Path(dominant_subpath).parent if self.colon in dominant_subpath: suffix = self.colon + dominant_subpath.split(self.colon)[-1] else: suffix = '' self.discard(key) - tmp_path = os.path.join(self._path, temp_subpath) - new_path = os.path.join(self._path, subdir, key + suffix) + tmp_path = self._path / temp_subpath + new_path = self._path / subdir / (key + suffix) if isinstance(message, MaildirMessage): os.utime(tmp_path, (os.path.getatime(tmp_path), message.get_date())) # No file modification should be done after the file is moved to its # final position in order to prevent race conditions with changes # from other programs - os.rename(tmp_path, new_path) + tmp_path.rename(new_path) def get_message(self, key): """Return a Message representation or raise a KeyError.""" subpath = self._lookup(key) - with open(os.path.join(self._path, subpath), 'rb') as f: + with open(self._path / subpath, 'rb') as f: if self._factory: msg = self._factory(f) else: @@ -382,17 +385,17 @@ def get_message(self, key): msg.set_subdir(subdir) if self.colon in name: msg.set_info(name.split(self.colon)[-1]) - msg.set_date(os.path.getmtime(os.path.join(self._path, subpath))) + msg.set_date(os.path.getmtime(self._path / subpath)) return msg def get_bytes(self, key): """Return a bytes representation or raise a KeyError.""" - with open(os.path.join(self._path, self._lookup(key)), 'rb') as f: + with open(self._path / self._lookup(key), 'rb') as f: return f.read().replace(linesep, b'\n') def get_file(self, key): """Return a file-like representation or raise a KeyError.""" - f = open(os.path.join(self._path, self._lookup(key)), 'rb') + f = open(self._path / self._lookup(key), 'rb') return _ProxyFile(f) def iterkeys(self): @@ -436,54 +439,53 @@ def close(self): def list_folders(self): """Return a list of folder names.""" result = [] - for entry in os.listdir(self._path): - if len(entry) > 1 and entry[0] == '.' and \ - os.path.isdir(os.path.join(self._path, entry)): - result.append(entry[1:]) + for entry in self._path.iterdir(): + if len(entry.name) > 1 and entry.name[0] == '.' and entry.is_dir(): + result.append(entry.name[1:]) return result def get_folder(self, folder): """Return a Maildir instance for the named folder.""" - return Maildir(os.path.join(self._path, '.' + folder), + return Maildir(self._path / ('.' + folder), factory=self._factory, create=False) def add_folder(self, folder): """Create a folder and return a Maildir instance representing it.""" - path = os.path.join(self._path, '.' + folder) + path = self._path / ('.' + folder) result = Maildir(path, factory=self._factory) - maildirfolder_path = os.path.join(path, 'maildirfolder') - if not os.path.exists(maildirfolder_path): + maildirfolder_path = path / 'maildirfolder' + if not maildirfolder_path.exists(): os.close(os.open(maildirfolder_path, os.O_CREAT | os.O_WRONLY, 0o666)) return result def remove_folder(self, folder): """Delete the named folder, which must be empty.""" - path = os.path.join(self._path, '.' + folder) - for entry in os.listdir(os.path.join(path, 'new')) + \ - os.listdir(os.path.join(path, 'cur')): + path = self._path / ('.' + folder) + for entry in itertools.chain((path / 'new').iterdir(), + (path / 'cur').iterdir()): if len(entry) < 1 or entry[0] != '.': raise NotEmptyError('Folder contains message(s): %s' % folder) - for entry in os.listdir(path): - if entry != 'new' and entry != 'cur' and entry != 'tmp' and \ - os.path.isdir(os.path.join(path, entry)): + for entry in path.iterdir(): + if entry.name != 'new' and entry.name != 'cur' and \ + entry.name != 'tmp' and entry.is_dir(): raise NotEmptyError("Folder contains subdirectory '%s': %s" % (folder, entry)) for root, dirs, files in os.walk(path, topdown=False): for entry in files: - os.remove(os.path.join(root, entry)) + (Path(root) / entry).unlink() for entry in dirs: - os.rmdir(os.path.join(root, entry)) - os.rmdir(path) + (Path(root) / entry).rmdir() + path.rmdir() def clean(self): """Delete old files in "tmp".""" now = time.time() - for entry in os.listdir(os.path.join(self._path, 'tmp')): - path = os.path.join(self._path, 'tmp', entry) - if now - os.path.getatime(path) > 129600: # 60 * 60 * 36 - os.remove(path) + for entry in (self._path / 'tmp').iterdir(): + # 129600 = 60 * 60 * 36 + if now - os.path.getatime(entry.resolve()) > 129600: + entry.unlink() _count = 1 # This is used to generate unique file names. @@ -497,9 +499,9 @@ def _create_tmp(self): hostname = hostname.replace(':', r'\072') uniq = "%s.M%sP%sQ%s.%s" % (int(now), int(now % 1 * 1e6), os.getpid(), Maildir._count, hostname) - path = os.path.join(self._path, 'tmp', uniq) + path = self._path / 'tmp' / uniq try: - os.stat(path) + path.stat() except FileNotFoundError: Maildir._count += 1 try: @@ -539,18 +541,17 @@ def _refresh(self): self._toc = {} for subdir in self._toc_mtimes: path = self._paths[subdir] - for entry in os.listdir(path): - p = os.path.join(path, entry) - if os.path.isdir(p): + for entry in path.iterdir(): + if entry.is_dir(): continue - uniq = entry.split(self.colon)[0] - self._toc[uniq] = os.path.join(subdir, entry) + uniq = entry.name.split(self.colon)[0] + self._toc[uniq] = os.path.join(subdir, entry.name) self._last_read = time.time() def _lookup(self, key): """Use TOC to return subpath for given key, or raise a KeyError.""" try: - if os.path.exists(os.path.join(self._path, self._toc[key])): + if (self._path / self._toc[key]).exists(): return self._toc[key] except KeyError: pass @@ -693,19 +694,19 @@ def flush(self): self._file_length = new_file.tell() except: new_file.close() - os.remove(new_file.name) + Path(new_file.name).unlink() raise _sync_close(new_file) # self._file is about to get replaced, so no need to sync. self._file.close() # Make sure the new file's mode is the same as the old file's - mode = os.stat(self._path).st_mode - os.chmod(new_file.name, mode) + mode = self._path.stat().st_mode + Path(new_file.name).chmod(mode) try: - os.rename(new_file.name, self._path) + Path(new_file.name).rename(self._path) except FileExistsError: - os.remove(self._path) - os.rename(new_file.name, self._path) + self._path.unlink() + Path(new_file.name).rename(self._path) self._file = open(self._path, 'rb+') self._toc = new_toc self._pending = False @@ -938,10 +939,10 @@ class MH(Mailbox): def __init__(self, path, factory=None, create=True): """Initialize an MH instance.""" Mailbox.__init__(self, path, factory, create) - if not os.path.exists(self._path): + if not self._path.exists(): if create: - os.mkdir(self._path, 0o700) - os.close(os.open(os.path.join(self._path, '.mh_sequences'), + self._path.mkdir(0o700) + os.close(os.open(self._path / '.mh_sequences', os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)) else: raise NoSuchMailboxError(self._path) @@ -954,7 +955,7 @@ def add(self, message): new_key = 1 else: new_key = max(keys) + 1 - new_path = os.path.join(self._path, str(new_key)) + new_path = self._path / str(new_key) f = _create_carefully(new_path) closed = False try: @@ -969,7 +970,7 @@ def add(self, message): _unlock_file(f) _sync_close(f) closed = True - os.remove(new_path) + new_path.unlink() raise if isinstance(message, MHMessage): self._dump_sequences(message, new_key) @@ -983,7 +984,7 @@ def add(self, message): def remove(self, key): """Remove the keyed message; raise KeyError if it doesn't exist.""" - path = os.path.join(self._path, str(key)) + path = self._path / str(key) try: f = open(path, 'rb+') except OSError as e: @@ -993,11 +994,11 @@ def remove(self, key): raise else: f.close() - os.remove(path) + path.unlink() def __setitem__(self, key, message): """Replace the keyed message; raise KeyError if it doesn't exist.""" - path = os.path.join(self._path, str(key)) + path = self._path / str(key) try: f = open(path, 'rb+') except OSError as e: @@ -1023,9 +1024,9 @@ def get_message(self, key): """Return a Message representation or raise a KeyError.""" try: if self._locked: - f = open(os.path.join(self._path, str(key)), 'rb+') + f = open(self._path / str(key), 'rb+') else: - f = open(os.path.join(self._path, str(key)), 'rb') + f = open(self._path / str(key), 'rb') except OSError as e: if e.errno == errno.ENOENT: raise KeyError('No message with key: %s' % key) @@ -1048,9 +1049,9 @@ def get_bytes(self, key): """Return a bytes representation or raise a KeyError.""" try: if self._locked: - f = open(os.path.join(self._path, str(key)), 'rb+') + f = open(self._path / str(key), 'rb+') else: - f = open(os.path.join(self._path, str(key)), 'rb') + f = open(self._path / str(key), 'rb') except OSError as e: if e.errno == errno.ENOENT: raise KeyError('No message with key: %s' % key) @@ -1068,7 +1069,7 @@ def get_bytes(self, key): def get_file(self, key): """Return a file-like representation or raise a KeyError.""" try: - f = open(os.path.join(self._path, str(key)), 'rb') + f = open(self._path / str(key), 'rb') except OSError as e: if e.errno == errno.ENOENT: raise KeyError('No message with key: %s' % key) @@ -1078,12 +1079,12 @@ def get_file(self, key): def iterkeys(self): """Return an iterator over keys.""" - return iter(sorted(int(entry) for entry in os.listdir(self._path) - if entry.isdigit())) + return iter(sorted(int(entry.name) for entry in self._path.iterdir() + if entry.name.isdigit())) def __contains__(self, key): """Return True if the keyed message exists, False otherwise.""" - return os.path.exists(os.path.join(self._path, str(key))) + return (self._path / str(key)).exists() def __len__(self): """Return a count of messages in the mailbox.""" @@ -1092,7 +1093,7 @@ def __len__(self): def lock(self): """Lock the mailbox.""" if not self._locked: - self._file = open(os.path.join(self._path, '.mh_sequences'), 'rb+') + self._file = open(self._path / '.mh_sequences', 'rb+') _lock_file(self._file) self._locked = True @@ -1116,37 +1117,37 @@ def close(self): def list_folders(self): """Return a list of folder names.""" result = [] - for entry in os.listdir(self._path): - if os.path.isdir(os.path.join(self._path, entry)): - result.append(entry) + for entry in self._path.iterdir(): + if entry.is_dir(): + result.append(entry.name) return result def get_folder(self, folder): """Return an MH instance for the named folder.""" - return MH(os.path.join(self._path, folder), + return MH(self._path / folder, factory=self._factory, create=False) def add_folder(self, folder): """Create a folder and return an MH instance representing it.""" - return MH(os.path.join(self._path, folder), + return MH(self._path / folder, factory=self._factory) def remove_folder(self, folder): """Delete the named folder, which must be empty.""" - path = os.path.join(self._path, folder) - entries = os.listdir(path) - if entries == ['.mh_sequences']: - os.remove(os.path.join(path, '.mh_sequences')) + path = self._path / folder + entries = list(path.iterdir()) + if len(entries) == 1 and entries[0].name == '.mh_sequences': + entries[0].unlink() elif entries == []: pass else: raise NotEmptyError('Folder not empty: %s' % self._path) - os.rmdir(path) + path.rmdir() def get_sequences(self): """Return a name-to-key-list dictionary to define each sequence.""" results = {} - with open(os.path.join(self._path, '.mh_sequences'), 'r', encoding='ASCII') as f: + with open(self._path / '.mh_sequences', 'r', encoding='ASCII') as f: all_keys = set(self.keys()) for line in f: try: @@ -1169,7 +1170,7 @@ def get_sequences(self): def set_sequences(self, sequences): """Set sequences using the given name-to-key-list dictionary.""" - f = open(os.path.join(self._path, '.mh_sequences'), 'r+', encoding='ASCII') + f = open(self._path / '.mh_sequences', 'r+', encoding='ASCII') try: os.close(os.open(f.name, os.O_WRONLY | os.O_TRUNC)) for name, keys in sequences.items(): @@ -1205,13 +1206,11 @@ def pack(self): if key - 1 != prev: changes.append((key, prev + 1)) try: - os.link(os.path.join(self._path, str(key)), - os.path.join(self._path, str(prev + 1))) + (self._path / str(key)).link_to(self._path / str(prev + 1)) except (AttributeError, PermissionError): - os.rename(os.path.join(self._path, str(key)), - os.path.join(self._path, str(prev + 1))) + (self._path / str(key)).rename(self._path / str(prev + 1)) else: - os.unlink(os.path.join(self._path, str(key))) + (self._path / str(key)).unlink() prev += 1 self._next_key = prev + 1 if len(changes) == 0: @@ -2084,30 +2083,31 @@ def _lock_file(f, dotlock=True): raise try: try: - os.link(pre_lock.name, f.name + '.lock') + Path(pre_lock.name).link_to(f.name + '.lock') dotlock_done = True except (AttributeError, PermissionError): - os.rename(pre_lock.name, f.name + '.lock') + Path(pre_lock.name).rename(f.name + '.lock') dotlock_done = True else: - os.unlink(pre_lock.name) + Path(pre_lock.name).unlink() except FileExistsError: - os.remove(pre_lock.name) + Path(pre_lock.name).unlink() raise ExternalClashError('dot lock unavailable: %s' % f.name) except: if fcntl: fcntl.lockf(f, fcntl.LOCK_UN) if dotlock_done: - os.remove(f.name + '.lock') + Path(f.name + '.lock').unlink() raise def _unlock_file(f): """Unlock file f using lockf and dot locking.""" if fcntl: fcntl.lockf(f, fcntl.LOCK_UN) - if os.path.exists(f.name + '.lock'): - os.remove(f.name + '.lock') + if Path(f.name + '.lock').exists(): + Path(f.name + '.lock').unlink() + def _create_carefully(path): """Create a file if it doesn't exist and open for reading and writing.""" diff --git a/Lib/test/test_mailbox.py b/Lib/test/test_mailbox.py index 6f891d413cd8f15..4579b8b272469b1 100644 --- a/Lib/test/test_mailbox.py +++ b/Lib/test/test_mailbox.py @@ -13,6 +13,7 @@ import textwrap import mailbox import glob +from pathlib import Path class TestBase: @@ -37,9 +38,9 @@ def _check_sample(self, msg): def _delete_recursively(self, target): # Delete a file or delete a directory recursively - if os.path.isdir(target): + if Path(target).is_dir(): support.rmtree(target) - elif os.path.exists(target): + elif Path(target).exists(): support.unlink(target) @@ -51,7 +52,7 @@ class TestMailbox(TestBase): _template = 'From: foo\n\n%s\n' def setUp(self): - self._path = support.TESTFN + self._path = Path(support.TESTFN) self._delete_recursively(self._path) self._box = self._factory(self._path) @@ -59,6 +60,11 @@ def tearDown(self): self._box.close() self._delete_recursively(self._path) + def test_string_path(self): + """Test construction with a string instead of path-like.""" + tmp_box = self._factory(os.fspath(self._path)) + self.assertEqual(self._path.resolve(), tmp_box._path.resolve()) + def test_add(self): # Add copies of a sample message keys = [] @@ -506,11 +512,11 @@ def test_popitem_and_flush_twice(self): def test_lock_unlock(self): # Lock and unlock the mailbox - self.assertFalse(os.path.exists(self._get_lock_path())) + self.assertFalse(self._get_lock_path().exists()) self._box.lock() - self.assertTrue(os.path.exists(self._get_lock_path())) + self.assertTrue(self._get_lock_path().exists()) self._box.unlock() - self.assertFalse(os.path.exists(self._get_lock_path())) + self.assertFalse(self._get_lock_path().exists()) def test_close(self): # Close mailbox and flush changes to disk @@ -546,7 +552,7 @@ def test_dump_message(self): def _get_lock_path(self): # Return the path of the dot lock file. May be overridden. - return self._path + '.lock' + return Path(os.fspath(self._path) + '.lock') class TestMailboxSuperclass(TestBase, unittest.TestCase): @@ -595,7 +601,7 @@ def setUp(self): self._box.colon = '!' def assertMailboxEmpty(self): - self.assertEqual(os.listdir(os.path.join(self._path, 'tmp')), []) + self.assertEqual(list((self._path / 'tmp').iterdir()), []) def test_add_MM(self): # Add a MaildirMessage instance @@ -603,8 +609,8 @@ def test_add_MM(self): msg.set_subdir('cur') msg.set_info('foo') key = self._box.add(msg) - self.assertTrue(os.path.exists(os.path.join(self._path, 'cur', '%s%sfoo' % - (key, self._box.colon)))) + self.assertTrue((self._path / 'cur' / ('%s%sfoo' % + (key, self._box.colon))).exists()) def test_get_MM(self): # Get a MaildirMessage instance @@ -668,17 +674,17 @@ def test_initialize_existing(self): # Initialize an existing mailbox self.tearDown() for subdir in '', 'tmp', 'new', 'cur': - os.mkdir(os.path.normpath(os.path.join(self._path, subdir))) + Path(os.path.normpath(self._path / subdir)).mkdir() self._box = mailbox.Maildir(self._path) self._check_basics() def _check_basics(self, factory=None): # (Used by test_open_new() and test_open_existing().) - self.assertEqual(self._box._path, os.path.abspath(self._path)) + self.assertEqual(self._box._path.resolve(), self._path.resolve()) self.assertEqual(self._box._factory, factory) for subdir in '', 'tmp', 'new', 'cur': - path = os.path.join(self._path, subdir) - mode = os.stat(path)[stat.ST_MODE] + path = self._path / subdir + mode = path.stat()[stat.ST_MODE] self.assertTrue(stat.S_ISDIR(mode), "Not a directory: '%s'" % path) def test_list_folders(self): @@ -695,7 +701,7 @@ def test_get_folder(self): self._box.add_folder('foo.bar') folder0 = self._box.get_folder('foo.bar') folder0.add(self._template % 'bar') - self.assertTrue(os.path.isdir(os.path.join(self._path, '.foo.bar'))) + self.assertTrue((self._path / '.foo.bar').is_dir()) folder1 = self._box.get_folder('foo.bar') self.assertEqual(folder1.get_string(folder1.keys()[0]), self._template % 'bar') @@ -721,21 +727,21 @@ def test_add_and_remove_folders(self): def test_clean(self): # Remove old files from 'tmp' - foo_path = os.path.join(self._path, 'tmp', 'foo') - bar_path = os.path.join(self._path, 'tmp', 'bar') + foo_path = self._path / 'tmp' / 'foo' + bar_path = self._path / 'tmp' / 'bar' with open(foo_path, 'w') as f: f.write("@") with open(bar_path, 'w') as f: f.write("@") self._box.clean() - self.assertTrue(os.path.exists(foo_path)) - self.assertTrue(os.path.exists(bar_path)) - foo_stat = os.stat(foo_path) + self.assertTrue(foo_path.exists()) + self.assertTrue(bar_path.exists()) + foo_stat = foo_path.stat() os.utime(foo_path, (time.time() - 129600 - 2, foo_stat.st_mtime)) self._box.clean() - self.assertFalse(os.path.exists(foo_path)) - self.assertTrue(os.path.exists(bar_path)) + self.assertFalse(foo_path.exists()) + self.assertTrue(bar_path.exists()) def test_create_tmp(self, repetitions=10): # Create files in tmp directory @@ -751,8 +757,8 @@ def test_create_tmp(self, repetitions=10): for x in range(repetitions): tmp_file = self._box._create_tmp() head, tail = os.path.split(tmp_file.name) - self.assertEqual(head, os.path.abspath(os.path.join(self._path, - "tmp")), + self.assertEqual(Path(head).resolve(), + (self._path / "tmp").resolve(), "File in wrong location: '%s'" % head) match = pattern.match(tail) self.assertIsNotNone(match, "Invalid file name: '%s'" % tail) @@ -779,7 +785,7 @@ def test_create_tmp(self, repetitions=10): tmp_file.seek(0) self.assertEqual(tmp_file.read(), _bytes_sample_message) tmp_file.close() - file_count = len(os.listdir(os.path.join(self._path, "tmp"))) + file_count = len(list((self._path / "tmp").iterdir())) self.assertEqual(file_count, repetitions, "Wrong file count: '%s' should be '%s'" % (file_count, repetitions)) @@ -857,7 +863,7 @@ def test_directory_in_folder (self): self._box.add(mailbox.Message(_sample_message)) # Create a stray directory - os.mkdir(os.path.join(self._path, 'cur', 'stray-dir')) + (self._path / 'cur' / 'stray-dir').mkdir() # Check that looping still works with the directory present. for msg in self._box: @@ -872,8 +878,8 @@ def test_file_permissions(self): key = self._box.add(msg) finally: os.umask(orig_umask) - path = os.path.join(self._path, self._box._lookup(key)) - mode = os.stat(path).st_mode + path = self._path / self._box._lookup(key) + mode = path.stat().st_mode self.assertFalse(mode & 0o111) @unittest.skipUnless(hasattr(os, 'umask'), 'test needs os.umask()') @@ -886,8 +892,8 @@ def test_folder_file_perms(self): finally: os.umask(orig_umask) - path = os.path.join(subfolder._path, 'maildirfolder') - st = os.stat(path) + path = subfolder._path / 'maildirfolder' + st = path.stat() perms = st.st_mode self.assertFalse((perms & 0o111)) # Execute bits should all be off. @@ -898,8 +904,7 @@ def test_reread(self): # Put the last modified times more than two seconds into the past # (because mtime may have a two second granularity) for subdir in ('cur', 'new'): - os.utime(os.path.join(self._box._path, subdir), - (time.time()-5,)*2) + os.utime(self._box._path / subdir, (time.time()-5,)*2) # Because mtime has a two second granularity in worst case (FAT), a # refresh is done unconditionally if called for within @@ -925,9 +930,9 @@ def refreshed(): # Now, write something into cur and remove it. This changes # the mtime and should cause a re-read. Note that "sleep # emulation" is still in effect, as skewfactor is -3. - filename = os.path.join(self._path, 'cur', 'stray-file') + filename = self._path / 'cur' / 'stray-file' support.create_empty_file(filename) - os.unlink(filename) + filename.unlink() self._box._refresh() self.assertTrue(refreshed()) @@ -942,12 +947,12 @@ def test_add_doesnt_rewrite(self): # Inode number changes if the contents are written to another # file which is then renamed over the original file. So we # must check that the inode number doesn't change. - inode_before = os.stat(self._path).st_ino + inode_before = self._path.stat().st_ino self._box.add(self._template % 0) self._box.flush() - inode_after = os.stat(self._path).st_ino + inode_after = self._path.stat().st_ino self.assertEqual(inode_before, inode_after) # Make sure the message was really added @@ -961,8 +966,8 @@ def test_permissions_after_flush(self): # Make the mailbox world writable. It's unlikely that the new # mailbox file would have these permissions after flush(), # because umask usually prevents it. - mode = os.stat(self._path).st_mode | 0o666 - os.chmod(self._path, mode) + mode = self._path.stat().st_mode | 0o666 + self._path.chmod(mode) self._box.add(self._template % 0) i = self._box.add(self._template % 1) @@ -970,7 +975,7 @@ def test_permissions_after_flush(self): self._box.remove(i) self._box.flush() - self.assertEqual(os.stat(self._path).st_mode, mode) + self.assertEqual(self._path.stat().st_mode, mode) class _TestMboxMMDF(_TestSingleFile): @@ -1125,14 +1130,14 @@ def test_file_perms(self): try: old_umask = os.umask(0o077) self._box.close() - os.unlink(self._path) + self._path.unlink() self._box = mailbox.mbox(self._path, create=True) self._box.add('') self._box.close() finally: os.umask(old_umask) - st = os.stat(self._path) + st = self._path.stat() perms = st.st_mode self.assertFalse((perms & 0o111)) # Execute bits should all be off. @@ -1169,7 +1174,7 @@ class TestMH(TestMailbox, unittest.TestCase): _factory = lambda self, path, factory=None: mailbox.MH(path, factory) def assertMailboxEmpty(self): - self.assertEqual(os.listdir(self._path), ['.mh_sequences']) + self.assertEqual([entry.name for entry in self._path.iterdir()], ['.mh_sequences']) def test_list_folders(self): # List folders @@ -1189,7 +1194,7 @@ def dummy_factory (s): new_folder = self._box.add_folder('foo.bar') folder0 = self._box.get_folder('foo.bar') folder0.add(self._template % 'bar') - self.assertTrue(os.path.isdir(os.path.join(self._path, 'foo.bar'))) + self.assertTrue((self._path / 'foo.bar').is_dir()) folder1 = self._box.get_folder('foo.bar') self.assertEqual(folder1.get_string(folder1.keys()[0]), self._template % 'bar') @@ -1296,7 +1301,7 @@ def test_pack(self): 'unseen':[1], 'bar':[3], 'replied':[3]}) def _get_lock_path(self): - return os.path.join(self._path, '.mh_sequences.lock') + return (self._path / '.mh_sequences.lock') class TestBabyl(_TestSingleFile, unittest.TestCase): @@ -2131,23 +2136,24 @@ class MaildirTestCase(unittest.TestCase): def setUp(self): # create a new maildir mailbox to work with: - self._dir = support.TESTFN - if os.path.isdir(self._dir): + self._dir = Path(support.TESTFN) + if self._dir.is_dir(): support.rmtree(self._dir) - elif os.path.isfile(self._dir): + elif self._dir.is_file(): support.unlink(self._dir) - os.mkdir(self._dir) - os.mkdir(os.path.join(self._dir, "cur")) - os.mkdir(os.path.join(self._dir, "tmp")) - os.mkdir(os.path.join(self._dir, "new")) + self._dir.mkdir() + (self._dir / "cur").mkdir() + (self._dir / "tmp").mkdir() + (self._dir / "new").mkdir() self._counter = 1 self._msgfiles = [] def tearDown(self): - list(map(os.unlink, self._msgfiles)) - support.rmdir(os.path.join(self._dir, "cur")) - support.rmdir(os.path.join(self._dir, "tmp")) - support.rmdir(os.path.join(self._dir, "new")) + for msgfile in self._msgfiles: + msgfile.unlink() + support.rmdir(self._dir / "cur") + support.rmdir(self._dir / "tmp") + support.rmdir(self._dir / "new") support.rmdir(self._dir) def createMessage(self, dir, mbox=False): @@ -2155,15 +2161,15 @@ def createMessage(self, dir, mbox=False): pid = self._counter self._counter += 1 filename = ".".join((str(t), str(pid), "myhostname", "mydomain")) - tmpname = os.path.join(self._dir, "tmp", filename) - newname = os.path.join(self._dir, dir, filename) + tmpname = self._dir / "tmp" / filename + newname = self._dir / dir / filename with open(tmpname, "w") as fp: self._msgfiles.append(tmpname) if mbox: fp.write(FROM_) fp.write(DUMMY_MESSAGE) try: - os.link(tmpname, newname) + tmpname.link_to(newname) except (AttributeError, PermissionError): with open(newname, "w") as fp: fp.write(DUMMY_MESSAGE) @@ -2301,10 +2307,10 @@ def test__all__(self): def test_main(): tests = (TestMailboxSuperclass, TestMaildir, TestMbox, TestMMDF, TestMH, - TestBabyl, TestMessage, TestMaildirMessage, TestMboxMessage, - TestMHMessage, TestBabylMessage, TestMMDFMessage, - TestMessageConversion, TestProxyFile, TestPartialFile, - MaildirTestCase, TestFakeMailBox, MiscTestCase) + TestBabyl, TestMessage, TestMaildirMessage, TestMboxMessage, + TestMHMessage, TestBabylMessage, TestMMDFMessage, + TestMessageConversion, TestProxyFile, TestPartialFile, + MaildirTestCase, TestFakeMailBox, MiscTestCase) support.run_unittest(*tests) support.reap_children() diff --git a/Misc/ACKS b/Misc/ACKS index 641ef0cace00e2f..39b4e805c5c5d99 100644 --- a/Misc/ACKS +++ b/Misc/ACKS @@ -761,6 +761,7 @@ Greg Humphreys Chris Hunt Eric Huss Nehal Hussain +Laurence Alexander Hurst Taihyun Hwang Jeremy Hylton Ludwig Hähne diff --git a/Misc/NEWS.d/next/Library/2020-06-19-10-16-57.bpo-41026.hEXB-B.rst b/Misc/NEWS.d/next/Library/2020-06-19-10-16-57.bpo-41026.hEXB-B.rst new file mode 100644 index 000000000000000..41fd9650de3880a --- /dev/null +++ b/Misc/NEWS.d/next/Library/2020-06-19-10-16-57.bpo-41026.hEXB-B.rst @@ -0,0 +1,2 @@ +:mod:`mailbox` now uses pathlib internally and supports mailbox paths using a +:term:`path-like object`. Patch by Laurence Alexander Hurst.