From 796663490012d53329d5f062385a26897e6d5291 Mon Sep 17 00:00:00 2001 From: Plamen Dimitrov Date: Wed, 11 Jul 2012 15:41:57 +0200 Subject: [PATCH] Internal cyrus formats wrapper was just implemented --- src/file_iterator.py | 37 +++++++++++++++++++++++------- src/imap_restore_mail.py | 8 +------ src/mail_iterator.py | 1 + src/unit_tester.py | 54 ++++++++++++++++++++++----------------------- 4 files changed, 56 insertions(+), 44 deletions(-) diff --git a/src/file_iterator.py b/src/file_iterator.py index edf6be7..506b4d2 100644 --- a/src/file_iterator.py +++ b/src/file_iterator.py @@ -32,14 +32,14 @@ class FileIterator: # mailboxes to update during file traversal acl_mailboxes = None # acls retrieved from a file - file_acls = None + _file_acls = None def __init__(self): """Creates a connection and a user session.""" self.created_mailboxes = [] self.acl_mailboxes = [] - self.file_acls = {} + self._file_acls = {} @classmethod def _message_read(cls, filename): @@ -54,7 +54,28 @@ class FileIterator: return message - def load_mailbox_list(self, mboxlistfile, original_user): + def get_mailbox_acls(self, mailbox, original_user): + """Get the acls loaded from a file for a given mailbox. + Should be used for access of mailbox acls instead of attribute""" + + # translate the mailbox to internal stored format + internal_mailbox = mailbox.replace("INBOX/", "user/" + original_user + "/") + internal_mailbox = internal_mailbox.replace(".", "^") + internal_mailbox = internal_mailbox.replace("/", ".") + logging.debug("Get acls for mailbox %s translated as %s", mailbox, internal_mailbox) + + # retrieve from internal dicitonary attribute of file acls + try: + mb_acls = self._file_acls[internal_mailbox] + except KeyError: + # no rights for the mailbox were found and warn if acl file loaded + if len(self._file_acls) > 0: + logging.warning("Could not find the acls for mailbox %s %s", mailbox, self._file_acls.keys()) + mb_acls = {} + + return mb_acls + + def load_mailbox_list(self, mboxlistfile): """Load the list of mailboxes and acl rights for each from file.""" try: @@ -71,10 +92,8 @@ class FileIterator: aclstr = linedata[1] # changes acls key encoding to internal cyrus format and makes it absolute (as folder). - key = linedata[0].replace("INBOX/", "user/" + original_user + "/") - key = key.replace(".", "^") - key = key.replace("/", ".") - + key = linedata[0] + # loop through acl rights string and build dictionary of users and rights while(aclstr != ""): try: @@ -83,10 +102,10 @@ class FileIterator: logging.warning("Illegal acl string in mailbox list dump: %s", line) aclstr = "" continue - aclstr = acldata[2] acls[acldata[0]] = acldata[1] + aclstr = acldata[2] - self.file_acls[key] = acls + self._file_acls[key] = acls except IOError: logging.warning("Could not open mboxlist file %s", mboxlistfile) diff --git a/src/imap_restore_mail.py b/src/imap_restore_mail.py index e681ec0..a6390e3 100644 --- a/src/imap_restore_mail.py +++ b/src/imap_restore_mail.py @@ -72,13 +72,7 @@ def update_mailboxes(storage, session, args): # add acls after all subfolders or their acls will be derived from parent folder for acl_mailbox in storage.acl_mailboxes: # find folder acls record and retrieve them - try: - mb_acls = storage.file_acls[acl_mailbox] - except KeyError: - # no rights for the mailbox were found and warn if acl file loaded - if len(storage.file_acls) > 0: - logging.warning("Could not find the acls for mailbox %s for user %s.", acl_mailbox, args.ouser) - mb_acls = {} + mb_acls = storage.get_mailbox_acls(acl_mailbox) session.add_acls(acl_mailbox, mb_acls, args.ouser, args.user) storage.created_mailboxes = [] storage.acl_mailboxes = [] diff --git a/src/mail_iterator.py b/src/mail_iterator.py index ccd5706..ffc8b64 100644 --- a/src/mail_iterator.py +++ b/src/mail_iterator.py @@ -110,6 +110,7 @@ class MailIterator: def add_acls(self, mailbox, mb_acls, original_user, target_user): """Add acls to mailbox.""" + mailbox = mailbox.replace("user/" + original_user + "/", "INBOX/") for acl_user in mb_acls: # (in case target user != original user): # - don't overwrite acls eventually set for the current targetuser diff --git a/src/unit_tester.py b/src/unit_tester.py index c8bf83d..5ad9af9 100644 --- a/src/unit_tester.py +++ b/src/unit_tester.py @@ -29,8 +29,6 @@ class FileContentsParse(unittest.TestCase): dummy_filename = None # dummy file descriptor to be used for parsing the acls dummy_file = None - # dummy original user - dummy_user = None # wrong naming conventions are due to python unittest library requirement def setUp(self): @@ -40,7 +38,7 @@ class FileContentsParse(unittest.TestCase): self.dummy_file = open(self.dummy_filename, "w") self.dummy_file.write("user.00schneider\t0 default 00schneider\tlrswipkxtecda\t\n" "user.mit^punkt.Gesendete Objekte\t0 default mit.punkt\tlrswipktecda\t\n" - "user.00schneider.ibx_poped\t0 default 00schneider\tlrswipkxtecda\t\n" + "user.00schneider.Gesendete Objekte\t0 default 00schneider\tlrswipkxtecda\t\n" "user.00schneider.ibx_sub\t0 default 00schneider\tlrswpkxtecda\t\n" "user.abrus\t0 default abrus\tlrswipkxtecda\t\n" "user.abrus.Aufgaben\t0 default abrus\tlrswipkxtecda\tabrus2\tlrswikxtecda\t\n" @@ -63,38 +61,38 @@ class FileContentsParse(unittest.TestCase): def test_mboxlist_parsed_line1(self): """Test whether a line was parsed correctly.""" - self.file_iter.load_mailbox_list(self.dummy_filename, self.dummy_user) - self.assertIn("user^00schneider", self.file_iter.file_acls) - self.assertIn("00schneider", self.file_iter.file_acls["user^00schneider"]) - self.assertEqual(self.file_iter.file_acls["user^00schneider"]["00schneider"], "lrswipkxtecda", "Wrong acls were parsed for a user.") + self.file_iter.load_mailbox_list(self.dummy_filename) + self.assertEqual(len(self.file_iter.get_mailbox_acls("INBOX/Gesendete Objekte", "00schneider")), 1) + self.assertIn("00schneider", self.file_iter.get_mailbox_acls("INBOX/Gesendete Objekte", "00schneider")) + self.assertEqual(self.file_iter.get_mailbox_acls("INBOX/Gesendete Objekte", "00schneider")["00schneider"], "lrswipkxtecda", "Wrong acls were parsed for a user.") def test_mboxlist_parsed_line2(self): """Test whether a line was parsed correctly.""" - self.file_iter.load_mailbox_list(self.dummy_filename, self.dummy_user) - self.assertIn("user^mit^punkt^Gesendete Objekte", self.file_iter.file_acls) - self.assertIn("mit.punkt", self.file_iter.file_acls["user^mit^punkt^Gesendete Objekte"]) - self.assertEqual(self.file_iter.file_acls["user^mit^punkt^Gesendete Objekte"]["mit.punkt"], "lrswipktecda", "Wrong acls were parsed for a user.") - + self.file_iter.load_mailbox_list(self.dummy_filename) + self.assertEqual(len(self.file_iter.get_mailbox_acls("INBOX/Gesendete Objekte", "mit.punkt")), 1) + self.assertIn("mit.punkt", self.file_iter.get_mailbox_acls("INBOX/Gesendete Objekte", "mit.punkt")) + self.assertEqual(self.file_iter.get_mailbox_acls("INBOX/Gesendete Objekte", "mit.punkt")["mit.punkt"], "lrswipktecda", "Wrong acls were parsed for a user.") + def test_mboxlist_totals(self): """Test whether there are missing parsed lines, mailboxs, and users.""" - self.file_iter.load_mailbox_list(self.dummy_filename, self.dummy_user) - self.assertEqual(len(self.file_iter.file_acls), 13) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 2) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - self.assertEqual(len(self.file_iter.file_acls.popitem()[1]), 1) - + self.file_iter.load_mailbox_list(self.dummy_filename) + self.assertEqual(len(self.file_iter._file_acls), 13) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 2) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + self.assertEqual(len(self.file_iter._file_acls.popitem()[1]), 1) + if __name__ == '__main__': unittest.main() \ No newline at end of file -- 1.7.1