diff options
| -rw-r--r-- | Doc/library/mhlib.rst | 201 | ||||
| -rw-r--r-- | Doc/library/netdata.rst | 1 | ||||
| -rw-r--r-- | Lib/mhlib.py | 997 | ||||
| -rw-r--r-- | Lib/test/test_mhlib.py | 347 | ||||
| -rw-r--r-- | Lib/test/test_pyclbr.py | 1 | ||||
| -rw-r--r-- | Misc/NEWS | 2 |
6 files changed, 2 insertions, 1547 deletions
diff --git a/Doc/library/mhlib.rst b/Doc/library/mhlib.rst deleted file mode 100644 index f6edf28..0000000 --- a/Doc/library/mhlib.rst +++ /dev/null @@ -1,201 +0,0 @@ -:mod:`mhlib` --- Access to MH mailboxes -======================================= - -.. module:: mhlib - :synopsis: Manipulate MH mailboxes from Python. -.. sectionauthor:: Skip Montanaro <skip@pobox.com> - -The :mod:`mhlib` module provides a Python interface to MH folders and their -contents. - -The module contains three basic classes, :class:`MH`, which represents a -particular collection of folders, :class:`Folder`, which represents a single -folder, and :class:`Message`, which represents a single message. - - -.. class:: MH([path[, profile]]) - - :class:`MH` represents a collection of MH folders. - - -.. class:: Folder(mh, name) - - The :class:`Folder` class represents a single folder and its messages. - - -.. class:: Message(folder, number[, name]) - - :class:`Message` objects represent individual messages in a folder. The Message - class is derived from :class:`mimetools.Message`. - - -.. _mh-objects: - -MH Objects ----------- - -:class:`MH` instances have the following methods: - - -.. method:: MH.error(format[, ...]) - - Print an error message -- can be overridden. - - -.. method:: MH.getprofile(key) - - Return a profile entry (``None`` if not set). - - -.. method:: MH.getpath() - - Return the mailbox pathname. - - -.. method:: MH.getcontext() - - Return the current folder name. - - -.. method:: MH.setcontext(name) - - Set the current folder name. - - -.. method:: MH.listfolders() - - Return a list of top-level folders. - - -.. method:: MH.listallfolders() - - Return a list of all folders. - - -.. method:: MH.listsubfolders(name) - - Return a list of direct subfolders of the given folder. - - -.. method:: MH.listallsubfolders(name) - - Return a list of all subfolders of the given folder. - - -.. method:: MH.makefolder(name) - - Create a new folder. - - -.. method:: MH.deletefolder(name) - - Delete a folder -- must have no subfolders. - - -.. method:: MH.openfolder(name) - - Return a new open folder object. - - -.. _mh-folder-objects: - -Folder Objects --------------- - -:class:`Folder` instances represent open folders and have the following methods: - - -.. method:: Folder.error(format[, ...]) - - Print an error message -- can be overridden. - - -.. method:: Folder.getfullname() - - Return the folder's full pathname. - - -.. method:: Folder.getsequencesfilename() - - Return the full pathname of the folder's sequences file. - - -.. method:: Folder.getmessagefilename(n) - - Return the full pathname of message *n* of the folder. - - -.. method:: Folder.listmessages() - - Return a list of messages in the folder (as numbers). - - -.. method:: Folder.getcurrent() - - Return the current message number. - - -.. method:: Folder.setcurrent(n) - - Set the current message number to *n*. - - -.. method:: Folder.parsesequence(seq) - - Parse msgs syntax into list of messages. - - -.. method:: Folder.getlast() - - Get last message, or ``0`` if no messages are in the folder. - - -.. method:: Folder.setlast(n) - - Set last message (internal use only). - - -.. method:: Folder.getsequences() - - Return dictionary of sequences in folder. The sequence names are used as keys, - and the values are the lists of message numbers in the sequences. - - -.. method:: Folder.putsequences(dict) - - Return dictionary of sequences in folder name: list. - - -.. method:: Folder.removemessages(list) - - Remove messages in list from folder. - - -.. method:: Folder.refilemessages(list, tofolder) - - Move messages in list to other folder. - - -.. method:: Folder.movemessage(n, tofolder, ton) - - Move one message to a given destination in another folder. - - -.. method:: Folder.copymessage(n, tofolder, ton) - - Copy one message to a given destination in another folder. - - -.. _mh-message-objects: - -Message Objects ---------------- - -The :class:`Message` class adds one method to those of -:class:`mimetools.Message`: - - -.. method:: Message.openmessage(n) - - Return a new open message object (costs a file descriptor). - diff --git a/Doc/library/netdata.rst b/Doc/library/netdata.rst index 2f27189..b331e24 100644 --- a/Doc/library/netdata.rst +++ b/Doc/library/netdata.rst @@ -15,7 +15,6 @@ on the Internet. json.rst mailcap.rst mailbox.rst - mhlib.rst mimetools.rst mimetypes.rst multifile.rst diff --git a/Lib/mhlib.py b/Lib/mhlib.py deleted file mode 100644 index b428439..0000000 --- a/Lib/mhlib.py +++ /dev/null @@ -1,997 +0,0 @@ -"""MH interface -- purely object-oriented (well, almost) - -Executive summary: - -import mhlib - -mh = mhlib.MH() # use default mailbox directory and profile -mh = mhlib.MH(mailbox) # override mailbox location (default from profile) -mh = mhlib.MH(mailbox, profile) # override mailbox and profile - -mh.error(format, ...) # print error message -- can be overridden -s = mh.getprofile(key) # profile entry (None if not set) -path = mh.getpath() # mailbox pathname -name = mh.getcontext() # name of current folder -mh.setcontext(name) # set name of current folder - -list = mh.listfolders() # names of top-level folders -list = mh.listallfolders() # names of all folders, including subfolders -list = mh.listsubfolders(name) # direct subfolders of given folder -list = mh.listallsubfolders(name) # all subfolders of given folder - -mh.makefolder(name) # create new folder -mh.deletefolder(name) # delete folder -- must have no subfolders - -f = mh.openfolder(name) # new open folder object - -f.error(format, ...) # same as mh.error(format, ...) -path = f.getfullname() # folder's full pathname -path = f.getsequencesfilename() # full pathname of folder's sequences file -path = f.getmessagefilename(n) # full pathname of message n in folder - -list = f.listmessages() # list of messages in folder (as numbers) -n = f.getcurrent() # get current message -f.setcurrent(n) # set current message -list = f.parsesequence(seq) # parse msgs syntax into list of messages -n = f.getlast() # get last message (0 if no messagse) -f.setlast(n) # set last message (internal use only) - -dict = f.getsequences() # dictionary of sequences in folder {name: list} -f.putsequences(dict) # write sequences back to folder - -f.createmessage(n, fp) # add message from file f as number n -f.removemessages(list) # remove messages in list from folder -f.refilemessages(list, tofolder) # move messages in list to other folder -f.movemessage(n, tofolder, ton) # move one message to a given destination -f.copymessage(n, tofolder, ton) # copy one message to a given destination - -m = f.openmessage(n) # new open message object (costs a file descriptor) -m is a derived class of mimetools.Message(rfc822.Message), with: -s = m.getheadertext() # text of message's headers -s = m.getheadertext(pred) # text of message's headers, filtered by pred -s = m.getbodytext() # text of message's body, decoded -s = m.getbodytext(0) # text of message's body, not decoded -""" - -# XXX To do, functionality: -# - annotate messages -# - send messages -# -# XXX To do, organization: -# - move IntSet to separate file -# - move most Message functionality to module mimetools - - -# Customizable defaults - -MH_PROFILE = '~/.mh_profile' -PATH = '~/Mail' -MH_SEQUENCES = '.mh_sequences' -FOLDER_PROTECT = 0o700 - - -# Imported modules - -import os -import sys -import re -import mimetools -import multifile -import shutil -from bisect import bisect - -__all__ = ["MH","Error","Folder","Message"] - -# Exported constants - -class Error(Exception): - pass - - -class MH: - """Class representing a particular collection of folders. - Optional constructor arguments are the pathname for the directory - containing the collection, and the MH profile to use. - If either is omitted or empty a default is used; the default - directory is taken from the MH profile if it is specified there.""" - - def __init__(self, path = None, profile = None): - """Constructor.""" - if profile is None: profile = MH_PROFILE - self.profile = os.path.expanduser(profile) - if path is None: path = self.getprofile('Path') - if not path: path = PATH - if not os.path.isabs(path) and path[0] != '~': - path = os.path.join('~', path) - path = os.path.expanduser(path) - if not os.path.isdir(path): raise Error('MH() path not found') - self.path = path - - def __repr__(self): - """String representation.""" - return 'MH(%r, %r)' % (self.path, self.profile) - - def error(self, msg, *args): - """Routine to print an error. May be overridden by a derived class.""" - sys.stderr.write('MH error: %s\n' % (msg % args)) - - def getprofile(self, key): - """Return a profile entry, None if not found.""" - return pickline(self.profile, key) - - def getpath(self): - """Return the path (the name of the collection's directory).""" - return self.path - - def getcontext(self): - """Return the name of the current folder.""" - context = pickline(os.path.join(self.getpath(), 'context'), - 'Current-Folder') - if not context: context = 'inbox' - return context - - def setcontext(self, context): - """Set the name of the current folder.""" - fn = os.path.join(self.getpath(), 'context') - f = open(fn, "w") - f.write("Current-Folder: %s\n" % context) - f.close() - - def listfolders(self): - """Return the names of the top-level folders.""" - folders = [] - path = self.getpath() - for name in os.listdir(path): - fullname = os.path.join(path, name) - if os.path.isdir(fullname): - folders.append(name) - folders.sort() - return folders - - def listsubfolders(self, name): - """Return the names of the subfolders in a given folder - (prefixed with the given folder name).""" - fullname = os.path.join(self.path, name) - # Get the link count so we can avoid listing folders - # that have no subfolders. - nlinks = os.stat(fullname).st_nlink - if nlinks <= 2: - return [] - subfolders = [] - subnames = os.listdir(fullname) - for subname in subnames: - fullsubname = os.path.join(fullname, subname) - if os.path.isdir(fullsubname): - name_subname = os.path.join(name, subname) - subfolders.append(name_subname) - # Stop looking for subfolders when - # we've seen them all - nlinks = nlinks - 1 - if nlinks <= 2: - break - subfolders.sort() - return subfolders - - def listallfolders(self): - """Return the names of all folders and subfolders, recursively.""" - return self.listallsubfolders('') - - def listallsubfolders(self, name): - """Return the names of subfolders in a given folder, recursively.""" - fullname = os.path.join(self.path, name) - # Get the link count so we can avoid listing folders - # that have no subfolders. - nlinks = os.stat(fullname).st_nlink - if nlinks <= 2: - return [] - subfolders = [] - subnames = os.listdir(fullname) - for subname in subnames: - if subname[0] == ',' or isnumeric(subname): continue - fullsubname = os.path.join(fullname, subname) - if os.path.isdir(fullsubname): - name_subname = os.path.join(name, subname) - subfolders.append(name_subname) - if not os.path.islink(fullsubname): - subsubfolders = self.listallsubfolders( - name_subname) - subfolders = subfolders + subsubfolders - # Stop looking for subfolders when - # we've seen them all - nlinks = nlinks - 1 - if nlinks <= 2: - break - subfolders.sort() - return subfolders - - def openfolder(self, name): - """Return a new Folder object for the named folder.""" - return Folder(self, name) - - def makefolder(self, name): - """Create a new folder (or raise os.error if it cannot be created).""" - protect = pickline(self.profile, 'Folder-Protect') - if protect and isnumeric(protect): - mode = int(protect, 8) - else: - mode = FOLDER_PROTECT - os.mkdir(os.path.join(self.getpath(), name), mode) - - def deletefolder(self, name): - """Delete a folder. This removes files in the folder but not - subdirectories. Raise os.error if deleting the folder itself fails.""" - fullname = os.path.join(self.getpath(), name) - for subname in os.listdir(fullname): - fullsubname = os.path.join(fullname, subname) - try: - os.unlink(fullsubname) - except os.error: - self.error('%s not deleted, continuing...' % - fullsubname) - os.rmdir(fullname) - - -numericprog = re.compile('^[1-9][0-9]*$') -def isnumeric(str): - return numericprog.match(str) is not None - -class Folder: - """Class representing a particular folder.""" - - def __init__(self, mh, name): - """Constructor.""" - self.mh = mh - self.name = name - if not os.path.isdir(self.getfullname()): - raise Error('no folder %s' % name) - - def __repr__(self): - """String representation.""" - return 'Folder(%r, %r)' % (self.mh, self.name) - - def error(self, *args): - """Error message handler.""" - self.mh.error(*args) - - def getfullname(self): - """Return the full pathname of the folder.""" - return os.path.join(self.mh.path, self.name) - - def getsequencesfilename(self): - """Return the full pathname of the folder's sequences file.""" - return os.path.join(self.getfullname(), MH_SEQUENCES) - - def getmessagefilename(self, n): - """Return the full pathname of a message in the folder.""" - return os.path.join(self.getfullname(), str(n)) - - def listsubfolders(self): - """Return list of direct subfolders.""" - return self.mh.listsubfolders(self.name) - - def listallsubfolders(self): - """Return list of all subfolders.""" - return self.mh.listallsubfolders(self.name) - - def listmessages(self): - """Return the list of messages currently present in the folder. - As a side effect, set self.last to the last message (or 0).""" - messages = [] - match = numericprog.match - append = messages.append - for name in os.listdir(self.getfullname()): - if match(name): - append(name) - messages = sorted(map(int, messages)) - if messages: - self.last = messages[-1] - else: - self.last = 0 - return messages - - def getsequences(self): - """Return the set of sequences for the folder.""" - sequences = {} - fullname = self.getsequencesfilename() - try: - f = open(fullname, 'r') - except IOError: - return sequences - while 1: - line = f.readline() - if not line: break - fields = line.split(':') - if len(fields) != 2: - self.error('bad sequence in %s: %s' % - (fullname, line.strip())) - key = fields[0].strip() - value = IntSet(fields[1].strip(), ' ').tolist() - sequences[key] = value - return sequences - - def putsequences(self, sequences): - """Write the set of sequences back to the folder.""" - fullname = self.getsequencesfilename() - f = None - for key, seq in sequences.items(): - s = IntSet('', ' ') - s.fromlist(seq) - if not f: f = open(fullname, 'w') - f.write('%s: %s\n' % (key, s.tostring())) - if not f: - try: - os.unlink(fullname) - except os.error: - pass - else: - f.close() - - def getcurrent(self): - """Return the current message. Raise Error when there is none.""" - seqs = self.getsequences() - try: - return max(seqs['cur']) - except (ValueError, KeyError): - raise Error("no cur message") - - def setcurrent(self, n): - """Set the current message.""" - updateline(self.getsequencesfilename(), 'cur', str(n), 0) - - def parsesequence(self, seq): - """Parse an MH sequence specification into a message list. - Attempt to mimic mh-sequence(5) as close as possible. - Also attempt to mimic observed behavior regarding which - conditions cause which error messages.""" - # XXX Still not complete (see mh-format(5)). - # Missing are: - # - 'prev', 'next' as count - # - Sequence-Negation option - all = self.listmessages() - # Observed behavior: test for empty folder is done first - if not all: - raise Error("no messages in %s" % self.name) - # Common case first: all is frequently the default - if seq == 'all': - return all - # Test for X:Y before X-Y because 'seq:-n' matches both - i = seq.find(':') - if i >= 0: - head, dir, tail = seq[:i], '', seq[i+1:] - if tail[:1] in '-+': - dir, tail = tail[:1], tail[1:] - if not isnumeric(tail): - raise Error("bad message list %s" % seq) - try: - count = int(tail) - except (ValueError, OverflowError): - # Can't use sys.maxsize because of i+count below - count = len(all) - try: - anchor = self._parseindex(head, all) - except Error as msg: - seqs = self.getsequences() - if not head in seqs: - if not msg: - msg = "bad message list %s" % seq - raise Error(msg).with_traceback(sys.exc_info()[2]) - msgs = seqs[head] - if not msgs: - raise Error("sequence %s empty" % head) - if dir == '-': - return msgs[-count:] - else: - return msgs[:count] - else: - if not dir: - if head in ('prev', 'last'): - dir = '-' - if dir == '-': - i = bisect(all, anchor) - return all[max(0, i-count):i] - else: - i = bisect(all, anchor-1) - return all[i:i+count] - # Test for X-Y next - i = seq.find('-') - if i >= 0: - begin = self._parseindex(seq[:i], all) - end = self._parseindex(seq[i+1:], all) - i = bisect(all, begin-1) - j = bisect(all, end) - r = all[i:j] - if not r: - raise Error("bad message list %s" % seq) - return r - # Neither X:Y nor X-Y; must be a number or a (pseudo-)sequence - try: - n = self._parseindex(seq, all) - except Error as msg: - seqs = self.getsequences() - if not seq in seqs: - if not msg: - msg = "bad message list %s" % seq - raise Error(msg) - return seqs[seq] - else: - if n not in all: - if isnumeric(seq): - raise Error("message %d doesn't exist" % n) - else: - raise Error("no %s message" % seq) - else: - return [n] - - def _parseindex(self, seq, all): - """Internal: parse a message number (or cur, first, etc.).""" - if isnumeric(seq): - try: - return int(seq) - except (OverflowError, ValueError): - return sys.maxsize - if seq in ('cur', '.'): - return self.getcurrent() - if seq == 'first': - return all[0] - if seq == 'last': - return all[-1] - if seq == 'next': - n = self.getcurrent() - i = bisect(all, n) - try: - return all[i] - except IndexError: - raise Error("no next message") - if seq == 'prev': - n = self.getcurrent() - i = bisect(all, n-1) - if i == 0: - raise Error("no prev message") - try: - return all[i-1] - except IndexError: - raise Error("no prev message") - raise Error() - - def openmessage(self, n): - """Open a message -- returns a Message object.""" - return Message(self, n) - - def removemessages(self, list): - """Remove one or more messages -- may raise os.error.""" - errors = [] - deleted = [] - for n in list: - path = self.getmessagefilename(n) - commapath = self.getmessagefilename(',' + str(n)) - try: - os.unlink(commapath) - except os.error: - pass - try: - os.rename(path, commapath) - except os.error as msg: - errors.append(msg) - else: - deleted.append(n) - if deleted: - self.removefromallsequences(deleted) - if errors: - if len(errors) == 1: - raise os.error(errors[0]) - else: - raise os.error('multiple errors:', errors) - - def refilemessages(self, list, tofolder, keepsequences=0): - """Refile one or more messages -- may raise os.error. - 'tofolder' is an open folder object.""" - errors = [] - refiled = {} - for n in list: - ton = tofolder.getlast() + 1 - path = self.getmessagefilename(n) - topath = tofolder.getmessagefilename(ton) - try: - os.rename(path, topath) - except os.error: - # Try copying - try: - shutil.copy2(path, topath) - os.unlink(path) - except (IOError, os.error) as msg: - errors.append(msg) - try: - os.unlink(topath) - except os.error: - pass - continue - tofolder.setlast(ton) - refiled[n] = ton - if refiled: - if keepsequences: - tofolder._copysequences(self, refiled.items()) - self.removefromallsequences(refiled.keys()) - if errors: - if len(errors) == 1: - raise os.error(errors[0]) - else: - raise os.error('multiple errors:', errors) - - def _copysequences(self, fromfolder, refileditems): - """Helper for refilemessages() to copy sequences.""" - fromsequences = fromfolder.getsequences() - tosequences = self.getsequences() - changed = 0 - for name, seq in fromsequences.items(): - try: - toseq = tosequences[name] - new = 0 - except KeyError: - toseq = [] - new = 1 - for fromn, ton in refileditems: - if fromn in seq: - toseq.append(ton) - changed = 1 - if new and toseq: - tosequences[name] = toseq - if changed: - self.putsequences(tosequences) - - def movemessage(self, n, tofolder, ton): - """Move one message over a specific destination message, - which may or may not already exist.""" - path = self.getmessagefilename(n) - # Open it to check that it exists - f = open(path) - f.close() - del f - topath = tofolder.getmessagefilename(ton) - backuptopath = tofolder.getmessagefilename(',%d' % ton) - try: - os.rename(topath, backuptopath) - except os.error: - pass - try: - os.rename(path, topath) - except os.error: - # Try copying - ok = 0 - try: - tofolder.setlast(None) - shutil.copy2(path, topath) - ok = 1 - finally: - if not ok: - try: - os.unlink(topath) - except os.error: - pass - os.unlink(path) - self.removefromallsequences([n]) - - def copymessage(self, n, tofolder, ton): - """Copy one message over a specific destination message, - which may or may not already exist.""" - path = self.getmessagefilename(n) - # Open it to check that it exists - f = open(path) - f.close() - del f - topath = tofolder.getmessagefilename(ton) - backuptopath = tofolder.getmessagefilename(',%d' % ton) - try: - os.rename(topath, backuptopath) - except os.error: - pass - ok = 0 - try: - tofolder.setlast(None) - shutil.copy2(path, topath) - ok = 1 - finally: - if not ok: - try: - os.unlink(topath) - except os.error: - pass - - def createmessage(self, n, txt): - """Create a message, with text from the open file txt.""" - path = self.getmessagefilename(n) - backuppath = self.getmessagefilename(',%d' % n) - try: - os.rename(path, backuppath) - except os.error: - pass - ok = 0 - BUFSIZE = 16*1024 - try: - f = open(path, "w") - while 1: - buf = txt.read(BUFSIZE) - if not buf: - break - f.write(buf) - f.close() - ok = 1 - finally: - if not ok: - try: - os.unlink(path) - except os.error: - pass - - def removefromallsequences(self, list): - """Remove one or more messages from all sequences (including last) - -- but not from 'cur'!!!""" - if hasattr(self, 'last') and self.last in list: - del self.last - sequences = self.getsequences() - changed = 0 - for name, seq in sequences.items(): - if name == 'cur': - continue - for n in list: - if n in seq: - seq.remove(n) - changed = 1 - if not seq: - del sequences[name] - if changed: - self.putsequences(sequences) - - def getlast(self): - """Return the last message number.""" - if not hasattr(self, 'last'): - self.listmessages() # Set self.last - return self.last - - def setlast(self, last): - """Set the last message number.""" - if last is None: - if hasattr(self, 'last'): - del self.last - else: - self.last = last - -class Message(mimetools.Message): - - def __init__(self, f, n, fp = None): - """Constructor.""" - self.folder = f - self.number = n - if fp is None: - path = f.getmessagefilename(n) - fp = open(path, 'r') - mimetools.Message.__init__(self, fp) - - def __repr__(self): - """String representation.""" - return 'Message(%s, %s)' % (repr(self.folder), self.number) - - def getheadertext(self, pred = None): - """Return the message's header text as a string. If an - argument is specified, it is used as a filter predicate to - decide which headers to return (its argument is the header - name converted to lower case).""" - if pred is None: - return ''.join(self.headers) - headers = [] - hit = 0 - for line in self.headers: - if not line[0].isspace(): - i = line.find(':') - if i > 0: - hit = pred(line[:i].lower()) - if hit: headers.append(line) - return ''.join(headers) - - def getbodytext(self, decode = 1): - """Return the message's body text as string. This undoes a - Content-Transfer-Encoding, but does not interpret other MIME - features (e.g. multipart messages). To suppress decoding, - pass 0 as an argument.""" - self.fp.seek(self.startofbody) - encoding = self.getencoding() - if not decode or encoding in ('', '7bit', '8bit', 'binary'): - return self.fp.read() - from io import StringIO - output = StringIO() - mimetools.decode(self.fp, output, encoding) - return output.getvalue() - - def getbodyparts(self): - """Only for multipart messages: return the message's body as a - list of SubMessage objects. Each submessage object behaves - (almost) as a Message object.""" - if self.getmaintype() != 'multipart': - raise Error('Content-Type is not multipart/*') - bdry = self.getparam('boundary') - if not bdry: - raise Error('multipart/* without boundary param') - self.fp.seek(self.startofbody) - mf = multifile.MultiFile(self.fp) - mf.push(bdry) - parts = [] - while mf.next(): - n = "%s.%r" % (self.number, 1 + len(parts)) - part = SubMessage(self.folder, n, mf) - parts.append(part) - mf.pop() - return parts - - def getbody(self): - """Return body, either a string or a list of messages.""" - if self.getmaintype() == 'multipart': - return self.getbodyparts() - else: - return self.getbodytext() - - -class SubMessage(Message): - - def __init__(self, f, n, fp): - """Constructor.""" - Message.__init__(self, f, n, fp) - if self.getmaintype() == 'multipart': - self.body = Message.getbodyparts(self) - else: - self.body = Message.getbodytext(self) - self.bodyencoded = Message.getbodytext(self, decode=0) - # XXX If this is big, should remember file pointers - - def __repr__(self): - """String representation.""" - f, n, fp = self.folder, self.number, self.fp - return 'SubMessage(%s, %s, %s)' % (f, n, fp) - - def getbodytext(self, decode = 1): - if not decode: - return self.bodyencoded - if type(self.body) == type(''): - return self.body - - def getbodyparts(self): - if type(self.body) == type([]): - return self.body - - def getbody(self): - return self.body - - -class IntSet: - """Class implementing sets of integers. - - This is an efficient representation for sets consisting of several - continuous ranges, e.g. 1-100,200-400,402-1000 is represented - internally as a list of three pairs: [(1,100), (200,400), - (402,1000)]. The internal representation is always kept normalized. - - The constructor has up to three arguments: - - the string used to initialize the set (default ''), - - the separator between ranges (default ',') - - the separator between begin and end of a range (default '-') - The separators must be strings (not regexprs) and should be different. - - The tostring() function yields a string that can be passed to another - IntSet constructor; __repr__() is a valid IntSet constructor itself. - """ - - # XXX The default begin/end separator means that negative numbers are - # not supported very well. - # - # XXX There are currently no operations to remove set elements. - - def __init__(self, data = None, sep = ',', rng = '-'): - self.pairs = [] - self.sep = sep - self.rng = rng - if data: self.fromstring(data) - - def reset(self): - self.pairs = [] - - def __cmp__(self, other): - return cmp(self.pairs, other.pairs) - - def __hash__(self): - return hash(self.pairs) - - def __repr__(self): - return 'IntSet(%r, %r, %r)' % (self.tostring(), self.sep, self.rng) - - def normalize(self): - self.pairs.sort() - i = 1 - while i < len(self.pairs): - alo, ahi = self.pairs[i-1] - blo, bhi = self.pairs[i] - if ahi >= blo-1: - self.pairs[i-1:i+1] = [(alo, max(ahi, bhi))] - else: - i = i+1 - - def tostring(self): - s = '' - for lo, hi in self.pairs: - if lo == hi: t = repr(lo) - else: t = repr(lo) + self.rng + repr(hi) - if s: s = s + (self.sep + t) - else: s = t - return s - - def tolist(self): - l = [] - for lo, hi in self.pairs: - m = list(range(lo, hi+1)) - l = l + m - return l - - def fromlist(self, list): - for i in list: - self.append(i) - - def clone(self): - new = IntSet() - new.pairs = self.pairs[:] - return new - - def min(self): - return self.pairs[0][0] - - def max(self): - return self.pairs[-1][-1] - - def contains(self, x): - for lo, hi in self.pairs: - if lo <= x <= hi: return True - return False - - def append(self, x): - for i in range(len(self.pairs)): - lo, hi = self.pairs[i] - if x < lo: # Need to insert before - if x+1 == lo: - self.pairs[i] = (x, hi) - else: - self.pairs.insert(i, (x, x)) - if i > 0 and x-1 == self.pairs[i-1][1]: - # Merge with previous - self.pairs[i-1:i+1] = [ - (self.pairs[i-1][0], - self.pairs[i][1]) - ] - return - if x <= hi: # Already in set - return - i = len(self.pairs) - 1 - if i >= 0: - lo, hi = self.pairs[i] - if x-1 == hi: - self.pairs[i] = lo, x - return - self.pairs.append((x, x)) - - def addpair(self, xlo, xhi): - if xlo > xhi: return - self.pairs.append((xlo, xhi)) - self.normalize() - - def fromstring(self, data): - new = [] - for part in data.split(self.sep): - list = [] - for subp in part.split(self.rng): - s = subp.strip() - list.append(int(s)) - if len(list) == 1: - new.append((list[0], list[0])) - elif len(list) == 2 and list[0] <= list[1]: - new.append((list[0], list[1])) - else: - raise ValueError('bad data passed to IntSet') - self.pairs = self.pairs + new - self.normalize() - - -# Subroutines to read/write entries in .mh_profile and .mh_sequences - -def pickline(file, key, casefold = 1): - try: - f = open(file, 'r') - except IOError: - return None - pat = re.escape(key) + ':' - prog = re.compile(pat, casefold and re.IGNORECASE) - while 1: - line = f.readline() - if not line: break - if prog.match(line): - text = line[len(key)+1:] - while 1: - line = f.readline() - if not line or not line[0].isspace(): - break - text = text + line - return text.strip() - return None - -def updateline(file, key, value, casefold = 1): - try: - f = open(file, 'r') - lines = f.readlines() - f.close() - except IOError: - lines = [] - pat = re.escape(key) + ':(.*)\n' - prog = re.compile(pat, casefold and re.IGNORECASE) - if value is None: - newline = None - else: - newline = '%s: %s\n' % (key, value) - for i in range(len(lines)): - line = lines[i] - if prog.match(line): - if newline is None: - del lines[i] - else: - lines[i] = newline - break - else: - if newline is not None: - lines.append(newline) - tempfile = file + "~" - f = open(tempfile, 'w') - for line in lines: - f.write(line) - f.close() - os.rename(tempfile, file) - - -# Test program - -def test(): - global mh, f - os.system('rm -rf $HOME/Mail/@test') - mh = MH() - def do(s): print(s); print(eval(s)) - do('mh.listfolders()') - do('mh.listallfolders()') - testfolders = ['@test', '@test/test1', '@test/test2', - '@test/test1/test11', '@test/test1/test12', - '@test/test1/test11/test111'] - for t in testfolders: do('mh.makefolder(%r)' % (t,)) - do('mh.listsubfolders(\'@test\')') - do('mh.listallsubfolders(\'@test\')') - f = mh.openfolder('@test') - do('f.listsubfolders()') - do('f.listallsubfolders()') - do('f.getsequences()') - seqs = f.getsequences() - seqs['foo'] = IntSet('1-10 12-20', ' ').tolist() - print(seqs) - f.putsequences(seqs) - do('f.getsequences()') - for t in reversed(testfolders): do('mh.deletefolder(%r)' % (t,)) - do('mh.getcontext()') - context = mh.getcontext() - f = mh.openfolder(context) - do('f.getcurrent()') - for seq in ('first', 'last', 'cur', '.', 'prev', 'next', - 'first:3', 'last:3', 'cur:3', 'cur:-3', - 'prev:3', 'next:3', - '1:3', '1:-3', '100:3', '100:-3', '10000:3', '10000:-3', - 'all'): - try: - do('f.parsesequence(%r)' % (seq,)) - except Error as msg: - print("Error:", msg) - stuff = os.popen("pick %r 2>/dev/null" % (seq,)).read() - list = map(int, stuff.split()) - print(list, "<-- pick") - do('f.listmessages()') - - -if __name__ == '__main__': - test() diff --git a/Lib/test/test_mhlib.py b/Lib/test/test_mhlib.py deleted file mode 100644 index aea83ba..0000000 --- a/Lib/test/test_mhlib.py +++ /dev/null @@ -1,347 +0,0 @@ -""" - Tests for the mhlib module - Nick Mathewson -""" - -### BUG: This suite doesn't currently test the mime functionality of -### mhlib. It should. - -import unittest -from test.test_support import run_unittest, TESTFN, TestSkipped -import os -import io -import sys -import mhlib - -if sys.platform.startswith(("win", "atheos")): - # mhlib.updateline() renames a file to the name of a file that already - # exists. That causes a reasonable OS <wink> to complain in test_sequence - # here, like the "OSError: [Errno 17] File exists" raised on Windows. - # mhlib's listsubfolders() and listallfolders() do something with - # link counts, and that causes test_listfolders() here to get back - # an empty list from its call of listallfolders(). - # The other tests here pass on Windows. - raise TestSkipped("skipped on %s -- " % sys.platform + - "too many Unix assumptions") - -_mhroot = TESTFN+"_MH" -_mhpath = os.path.join(_mhroot, "MH") -_mhprofile = os.path.join(_mhroot, ".mh_profile") - -def normF(f): - return os.path.join(*f.split('/')) - -def writeFile(fname, contents): - dir = os.path.split(fname)[0] - if dir and not os.path.exists(dir): - mkdirs(dir) - f = open(fname, 'w') - f.write(contents) - f.close() - -def readFile(fname): - f = open(fname) - r = f.read() - f.close() - return r - -def writeProfile(dict): - contents = [ "%s: %s\n" % (k, v) for k, v in dict.items() ] - writeFile(_mhprofile, "".join(contents)) - -def writeContext(folder): - folder = normF(folder) - writeFile(os.path.join(_mhpath, "context"), - "Current-Folder: %s\n" % folder) - -def writeCurMessage(folder, cur): - folder = normF(folder) - writeFile(os.path.join(_mhpath, folder, ".mh_sequences"), - "cur: %s\n"%cur) - -def writeMessage(folder, n, headers, body): - folder = normF(folder) - headers = "".join([ "%s: %s\n" % (k, v) for k, v in headers.items() ]) - contents = "%s\n%s\n" % (headers,body) - mkdirs(os.path.join(_mhpath, folder)) - writeFile(os.path.join(_mhpath, folder, str(n)), contents) - -def getMH(): - return mhlib.MH(os.path.abspath(_mhpath), _mhprofile) - -def sortLines(s): - lines = s.split("\n") - lines = [ line.strip() for line in lines if len(line) >= 2 ] - lines.sort() - return lines - -# These next 2 functions are copied from test_glob.py. -def mkdirs(fname): - if os.path.exists(fname) or fname == '': - return - base, file = os.path.split(fname) - mkdirs(base) - os.mkdir(fname) - -def deltree(fname): - if not os.path.exists(fname): - return - for f in os.listdir(fname): - fullname = os.path.join(fname, f) - if os.path.isdir(fullname): - deltree(fullname) - else: - try: - os.unlink(fullname) - except: - pass - try: - os.rmdir(fname) - except: - pass - -class MhlibTests(unittest.TestCase): - def setUp(self): - deltree(_mhroot) - mkdirs(_mhpath) - writeProfile({'Path' : os.path.abspath(_mhpath), - 'Editor': 'emacs', - 'ignored-attribute': 'camping holiday'}) - # Note: These headers aren't really conformant to RFC822, but - # mhlib shouldn't care about that. - - # An inbox with a couple of messages. - writeMessage('inbox', 1, - {'From': 'Mrs. Premise', - 'To': 'Mrs. Conclusion', - 'Date': '18 July 2001'}, "Hullo, Mrs. Conclusion!\n") - writeMessage('inbox', 2, - {'From': 'Mrs. Conclusion', - 'To': 'Mrs. Premise', - 'Date': '29 July 2001'}, "Hullo, Mrs. Premise!\n") - - # A folder with many messages - for i in list(range(5, 101))+list(range(101, 201, 2)): - writeMessage('wide', i, - {'From': 'nowhere', 'Subject': 'message #%s' % i}, - "This is message number %s\n" % i) - - # A deeply nested folder - def deep(folder, n): - writeMessage(folder, n, - {'Subject': 'Message %s/%s' % (folder, n) }, - "This is message number %s in %s\n" % (n, folder) ) - deep('deep/f1', 1) - deep('deep/f1', 2) - deep('deep/f1', 3) - deep('deep/f2', 4) - deep('deep/f2', 6) - deep('deep', 3) - deep('deep/f2/f3', 1) - deep('deep/f2/f3', 2) - - def tearDown(self): - deltree(_mhroot) - - def test_basic(self): - writeContext('inbox') - writeCurMessage('inbox', 2) - mh = getMH() - - eq = self.assertEquals - eq(mh.getprofile('Editor'), 'emacs') - eq(mh.getprofile('not-set'), None) - eq(mh.getpath(), os.path.abspath(_mhpath)) - eq(mh.getcontext(), 'inbox') - - mh.setcontext('wide') - eq(mh.getcontext(), 'wide') - eq(readFile(os.path.join(_mhpath, 'context')), - "Current-Folder: wide\n") - - mh.setcontext('inbox') - - inbox = mh.openfolder('inbox') - eq(inbox.getfullname(), - os.path.join(os.path.abspath(_mhpath), 'inbox')) - eq(inbox.getsequencesfilename(), - os.path.join(os.path.abspath(_mhpath), 'inbox', '.mh_sequences')) - eq(inbox.getmessagefilename(1), - os.path.join(os.path.abspath(_mhpath), 'inbox', '1')) - - def test_listfolders(self): - mh = getMH() - eq = self.assertEquals - - folders = mh.listfolders() - folders.sort() - eq(folders, ['deep', 'inbox', 'wide']) - - folders = mh.listallfolders() - folders.sort() - tfolders = sorted(map(normF, ['deep', 'deep/f1', 'deep/f2', - 'deep/f2/f3', 'inbox', 'wide'])) - eq(folders, tfolders) - - folders = mh.listsubfolders('deep') - folders.sort() - eq(folders, list(map(normF, ['deep/f1', 'deep/f2']))) - - folders = mh.listallsubfolders('deep') - folders.sort() - eq(folders, list(map(normF, ['deep/f1', 'deep/f2', 'deep/f2/f3']))) - eq(mh.listsubfolders(normF('deep/f2')), [normF('deep/f2/f3')]) - - eq(mh.listsubfolders('inbox'), []) - eq(mh.listallsubfolders('inbox'), []) - - def test_sequence(self): - mh = getMH() - eq = self.assertEquals - writeCurMessage('wide', 55) - - f = mh.openfolder('wide') - all = f.listmessages() - eq(all, list(range(5, 101))+list(range(101, 201, 2))) - eq(f.getcurrent(), 55) - f.setcurrent(99) - eq(readFile(os.path.join(_mhpath, 'wide', '.mh_sequences')), - 'cur: 99\n') - - def seqeq(seq, val): - eq(f.parsesequence(seq), val) - - seqeq('5-55', list(range(5, 56))) - seqeq('90-108', list(range(90, 101))+list(range(101, 109, 2))) - seqeq('90-108', list(range(90, 101))+list(range(101, 109, 2))) - - seqeq('10:10', list(range(10, 20))) - seqeq('10:+10', list(range(10, 20))) - seqeq('101:10', list(range(101, 121, 2))) - - seqeq('cur', [99]) - seqeq('.', [99]) - seqeq('prev', [98]) - seqeq('next', [100]) - seqeq('cur:-3', [97, 98, 99]) - seqeq('first-cur', list(range(5, 100))) - seqeq('150-last', list(range(151, 201, 2))) - seqeq('prev-next', [98, 99, 100]) - - lowprimes = [5, 7, 11, 13, 17, 19, 23, 29] - lowcompos = [x for x in range(5, 31) if not x in lowprimes ] - f.putsequences({'cur': [5], - 'lowprime': lowprimes, - 'lowcompos': lowcompos}) - seqs = readFile(os.path.join(_mhpath, 'wide', '.mh_sequences')) - seqs = sortLines(seqs) - eq(seqs, ["cur: 5", - "lowcompos: 6 8-10 12 14-16 18 20-22 24-28 30", - "lowprime: 5 7 11 13 17 19 23 29"]) - - seqeq('lowprime', lowprimes) - seqeq('lowprime:1', [5]) - seqeq('lowprime:2', [5, 7]) - seqeq('lowprime:-2', [23, 29]) - - ## Not supported - #seqeq('lowprime:first', [5]) - #seqeq('lowprime:last', [29]) - #seqeq('lowprime:prev', [29]) - #seqeq('lowprime:next', [29]) - - def test_modify(self): - mh = getMH() - eq = self.assertEquals - - mh.makefolder("dummy1") - self.assert_("dummy1" in mh.listfolders()) - path = os.path.join(_mhpath, "dummy1") - self.assert_(os.path.exists(path)) - - f = mh.openfolder('dummy1') - def create(n): - msg = "From: foo\nSubject: %s\n\nDummy Message %s\n" % (n,n) - f.createmessage(n, io.StringIO(msg)) - - create(7) - create(8) - create(9) - - eq(readFile(f.getmessagefilename(9)), - "From: foo\nSubject: 9\n\nDummy Message 9\n") - - eq(f.listmessages(), [7, 8, 9]) - files = os.listdir(path) - files.sort() - eq(files, ['7', '8', '9']) - - f.removemessages(['7', '8']) - files = os.listdir(path) - files.sort() - eq(files, [',7', ',8', '9']) - eq(f.listmessages(), [9]) - create(10) - create(11) - create(12) - - mh.makefolder("dummy2") - f2 = mh.openfolder("dummy2") - eq(f2.listmessages(), []) - f.movemessage(10, f2, 3) - f.movemessage(11, f2, 5) - eq(f.listmessages(), [9, 12]) - eq(f2.listmessages(), [3, 5]) - eq(readFile(f2.getmessagefilename(3)), - "From: foo\nSubject: 10\n\nDummy Message 10\n") - - f.copymessage(9, f2, 4) - eq(f.listmessages(), [9, 12]) - eq(readFile(f2.getmessagefilename(4)), - "From: foo\nSubject: 9\n\nDummy Message 9\n") - - f.refilemessages([9, 12], f2) - eq(f.listmessages(), []) - eq(f2.listmessages(), [3, 4, 5, 6, 7]) - eq(readFile(f2.getmessagefilename(7)), - "From: foo\nSubject: 12\n\nDummy Message 12\n") - # XXX This should check that _copysequences does the right thing. - - mh.deletefolder('dummy1') - mh.deletefolder('dummy2') - self.assert_('dummy1' not in mh.listfolders()) - self.assert_(not os.path.exists(path)) - - def test_read(self): - mh = getMH() - eq = self.assertEquals - - f = mh.openfolder('inbox') - msg = f.openmessage(1) - # Check some basic stuff from rfc822 - eq(msg.getheader('From'), "Mrs. Premise") - eq(msg.getheader('To'), "Mrs. Conclusion") - - # Okay, we have the right message. Let's check the stuff from - # mhlib. - lines = sortLines(msg.getheadertext()) - eq(lines, ["Date: 18 July 2001", - "From: Mrs. Premise", - "To: Mrs. Conclusion"]) - lines = sortLines(msg.getheadertext(lambda h: len(h)==4)) - eq(lines, ["Date: 18 July 2001", - "From: Mrs. Premise"]) - eq(msg.getbodytext(), "Hullo, Mrs. Conclusion!\n\n") - eq(msg.getbodytext(0), "Hullo, Mrs. Conclusion!\n\n") - - # XXXX there should be a better way to reclaim the file handle - msg.fp.close() - del msg - - -def test_main(): - run_unittest(MhlibTests) - - -if __name__ == "__main__": - test_main() diff --git a/Lib/test/test_pyclbr.py b/Lib/test/test_pyclbr.py index 25c7d85..64c8f18 100644 --- a/Lib/test/test_pyclbr.py +++ b/Lib/test/test_pyclbr.py @@ -156,7 +156,6 @@ class PyclbrTest(TestCase): # These were once about the 10 longest modules cm('random', ignore=('Random',)) # from _random import Random as CoreGenerator cm('cgi', ignore=('log',)) # set with = in module - cm('mhlib') cm('urllib', ignore=('getproxies_registry', 'proxy_bypass_registry', 'open_https', @@ -21,6 +21,8 @@ Extension Modules Library ------- +- The mhlib module has been removed. + - The ihooks module has been removed. - The fpformat module has been removed. |