diff options
author | Jeroen van Meeuwen (Kolab Systems) <vanmeeuwen@kolabsys.com> | 2011-09-26 11:58:56 +0100 |
---|---|---|
committer | Jeroen van Meeuwen (Kolab Systems) <vanmeeuwen@kolabsys.com> | 2011-09-26 11:58:56 +0100 |
commit | 6a1290ef060c4a7982b400b04f2771a688124c21 (patch) | |
tree | 369004a5bf4d82126e32621541a948b4aaa5e05c /pykolab | |
parent | 12e585451d2089fb688f299342a8a4e6e5c614f6 (diff) | |
download | pykolab-6a1290ef060c4a7982b400b04f2771a688124c21.tar.gz |
Update IMAP libs to be more consistent in terminology
Diffstat (limited to 'pykolab')
-rw-r--r-- | pykolab/imap/__init__.py | 56 | ||||
-rw-r--r-- | pykolab/imap/cyrus.py | 122 |
2 files changed, 96 insertions, 82 deletions
diff --git a/pykolab/imap/__init__.py b/pykolab/imap/__init__.py index 056b916..003db79 100644 --- a/pykolab/imap/__init__.py +++ b/pykolab/imap/__init__.py @@ -130,6 +130,7 @@ class IMAP(object): log.debug(_("Found old INBOX folder %s") %(old_inbox), level=8) if not self.has_folder(inbox): + log.info(_("Renaming INBOX from %s to %s") %(old_inbox,inbox)) self.imap.rename(old_inbox,inbox) self.inbox_folders.append(inbox) else: @@ -156,6 +157,12 @@ class IMAP(object): if type(user) == dict: if user.has_key(_match_attr): inbox_folders.append(user[_match_attr].lower()) + else: + # If the user passed on to this function does not have + # a key for _match_attr, then we have to bail out and + # continue + continue + elif type(user) == str: inbox_folders.append(user.lower()) @@ -243,6 +250,12 @@ class IMAP(object): ) def set_user_folder_quota(self, users=[], primary_domain=None, secondary_domain=[], folders=[]): + """ + + Sets the quota in IMAP using the authentication and authorization + database 'quota' attribute for the users listed in parameter 'users' + """ + self.connect() if conf.has_option(primary_domain, 'quota_attribute'): @@ -255,7 +268,7 @@ class IMAP(object): default_quota = auth.domain_default_quota(primary_domain) - if default_quota == "": + if default_quota == "" or default_quota == None: default_quota = 0 if len(users) == 0: @@ -266,7 +279,6 @@ class IMAP(object): if type(user) == dict: if user.has_key(_quota_attr): - #print "user has key" if type(user[_quota_attr]) == list: quota = user[_quota_attr].pop(0) elif type(user[_quota_attr]) == str: @@ -296,13 +308,14 @@ class IMAP(object): except: # TODO: Go in fact correct the quota. log.warning(_("Cannot get current IMAP quota for folder %s") %(folder)) - continue + used = 0 + current_quota = 0 new_quota = conf.plugins.exec_hook("set_user_folder_quota", kw={ 'used': used, 'current_quota': current_quota, - 'new_quota': int(quota), - 'default_quota': default_quota + 'new_quota': (int)(quota), + 'default_quota': (int)(default_quota) } ) @@ -317,7 +330,7 @@ class IMAP(object): auth.set_user_attribute(primary_domain, user, _quota_attr, new_quota) if not int(current_quota) == int(quota): - #log.info(_("Correcting quota for %s to %s (currently %s)") %(folder, quota, current_quota)) + log.info(_("Correcting quota for %s to %s (currently %s)") %(folder, quota, current_quota)) self.imap._setquota(folder, quota) def set_user_mailhost(self, users=[], primary_domain=None, secondary_domain=[], folders=[]): @@ -339,7 +352,6 @@ class IMAP(object): if type(user) == dict: if user.has_key(_mailserver_attr): - #print "user has key" if type(user[_mailserver_attr]) == list: _mailserver = user[_mailserver_attr].pop(0) elif type(user[_mailserver_attr]) == str: @@ -361,20 +373,18 @@ class IMAP(object): folder = folder.lower() - _current_mailserver = self.imap.find_mailbox_server(folder) + _current_mailserver = self.imap.find_mailfolder_server(folder) if not _mailserver == None: # TODO: if not _current_mailserver == _mailserver: self.imap._xfer(folder, _current_mailserver, _mailserver) - - auth.set_user_attribute(primary_domain, user, _mailserver_attr, _mailserver) else: auth.set_user_attribute(primary_domain, user, _mailserver_attr, _current_mailserver) - def parse_mailbox(self, mailbox): + def parse_mailfolder(self, mailfolder): self.connect() - return self.imap.parse_mailbox(mailbox) + return self.imap.parse_mailfolder(mailfolder) def expunge_user_folders(self, inbox_folders=None): """ @@ -408,27 +418,32 @@ class IMAP(object): continue else: log.info(_("Folder has no corresponding user (1): %s") %(folder)) - self.delete("user/%s" %(folder)) + self.delete_mailfolder("user/%s" %(folder)) except: log.info(_("Folder has no corresponding user (2): %s") %(folder)) try: - self.delete("user/%s" %(folder)) + self.delete_mailfolder("user/%s" %(folder)) except: pass - def delete(self, mailbox_path): - mbox_parts = self.parse_mailbox(mailbox_path) + def delete_mailfolder(self, mailfolder_path): + """ + Deletes a mail folder described by mailfolder_path. + """ + + mbox_parts = self.parse_mailfolder(mailfolder_path) if mbox_parts == None: # We got user identifier only log.error(_("Please don't give us just a user identifier")) return - #self.imap.dm(mailbox_path) + self.imap.dm(mailfolder_path) clean_acls = False section = False + if mbox_parts['domain']: if conf.has_option(mbox_parts['domain'], 'delete_clean_acls'): section = mbox_parts['domain'] @@ -511,6 +526,7 @@ class IMAP(object): for folder in _folders: folder_name = None if len(folder.split('@')) > 1: + # TODO: acceptable domain name spaces #acceptable = False #for domain_name_re in acceptable_domain_name_res: #prog = re.compile(domain_name_re) @@ -531,8 +547,6 @@ class IMAP(object): if not folder_name in folders: folders.append(folder_name) - #print folders - return folders def synchronize(self, users=[], primary_domain=None, secondary_domains=[]): @@ -550,5 +564,5 @@ class IMAP(object): def lm(self, *args, **kw): return self.imap.lm(*args, **kw) - def undelete(self, *args, **kw): - self.imap.undelete(*args, **kw) + def undelete_mailfolder(self, *args, **kw): + self.imap.undelete_mailfolder(*args, **kw) diff --git a/pykolab/imap/cyrus.py b/pykolab/imap/cyrus.py index ac864aa..c0f8c04 100644 --- a/pykolab/imap/cyrus.py +++ b/pykolab/imap/cyrus.py @@ -26,8 +26,8 @@ import pykolab from pykolab.translate import _ +log = pykolab.getLogger('pykolab.imap') conf = pykolab.getConf() -log = pykolab.getLogger('pykolab.imap.cyrus') imap = pykolab.imap @@ -104,103 +104,103 @@ class Cyrus(cyruslib.CYRUS): if not self.murder: log.debug(_("This system is not part of a murder topology"), level=8) - def find_mailbox_server(self, mailbox): + def find_mailfolder_server(self, mailfolder): annotations = {} - _mailbox = self.parse_mailbox(mailbox) - prefix = _mailbox['path_parts'].pop(0) - mbox = _mailbox['path_parts'].pop(0) - if not _mailbox['domain'] == None: - mailbox = "%s%s%s@%s" %(prefix,self.seperator,mbox,_mailbox['domain']) + _mailfolder = self.parse_mailfolder(mailfolder) + prefix = _mailfolder['path_parts'].pop(0) + mbox = _mailfolder['path_parts'].pop(0) + if not _mailfolder['domain'] == None: + mailfolder = "%s%s%s@%s" %(prefix,self.seperator,mbox,_mailfolder['domain']) # TODO: Workaround for undelete - if len(self.lm(mailbox)) < 1: + if len(self.lm(mailfolder)) < 1: return self.server if not self.murder: return self.server - log.debug(_("Checking actual backend server for folder %s through annotations") %(mailbox), level=8) - if self.mbox.has_key(mailbox): - return self.mbox[mailbox] + log.debug(_("Checking actual backend server for folder %s through annotations") %(mailfolder), level=8) + if self.mbox.has_key(mailfolder): + return self.mbox[mailfolder] max_tries = 20 num_try = 0 while 1: num_try += 1 - annotations = self._getannotation(mailbox, "/vendor/cmu/cyrus-imapd/server") + annotations = self._getannotation(mailfolder, "/vendor/cmu/cyrus-imapd/server") - if annotations.has_key(mailbox): + if annotations.has_key(mailfolder): break if max_tries <= num_try: log.error(_("Could not get the annotations after %s tries.") %(num_try)) - annotations = { mailbox: { '/vendor/cmu/cyrus-imapd/server': self.server }} + annotations = { mailfolder: { '/vendor/cmu/cyrus-imapd/server': self.server }} break - log.warning(_("No annotations for %s: %r") %(mailbox,annotations)) + log.warning(_("No annotations for %s: %r") %(mailfolder,annotations)) time.sleep(1) - server = annotations[mailbox]['/vendor/cmu/cyrus-imapd/server'] - self.mbox[mailbox] = server + server = annotations[mailfolder]['/vendor/cmu/cyrus-imapd/server'] + self.mbox[mailfolder] = server if not server == self.server: if imap._imap.has_key(server): - if not imap._imap[server].mbox.has_key(mailbox): - imap._imap[server].mbox[mailbox] = server + if not imap._imap[server].mbox.has_key(mailfolder): + imap._imap[server].mbox[mailfolder] = server - log.debug(_("Server for INBOX folder %s is %s") %(mailbox,server), level=8) + log.debug(_("Server for INBOX folder %s is %s") %(mailfolder,server), level=8) return server - def _setquota(self, mailbox, quota): + def _setquota(self, mailfolder, quota): """ Login to the actual backend server. """ - server = self.find_mailbox_server(mailbox) + server = self.find_mailfolder_server(mailfolder) imap.connect('imap://%s:143' %(server)) - log.debug(_("Setting quota for INBOX folder %s to %s") %(mailbox,quota), level=8) + log.debug(_("Setting quota for INBOX folder %s to %s") %(mailfolder,quota), level=8) try: - imap.setquota(mailbox, quota) + imap.setquota(mailfolder, quota) except: - log.error(_("Could not set quota for mailbox %s") %(mailbox)) + log.error(_("Could not set quota for mailfolder %s") %(mailfolder)) - def _rename(self, from_mailbox, to_mailbox, partition=None): + def _rename(self, from_mailfolder, to_mailfolder, partition=None): """ Login to the actual backend server, then rename. """ - server = self.find_mailbox_server(from_mailbox) + server = self.find_mailfolder_server(from_mailfolder) imap.connect('imap://%s:143' %(server)) - log.debug(_("Moving INBOX folder %s to %s") %(from_mailbox,to_mailbox), level=8) - imap.rename(from_mailbox, to_mailbox, partition) + log.debug(_("Moving INBOX folder %s to %s") %(from_mailfolder,to_mailfolder), level=8) + imap.rename(from_mailfolder, to_mailfolder, partition) def _getannotation(self, *args, **kw): imap.connect() return imap.getannotation(*args, **kw) - def _setannotation(self, mailbox, annotation, value): + def _setannotation(self, mailfolder, annotation, value): """ Login to the actual backend server, then set annotation. """ - server = self.find_mailbox_server(mailbox) + server = self.find_mailfolder_server(mailfolder) imap.connect('imap://%s:143' %(server)) - log.debug(_("Setting annotation %s on folder %s") %(annotation,mailbox), level=8) - imap.setannotation(mailbox, annotation, value) + log.debug(_("Setting annotation %s on folder %s") %(annotation,mailfolder), level=8) + imap.setannotation(mailfolder, annotation, value) - def _xfer(self, mailbox, current_server, new_server): + def _xfer(self, mailfolder, current_server, new_server): imap.connect('imap://%s:143' %(current_server)) - log.debug(_("Transferring folder %s from %s to %s") %(mailbox, current_server, new_server), level=8) - imap.xfer(mailbox, new_server) + log.debug(_("Transferring folder %s from %s to %s") %(mailfolder, current_server, new_server), level=8) + imap.xfer(mailfolder, new_server) - def undelete(self, mailbox, to_mailbox=None, recursive=True): + def undelete_mailfolder(self, mailfolder, to_mailfolder=None, recursive=True): """ - Login to the actual backend server, then "undelete" the mailbox. + Login to the actual backend server, then "undelete" the mailfolder. - 'mailbox' may be a string representing either of the following two + 'mailfolder' may be a string representing either of the following two options; - the fully qualified pathof the deleted folder in its current @@ -213,33 +213,33 @@ class Cyrus(cyruslib.CYRUS): "user/userid[@domain]" - 'to_mailbox' may be the target folder to "undelete" the deleted + 'to_mailfolder' may be the target folder to "undelete" the deleted folder to. If not specified, the original folder name is used. """ # Placeholder for folders we have recovered already. target_folders = [] - mailbox = self.parse_mailbox(mailbox) + mailfolder = self.parse_mailfolder(mailfolder) - undelete_folders = self._find_deleted_folder(mailbox) + undelete_folders = self._find_deleted_folder(mailfolder) - if not to_mailbox == None: - target_mbox = self.parse_mailbox(to_mailbox) + if not to_mailfolder == None: + target_mbox = self.parse_mailfolder(to_mailfolder) else: - target_mbox = mailbox + target_mbox = mailfolder for undelete_folder in undelete_folders: - undelete_mbox = self.parse_mailbox(undelete_folder) + undelete_mbox = self.parse_mailfolder(undelete_folder) prefix = undelete_mbox['path_parts'].pop(0) mbox = undelete_mbox['path_parts'].pop(0) - if to_mailbox == None: + if to_mailfolder == None: target_folder = self.seperator.join([prefix,mbox]) else: target_folder = self.seperator.join(target_mbox['path_parts']) - if not to_mailbox == None: + if not to_mailfolder == None: target_folder = "%s%s%s" %(target_folder,self.seperator,mbox) if not len(undelete_mbox['path_parts']) == 0: @@ -255,29 +255,29 @@ class Cyrus(cyruslib.CYRUS): log.info(_("Undeleting %s to %s") %(undelete_folder,target_folder)) - target_server = self.find_mailbox_server(target_folder) + target_server = self.find_mailfolder_server(target_folder) if not target_server == self.server: self.xfer(undelete_folder,target_server) self.rename(undelete_folder,target_folder) - def parse_mailbox(self, mailbox): + def parse_mailfolder(self, mailfolder): """ - Parse a mailbox name to it's parts. + Parse a mailfolder name to it's parts. - Takes a fully qualified mailbox or mailbox sub-folder. + Takes a fully qualified mailfolder or mailfolder sub-folder. """ mbox = { 'domain': None } # Split off the virtual domain identifier, if any - if len(mailbox.split('@')) > 1: - mbox['domain'] = mailbox.split('@')[1] - mbox['path_parts'] = mailbox.split('@')[0].split(self.seperator) + if len(mailfolder.split('@')) > 1: + mbox['domain'] = mailfolder.split('@')[1] + mbox['path_parts'] = mailfolder.split('@')[0].split(self.seperator) else: - mbox['path_parts'] = mailbox.split(self.seperator) + mbox['path_parts'] = mailfolder.split(self.seperator) # See if the path that has been specified is the current location for # the deleted folder, or the original location, we have to find the deleted @@ -298,10 +298,10 @@ class Cyrus(cyruslib.CYRUS): # Verify that the input for the deleted folder is actually a # deleted folder. - verify_folder_search = "%(dp)s%(sep)s%(mailbox)s" % { + verify_folder_search = "%(dp)s%(sep)s%(mailfolder)s" % { 'dp': deleted_prefix, 'sep': self.seperator, - 'mailbox': self.seperator.join(mbox['path_parts']) + 'mailfolder': self.seperator.join(mbox['path_parts']) } if not mbox['domain'] == None: @@ -330,15 +330,15 @@ class Cyrus(cyruslib.CYRUS): def _find_deleted_folder(self, mbox): """ - Give me the parts that are in an original mailbox name and I'll find + Give me the parts that are in an original mailfolder name and I'll find the deleted folder name. TODO: It finds virtdomain folders for non-virtdomain searches. """ - deleted_folder_search = "%(deleted_prefix)s%(seperator)s%(mailbox)s%(seperator)s*" % { + deleted_folder_search = "%(deleted_prefix)s%(seperator)s%(mailfolder)s%(seperator)s*" % { # TODO: The prefix used is configurable 'deleted_prefix': "DELETED", - 'mailbox': self.seperator.join(mbox['path_parts']), + 'mailfolder': self.seperator.join(mbox['path_parts']), 'seperator': self.seperator, } |