Browse IMAP folders in service discovery

darcs-hash:20071022182523-86b55-444ef63ae52e5a0f274a1ab3bea89d9fc91ae1fb.gz
This commit is contained in:
David Rousselie
2007-10-22 20:25:23 +02:00
parent 95564d826e
commit 98510aa128
4 changed files with 108 additions and 21 deletions

View File

@@ -110,20 +110,17 @@ class IMAPAccountDiscoGetItemsHandler(DiscoHandler):
if account_class is not None and _account is not None:
disco_items = DiscoItems()
if imap_dir is None:
# TODO : test if INBOX really exist, is it possible to retrieve default dir from IMAP ?
DiscoItem(disco_items,
JID(unicode(_account.jid) + "/" + account_type
+ "/INBOX"),
account_type + "/" + account_name + "/INBOX",
"INBOX")
else:
for subdir in _account.ls_dir(imap_dir):
imap_dir = ""
subdirs = _account.ls_dir(imap_dir)
if imap_dir != "":
imap_dir += "/"
for subdir in subdirs:
DiscoItem(disco_items,
JID(unicode(_account.jid) + "/" + account_type + \
"/" + imap_dir + "/" + subdir),
"/" + imap_dir + subdir),
account_type + "/" + account_name + "/" + imap_dir
+ "/" + subdir,
+ subdir,
subdir)
return [disco_items]
return []
# TODO : implement get_info on imap_Dir

View File

@@ -145,7 +145,9 @@ class MockIMAPAccount(MockMailAccount, IMAPAccount):
MockMailAccount._init(self)
def ls_dir(self, imap_dir):
if imap_dir == "INBOX":
if imap_dir == "":
return ["INBOX"]
elif imap_dir == "INBOX":
return ["dir1", "dir2"]
elif imap_dir == "INBOX/dir1":
return ["subdir1", "subdir2"]

View File

@@ -21,6 +21,7 @@
## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
##
import re
import sys
import logging
import email
@@ -394,6 +395,9 @@ class IMAPAccount(MailAccount):
def _init(self, *args, **kw):
MailAccount._init(self, *args, **kw)
self.__logger = logging.getLogger("jmc.IMAPConnection")
self._regexp_list = re.compile("\((.*)\) \"(.)\" \"?([^\"]*)\"?$")
self.__cached_folders = {}
self.default_delimiter = "."
def get_type(self):
if self.ssl:
@@ -458,6 +462,48 @@ class IMAPAccount(MailAccount):
type = property(get_type)
def _add_full_path_to_cache(self, folder_array):
current_dir = self.__cached_folders
for folder in folder_array:
if not current_dir.has_key(folder):
current_dir[folder] = {}
current_dir = current_dir[folder]
def _build_folder_cache(self):
if self.connected:
typ, data = self.connection.list()
if typ == 'OK':
for line in data:
match = self._regexp_list.match(line)
if match is not None:
self.default_delimiter = match.group(2)
subdir = match.group(3).split(self.default_delimiter)
self._add_full_path_to_cache(subdir)
return self.__cached_folders
def ls_dir(self, imap_dir):
"""
imap_dir: IMAP directory to list. subdirs must be delimited by '/'
"""
# TODO : implement with cache (ie. only one connection)
self.__logger.debug("Listing IMAP dir '" + str(imap_dir) + "'")
if self.__cached_folders == {}:
if not self.connected:
self.connect()
self._build_folder_cache()
self.disconnect()
if imap_dir == "":
folder_array = []
else:
folder_array = imap_dir.split('/')
current_folder = self.__cached_folders
for folder in folder_array:
if not current_folder.has_key(folder):
return []
else:
current_folder = current_folder[folder]
return current_folder.keys()
class POP3Account(MailAccount):
nb_mail = IntCol(default=0)
lastmail = IntCol(default=0)

View File

@@ -314,6 +314,48 @@ class IMAPAccount_TestCase(InheritableAccount_TestCase):
(u"From : None\nSubject : None\n\nbody text\r\n\n", \
u"None")))
test_build_folder_cache = make_test(\
[lambda data: '* LIST () "." "INBOX"\r\n' + \
'* LIST () "." "INBOX.dir1"\r\n' + \
'* LIST () "." "INBOX.dir1.subdir1"\r\n' + \
'* LIST () "." "INBOX.dir1.subdir2"\r\n' + \
'* LIST () "." "INBOX.dir2"\r\n' + \
data.split()[0] + ' OK LIST completed\r\n'],
["^[^ ]* LIST \"\" \*"],
lambda self: self.assertEquals(self.imap_account._build_folder_cache(),
{"INBOX":
{"dir1":
{"subdir1": {},
"subdir2": {}},
"dir2": {}}}))
def test_ls_dir_base(self):
self.test_build_folder_cache()
self.assertEquals(self.imap_account.ls_dir(""),
["INBOX"])
def test_ls_dir_subdir(self):
self.test_build_folder_cache()
result = self.imap_account.ls_dir("INBOX")
result.sort()
self.assertEquals(result,
["dir1", "dir2"])
def test_ls_dir_subsubdir_delim1(self):
self.test_build_folder_cache()
self.imap_account.default_delimiter = "."
result = self.imap_account.ls_dir("INBOX/dir1")
result.sort()
self.assertEquals(result,
["subdir1", "subdir2"])
def test_ls_dir_subsubdir_delim2(self):
self.test_build_folder_cache()
result = self.imap_account.ls_dir("INBOX/dir1")
result.sort()
self.assertEquals(result,
["subdir1", "subdir2"])
class SMTPAccount_TestCase(Account_TestCase):
def setUp(self):
JCLTestCase.setUp(self, tables=[Account, ExampleAccount, User,