Kolab Groupware OBS

bonnie-0.1.tar.gz/INSTALL Changed

bonnie-0.1.tar.gz/bonnie/broker/__init__.py Changed

@@ -1,3 +1,28 @@
+# -*- coding: utf-8 -*-
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Jeroen van Meeuwen (Kolab Systems) <vanmeeuwen a kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later
+# version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
+# USA.
+#
+
+"""
+    This is the broker for Bonnie.
+"""
+
 import brokers
 
 class BonnieBroker(object):
@@ -6,8 +31,9 @@
 
     def __init__(self, *args, **kw):
         for _class in brokers.list_classes():
-            __class = _class()
-            self.broker_modules[__class] = __class.register(callback=self.register_broker)
+            module = _class()
+            module.register(callback=self.register_broker)
+            self.broker_modules[_class] = module
 
     def register_broker(self, interests):
         """
@@ -24,3 +50,10 @@
         for interest, hows in self.broker_interests.iteritems():
             for how in hows:
                 how()
+
+    def terminate(self, *args, **kw):
+        for module in self.broker_modules.values():
+            if hasattr(module, 'terminate'):
+                module.terminate()
+            else:
+                module.running = False

bonnie-0.1.tar.gz/bonnie/broker/brokers/zmq_broker/__init__.py Changed

@@ -20,7 +20,7 @@
 #
 
 """
-    This is the broker for Bonnie.
+    This is the ZMQ broker implementation for Bonnie.
 """
 
 import copy
@@ -31,28 +31,37 @@
 
 import bonnie
 conf = bonnie.getConf()
-log = bonnie.getLogger('broker.ZMQBroker')
+log = bonnie.getLogger('bonnie.broker.ZMQBroker')
 
 from job import Job
 from collector import Collector
 from worker import Worker
+from bonnie.broker import persistence
 
 class ZMQBroker(object):
+    running = False
+
     def __init__(self):
-        self.worker_jobs = []
-        self.collect_jobs = []
+        self.worker_jobs = persistence.List('worker_jobs', Job)
+        self.collect_jobs = persistence.List('collect_jobs', Job)
         self.collectors = {}
         self.workers = {}
+        self.collector_interests = []
 
     def register(self, callback):
         callback({ '_all': self.run })
 
-    def collector_add(self, _collector_id, _state):
-        log.debug("Adding collector %s" % (_collector_id), level=5)
+    def collector_add(self, _collector_id, _state, _interests):
+        log.debug("Adding collector %s for %r" % (_collector_id, _interests), level=5)
 
-        collector = Collector(_collector_id, _state)
+        collector = Collector(_collector_id, _state, _interests)
         self.collectors[_collector_id] = collector
 
+        # regisrer the reported interests
+        if len(_interests) > 0:
+            _interests.extend(self.collector_interests)
+            self.collector_interests = list(set(_interests))
+
     def collect_job_allocate(self, _collector_id):
         jobs = [x for x in self.collect_jobs if x.collector_id == _collector_id and x.state == b"PENDING"]
 
@@ -66,9 +75,9 @@
     def collect_jobs_with_status(self, _state, collector_id=None):
         return [x for x in self.collect_jobs if x.state == _state and x.collector_id == collector_id]
 
-    def collector_set_status(self, _collector_id, _state):
+    def collector_set_status(self, _collector_id, _state, _interests):
         if not self.collectors.has_key(_collector_id):
-            self.collector_add(_collector_id, _state)
+            self.collector_add(_collector_id, _state, _interests)
         else:
             self.collectors[_collector_id].set_status(_state)
 
@@ -82,7 +91,6 @@
         job = Job(
                 notification=_notification,
                 state=b"PENDING",
-                worker=None,
                 client_id=client_id,
                 collector_id=collector_id,
             )
@@ -107,13 +115,13 @@
         job = jobs.pop()
 
         job.set_state(b"ALLOC")
-        job.set_worker(self.workers[_worker_id])
+        job.set_worker(_worker_id)
 
         return job.uuid
 
     def worker_job_done(self, _job_uuid):
         for job in [x for x in self.worker_jobs if x.uuid == _job_uuid]:
-            del self.worker_jobs[self.worker_jobs.index(job)]
+            self.worker_jobs.delete(job)
         log.debug("Worker job done: %s;" % (_job_uuid), level=8)
 
     def worker_job_free(self, _job_uuid):
@@ -132,7 +140,7 @@
         return [x for x in self.worker_jobs if x.state == _state]
 
     def worker_jobs_with_worker(self, _worker_id):
-        return [x for x in self.worker_jobs if x.worker == self.workers[_worker_id]]
+        return [x for x in self.worker_jobs if x.worker_id == _worker_id]
 
     def worker_add(self, _worker_id, _state):
         log.debug("Adding worker %s (%s)" % (_worker_id, _state), level=5)
@@ -172,6 +180,7 @@
 
     def run(self):
         log.info("Starting")
+        self.running = True
 
         context = zmq.Context()
 
@@ -213,8 +222,23 @@
         poller.register(self.worker_router, zmq.POLLIN)
         poller.register(self.controller, zmq.POLLIN)
 
-        while True:
-            sockets = dict(poller.poll(1000))
+        # reset existing jobs in self.worker_jobs and self.collect_jobs to status PENDING (?)
+        # this will re-assign them to workers and collectors after a broker restart
+        for job in self.worker_jobs:
+            job.set_state(b"PENDING")
+
+        for job in self.collect_jobs:
+            job.set_state(b"PENDING")
+
+        persistence.syncronize()
+
+        while self.running:
+            try:
+                sockets = dict(poller.poll(1000))
+            except Exception, e:
+                log.error("zmq.Poller error: %r", e)
+                sockets = dict()
+
             self.workers_expire()
 
             if self.controller in sockets:
@@ -230,7 +254,7 @@
                     if _message[1] == b"DONE":
                         self.worker_job_done(_message[2])
 
-                    if _message[1] in [b"FETCH", b"HEADER", b"GETMETADATA", b"GETACL"]:
+                    if _message[1] in self.collector_interests:
                         _job_uuid = _message[2]
                         self.transit_job_collect(_job_uuid, _message[1])
 
@@ -254,7 +278,8 @@
                     if _message[1] == b"STATE":
                         _collector_id = _message[0]
                         _state = _message[2]
-                        self.collector_set_status(_collector_id, _state)
+                        _interests = _message[3]
+                        self.collector_set_status(_collector_id, _state, _interests.split(","))
 
                     if _message[1] == b"DONE":
                         _collector_id = _message[0]
@@ -290,7 +315,13 @@
                     job = self.collect_job_allocate(collector)
                     self.collector_router.send_multipart([job.collector_id, job.command, job.uuid, job.notification])
 
+            # synchronize job lists to persistent storage
+            persistence.syncronize()
+
+
+        log.info("Shutting down")
 
+        persistence.syncronize()
         dealer_router.close()
         self.controller.close()
         self.collector_router.close()
@@ -302,12 +333,12 @@
             job.set_state(b"PENDING")
             job.set_command(_command)
             self.collect_jobs.append(job)
-            del self.worker_jobs[self.worker_jobs.index(job)]
+            self.worker_jobs.remove(job)
 
     def transit_job_worker(self, _job_uuid, _notification):
         for job in [x for x in self.collect_jobs if x.uuid == _job_uuid]:
             job.set_state(b"PENDING")
             job.notification = _notification
             self.worker_jobs.append(job)
-            del self.collect_jobs[self.collect_jobs.index(job)]
+            self.collect_jobs.remove(job)

bonnie-0.1.tar.gz/bonnie/broker/brokers/zmq_broker/collector.py Changed

bonnie-0.1.tar.gz/bonnie/broker/brokers/zmq_broker/job.py Changed

@@ -21,20 +21,26 @@
 
 import hashlib
 import time
+from bonnie.broker.persistence import db, PersistentBase
 
-class Job(object):
-    state = None
-    worker_id = None
-    timestamp = None
-    notification = None
-    client_id = None
+class Job(PersistentBase):
+    __tablename__ = 'jobs'
+    # use binary types because ZMQ requires binary strings
+    uuid = db.Column(db.LargeBinary(128), primary_key=True)
+    type = db.Column(db.String(16))
+    state = db.Column(db.String(16))
+    timestamp = db.Column(db.Float)
+    notification = db.Column(db.LargeBinary)
+    worker_id = db.Column(db.String(64))
+    client_id = db.Column(db.String(64))
+    collector_id = db.Column(db.LargeBinary(64))
+    command = db.Column(db.LargeBinary(32))
 
-    def __init__(self, state=None, worker=None, notification=None, client_id=None, collector_id=None):
-        self.uuid = hashlib.sha224(notification).hexdigest()
+    def __init__(self, state=None, notification=None, worker_id=None, client_id=None, collector_id=None):
         self.uuid = "%s.%s" % (hashlib.sha224(notification).hexdigest(), time.time())
         self.state = state
-        self.worker = worker
         self.notification = notification
+        self.worker_id = worker_id
         self.client_id = client_id
         self.collector_id = collector_id
         self.timestamp = time.time()
@@ -51,8 +57,8 @@
     def set_state(self, state):
         self.state = state
 
-    def set_worker(self, worker):
-        self.worker = worker
+    def set_worker(self, worker_id):
+        self.worker_id = worker_id
 
     def set_command(self, cmd):
         self.command = cmd

bonnie-0.1.tar.gz/bonnie/broker/persistence.py Added

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Thomas Bruederli <bruederli at kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later
+# version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
+# USA.
+#
+
+import sqlalchemy as db
+from sqlalchemy.orm import sessionmaker, relationship
+from sqlalchemy.orm.util import has_identity
+from sqlalchemy.orm.attributes import init_collection
+from sqlalchemy.ext.declarative import declarative_base
+
+import bonnie
+conf = bonnie.getConf()
+log = bonnie.getLogger('bonnie.broker.persistence')
+
+PersistentBase = declarative_base()
+
+# an db engine, which the Session will use for connection resources
+engine = db.create_engine(conf.get('broker', 'persistence_sql_uri', 'sqlite://'))
+
+# create a configured "Session" class
+Session = sessionmaker(bind=engine)
+session = Session()
+
+
+class PlistCollection(list):
+    """
+        Extended list collection to add some handy utility methods
+    """
+    def delete(self, item):
+        self.remove(item)
+        if has_identity(item):
+            session.delete(item)
+        else:
+            session.expunge(item)
+
+class PersistentList(PersistentBase):
+    """
+        Container class representing a persistent list object
+    """
+    __tablename__ = 'plists'
+    listname = db.Column(db.String(32), primary_key=True)
+    # items = relationship('PlistItem')
+
+    def __init__(self, name):
+        self.listname = name
+
+
+#### module functions
+
+__list_classes = {}
+__list_instances = {}
+
+def List(name, _type):
+    """
+        Factory function to return a list-like collection with persistent storage capabilities
+    """
+    if __list_instances.has_key(name):
+        return __list_instances[name].items
+
+    # create new list class to handle items (relations) of the given type
+    _class_name = 'PersistentList' + _type.__name__
+
+    if __list_classes.has_key(_class_name):
+        _plistclass = __list_classes[_class_name]
+    else:
+        # determine foreign key type
+        if hasattr(_type, '__key__') and _type.__table__.columns.has_key(_type.__key__):
+            reltype = _type.__table__.columns[_type.__key__].type
+        elif hasattr(_type.__table__, 'primary_key'):
+            for col in _type.__table__.primary_key.columns:
+                _type.__key__ = col.name
+                reltype = col.type
+                break
+        else:
+            _type.__key__ = 'id'
+            reltype = db.String(256)
+
+        # we establish a many-to-many relation using an association table
+        association_table = db.Table('plistitems_' + _type.__tablename__, PersistentBase.metadata,
+            db.Column('listname', db.Integer, db.ForeignKey('plists.listname'), index=True),
+            db.Column('ref', reltype, db.ForeignKey(_type.__tablename__ + '.' + _type.__key__), index=True)
+        )
+
+        # create a list container class with a relationship to the list item type
+        _plistclass = type(_class_name, (PersistentList,), {
+            'items': relationship(_type, secondary=association_table, backref='list', collection_class=PlistCollection)
+        })
+        __list_classes[_class_name] = _plistclass
+
+        PersistentBase.metadata.create_all(engine)
+
+    try:
+        lst = session.query(_plistclass).filter(PersistentList.listname == name).one()
+    except:
+        lst = _plistclass(name)
+        session.add(lst)
+
+    # remember this instance for later calls and to avoid garbage collection
+    __list_instances[name] = lst
+
+    # return the collection og list items
+    return lst.items
+
+
+def syncronize():
+    """
+        Synchronize to persistent storage
+    """
+    if len(session.new) > 0 or len(session.dirty) > 0 or len(session.deleted) > 0:
+        log.debug("session.commit(); new=%r; dirty=%r; deleted=%r" % (session.new, session.dirty, session.deleted), level=9)
+        session.commit()

bonnie-0.1.tar.gz/bonnie/collector/__init__.py Changed

@@ -20,186 +20,69 @@
 # USA.
 #
 
-import elasticsearch
-import json
 import os
-import subprocess
-import time
-import urllib
-import urlparse
-
 import inputs
+import handlers
 
 import bonnie
+from bonnie.utils import parse_imap_uri
+
 conf = bonnie.getConf()
-log = bonnie.getLogger('collector')
+log = bonnie.getLogger('bonnie.collector')
 
 class BonnieCollector(object):
     input_interests = {}
     input_modules = {}
 
-    def __init__(self, *args, **kw):
-        for _class in inputs.list_classes():
-            __class = _class()
-            self.input_modules[__class] = __class.register(callback=self.register_input)
-
-    def expand_uidset(uidset):
-        _uids = []
-        for _uid in uidset.split(','):
-            if len(_uid.split(':')) > 1:
-                for __uid in range((int)(_uid.split(':')[0]), (int)(_uid.split(':')[1])+1):
-                    _uids.append("%d" % (__uid))
-            else:
-                _uids.append(str(_uid))
-
-        return _uids
-
-    def retrieve_contents_for_messages(self, notification):
-        messageContents = {}
-
-        notification = json.loads(notification)
-        log.debug("FETCH for %r" % (notification), level=9)
-
-        split_uri = urlparse.urlsplit(notification['uri'])
-
-        if len(split_uri.netloc.split('@')) == 3:
-            (username, domain, server) = split_uri.netloc.split('@')
-        elif len(split_uri.netloc.split('@')) == 2:
-            (username, server) = split_uri.netloc.split('@')
-            domain = None
-        elif len(split_uri.netloc.split('@')) == 1:
-            username = None
-            domain = None
-            server = split_uri.netloc
-
-        # First, .path == '/Calendar/Personal%20Calendar;UIDVALIDITY=$x[/;UID=$y]
-        # Take everything after the first slash, and omit any INBOX/ stuff.
-        path_part = '/'.join([x for x in split_uri.path.split('/') if not x == "INBOX"][1:])
-
-        # Second, .path == 'Calendar/Personal%20Calendar;UIDVALIDITY=$x[/;UID=$y]
-        # Take everything before the first ';' (but only actually take everything
-        # before the first ';' in the next step, here we still have use for it).
-        # TODO: parse the path/query parameters into a dict
-        path_part = path_part.split(';')
-
-        if notification.has_key('uidset'):
-            message_uids = expand_uidset(notification['uidset'])
-        elif notification.has_key('vnd.cmu.oldUidset'):
-            message_uids = expand_uidset(notification['vnd.cmu.oldUidset'])
-        elif len(path_part) == 3:
-            # Use or abuse the length of path_parts at this moment to extract the message UID
-            message_uids = [ path_part[2].split('=')[1] ]
-            notification['uidset'] = message_uids
-
-
-        # Third, .path == 'Calendar/Personal%20Calendar
-        # Decode the url encoding
-        folder_name = urllib.unquote(path_part[0])
-
-        # Through filesystem
-        # To get the mailbox path, use:
-        # TODO: Assumption #1 is we are using virtual domains, and this domain does
-        # TODO: Assumption #2 is the mailbox in question is a user mailbox
-        # TODO: Assumption #3 is we use the unix hierarchy separator
-
-        # Translate the folder name in to a fully qualified folder path such as it
-        # would be used by a cyrus administrator.
-        #
-        # TODO: Other Users (covered, the netloc has the username the suffix is the
-        # original folder name).
-        #
-        # TODO: Shared Folders.
-        if not username == None:
-            if folder_name == "INBOX":
-                folder_path = 'user/%s@%s' % (username, domain)
-            else:
-                folder_path = 'user/%s/%s@%s' % (username, folder_name, domain)
-        else:
-            folder_path = folder_name
-
-        # TODO: Check if this file exists and is actually executable
-        # New in Python 2.7:
-        if hasattr(subprocess, 'check_output'):
-            mailbox_path = subprocess.check_output(
-                    ["/usr/lib/cyrus-imapd/mbpath", folder_path]
-                ).strip()
-
-            log.debug("Using mailbox path: %r" % (mailbox_path), level=8)
-
-        else:
-            # Do it the old-fashioned way
-            p1 = subprocess.Popen(
-                    ["/usr/lib/cyrus-imapd/mbpath", folder_path],
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE
-                )
-
-            (stdout, stderr) = p1.communicate()
-
-            mailbox_path = stdout.strip()
-            log.debug("Using mailbox path: %r" % (mailbox_path), level=8)
-
-        # TODO: Assumption #4 is we use altnamespace
-        if not folder_name == "INBOX":
-            if not len(folder_name.split('@')) > 0:
-                mailbox_path = "%s/%s" % (mailbox_path, folder_name)
-
-        # using message file path /var/spool/imap/domain/e/example.org/k/lists/kolab^org/devel/lists/kolab.org/devel@example.org/1.
-        for message_uid in message_uids:
-            message_file_path = "%s/%s." % (mailbox_path, message_uid)
+    handler_interests = {}
+    handler_modules = {}
 
-            log.debug("Open message file: %r" % (message_file_path), level=8)
 
-            if os.access(message_file_path, os.R_OK):
-                fp = open(message_file_path, 'r')
-                data = fp.read()
-                fp.close()
-
-                # TODO: parse mime message and return a dict
-                messageContents[message_uid] = data
-
-        notification['messageContent'] = messageContents
-
-        return json.dumps(notification)
-
-    def retrieve_headers_for_messages(self, notification):
-        messageHeaders = {}
-
-        notification = json.loads(notification)
-        log.debug("HEADERS for %r" % (notification), level=9)
-
-        # TODO: resovle message file system path
-        # TODO: read file line by line until we reach an empty line
-        # TODO: decode quoted-pritable or base64 encoded headers
-
-        notification['messageHeaders'] = messageHeaders
+    def __init__(self, *args, **kw):
+        # TODO: read active input module from config collector.input_modules
+        for _class in inputs.list_classes():
+            module = _class()
+            module.register(callback=self.register_input)
+            self.input_modules[_class] = module
 
-        return json.dumps(notification)
+        # TODO: read active handler module from config collector.handler_modules
+        for _class in handlers.list_classes():
+            handler = _class()
+            handler.register(callback=self.register_handler)
+            self.handler_modules[_class] = handler
 
     def execute(self, command, notification):
         """
-            Our goal is to collect whatever message contents
-            for the messages referenced in the notification.
+            Dispatch collector job to the according handler(s)
         """
         log.debug("Executing collection command %s" % (command), level=8)
-        if command == "FETCH":
-            notification = self.retrieve_contents_for_messages(notification)
-        elif command == "HEADER":
-            notification = self.retrieve_headers_for_messages(notification)
-        elif command == "GETMETADATA":
-            pass
-        elif command == "GETACL":
-            pass
+
+        if self.handler_interests.has_key(command):
+            for interest in self.handler_interests[command]:
+                notification = interest['callback'](notification=notification)
 
         return notification
-        #self.output(notification)

bonnie-0.1.tar.gz/bonnie/collector/handlers Added

bonnie-0.1.tar.gz/bonnie/collector/handlers/__init__.py Added

bonnie-0.1.tar.gz/bonnie/collector/handlers/imapdata.py Added

@@ -0,0 +1,92 @@
+# -*- coding: utf-8 -*-
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Thomas Bruederli <bruederli at kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later version
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+#
+
+import json
+import pykolab
+from pykolab.imap import IMAP
+
+import bonnie
+from bonnie.utils import parse_imap_uri
+from bonnie.utils import imap_folder_path
+
+conf = bonnie.getConf()
+log = bonnie.getLogger('bonnie.collector.IMAPDataHandler')
+
+class IMAPDataHandler(object):
+    """
+        Collector handler to provide metadata from IMAP
+    """
+
+    def __init__(self, *args, **kw):
+        # load pykolab conf
+        conf = pykolab.getConf()
+        if not hasattr(conf, 'defaults'):
+            conf.finalize_conf()
+
+        self.imap = IMAP()
+
+    def register(self, callback):
+        interests = {
+            'GETMETADATA':  { 'callback': self.get_imap_folder_metadata },
+            'GETACL':  { 'callback': self.get_imap_folder_acl }
+        }
+
+        callback(interests)
+
+    def get_imap_folder_metadata(self, notification):
+        notification = json.loads(notification)
+        log.debug("GETMETADATA for %r" % (notification), level=9)
+
+        # split the uri parameter into useful parts
+        uri = parse_imap_uri(notification['uri'])
+        folder_path = imap_folder_path(uri)
+
+        # get metadata using pykolab's imap module
+        metadata = {}
+        try:
+            self.imap.connect()
+            metadata = self.imap.get_metadata(folder_path)[folder_path]
+            self.imap.disconnect()
+        except Exception, e:
+            log.warning("Failed to get metadata for %r: %r", folder_path, e)
+
+        notification['metadata'] = metadata
+
+        return json.dumps(notification)
+
+    def get_imap_folder_acl(self, notification):
+        notification = json.loads(notification)
+        log.debug("GETACL for %r" % (notification), level=9)
+
+        # split the uri parameter into useful parts
+        uri = parse_imap_uri(notification['uri'])
+        folder_path = imap_folder_path(uri)
+
+        # get folder acls using pykolab's imap module
+        acls = {}
+        try:
+            self.imap.connect()
+            acls = self.imap.list_acls(folder_path)
+            self.imap.disconnect()
+        except Exception, e:
+            log.warning("Failed to get ACLs for %r: %r", folder_path, e)
+
+        notification['acl'] = acls
+
+        return json.dumps(notification)

bonnie-0.1.tar.gz/bonnie/collector/handlers/messagedata.py Added

@@ -0,0 +1,151 @@
+# -*- coding: utf-8 -*-
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Thomas Bruederli <bruederli at kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later version
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+#
+
+import os
+import time
+import json
+
+from email import message_from_string
+
+import bonnie
+from bonnie import utils
+from bonnie.utils import expand_uidset
+from bonnie.utils import parse_imap_uri
+from bonnie.utils import decode_message_headers
+
+conf = bonnie.getConf()
+log = bonnie.getLogger('bonnie.collector.MessageDataHandler')
+
+class MessageDataHandler(object):
+
+    def __init__(self, *args, **kw):
+        pass
+
+    def register(self, callback):
+        interests = {
+            'FETCH':  { 'callback': self.retrieve_contents_for_messages },
+            'HEADER': { 'callback': self.retrieve_headers_for_messages }
+        }
+
+        callback(interests)
+
+    def retrieve_contents_for_messages(self, notification):
+        notification = json.loads(notification)
+        log.debug("FETCH for %r" % (notification), level=9)
+
+        (messageContents, messageHeaders) = self._retrieve_contents_for_messages(notification)
+
+        # set new notification properties, even if empty
+        notification['messageContent'] = messageContents
+        notification['messageHeaders'] = messageHeaders
+
+        return json.dumps(notification)
+
+    def retrieve_headers_for_messages(self, notification):
+        notification = json.loads(notification)
+        log.debug("HEADERS for %r" % (notification), level=9)
+
+        (messageContents, messageHeaders) = self._retrieve_contents_for_messages(notification, True)
+
+        # set new notification properties, even if empty
+        notification['messageHeaders'] = messageHeaders
+
+        return json.dumps(notification)
+
+    def _retrieve_contents_for_messages(self, notification, headers_only=False):
+        """
+            Helper method to deliver message contents/headers
+        """
+        messageContents = {}
+        messageHeaders = {}
+
+        # split the uri parameter into useful parts
+        uri = parse_imap_uri(notification['uri'])
+
+        if notification.has_key('uidset'):
+            message_uids = expand_uidset(notification['uidset'])
+        elif notification.has_key('vnd.cmu.oldUidset'):
+            message_uids = expand_uidset(notification['vnd.cmu.oldUidset'])
+        elif uri.has_key('UID'):
+            message_uids = [ uri['UID'] ]
+            notification['uidset'] = ','.join(message_uids)
+
+        # resolve uri into a mailbox path on the local file stystem
+        mailbox_path = utils.imap_mailbox_fs_path(uri)
+
+        log.debug("Using mailbox path: %r" % (mailbox_path), level=8)
+
+        # mailbox exists, try reading the message files
+        if os.path.exists(mailbox_path):
+            for message_uid in message_uids:
+                # using message file path like /var/spool/imap/domain/e/example.org/k/lists/kolab^org/devel/lists/kolab.org/devel@example.org/1.
+                message_file_path = "%s/%s." % (mailbox_path, message_uid)
+
+                log.debug("Open message file: %r" % (message_file_path), level=8)
+
+                attempts = 5
+                while attempts > 0:
+                    attempts -= 1
+
+                    if os.access(message_file_path, os.R_OK):
+                        fp = open(message_file_path, 'r')
+
+                        if headers_only:
+                            data = ''
+                            for line in fp:
+                                data += line
+                                if line.strip() == '':
+                                    break;
+                            data += "\r\n"
+                        else:
+                            data = fp.read()
+
+                        fp.close()
+
+                        # use email lib to parse message headers
+                        try:
+                            # find header delimiter
+                            pos = data.find("\r\n\r\n")
+                            message = message_from_string(data[0:pos])
+                            headers = decode_message_headers(message)
+                        except:
+                            headers = dict()
+                            messageHeaders[message_uid] = headers
+
+                        # append raw message data and parsed headers
+                        messageContents[message_uid] = data
+                        messageHeaders[message_uid] = headers
+                        break
+
+                    elif attempts > 0:
+                        log.debug("Failed to open message file %r; retry %d more times" % (
+                            message_file_path, attempts
+                        ), level=5)
+                    else:
+                        log.warning("Failed to open message file %r for uri=%s; uid=%s" % (
+                            message_file_path, notification, message_uid
+                        ))
+
+                    time.sleep(1)
+                # end while
+            # end for
+        else:
+            log.warning("Mailbox path %r does not exits for uri=%s" % (mailbox_path, notification['uri']))
+
+        return (messageContents, messageHeaders)

bonnie-0.1.tar.gz/bonnie/collector/inputs/zmq_input.py Changed

@@ -34,10 +34,11 @@
 
 import bonnie
 conf = bonnie.getConf()
-log = bonnie.getLogger('collector.ZMQInput')
+log = bonnie.getLogger('bonnie.collector.ZMQInput')
 
 class ZMQInput(object):
     state = b"READY"
+    running = False
 
     def __init__(self, *args, **kw):
         self.context = zmq.Context()
@@ -61,18 +62,25 @@
     def register(self, *args, **kw):
         pass
 
-    def report_state(self):
-        log.debug("[%s] Reporting state %s" % (self.identity, self.state), level=9)
-        self.collector.send_multipart([b"STATE", self.state])
+    def report_state(self, interests=[]):
+        log.debug("[%s] Reporting state %s, %r" % (self.identity, self.state, interests), level=9)
+        self.collector.send_multipart([b"STATE", self.state, ",".join(interests)])
         self.report_timestamp = time.time()
 
-    def run(self, callback=None):
+    def run(self, callback=None, interests=[]):
         log.info("[%s] starting", self.identity)
 
-        self.report_state()
+        self.running = True
 
-        while True:
-            sockets = dict(self.poller.poll(1000))
+        # report READY state with interests
+        self.report_state(interests)
+
+        while self.running:
+            try:
+                sockets = dict(self.poller.poll(1000))
+            except Exception, e:
+                log.error("zmq.Poller error: %r", e)
+                sockets = dict()
 
             if self.report_timestamp < (time.time() - 60):
                 self.report_state()
@@ -97,4 +105,5 @@
 
                             self.collector.send_multipart([b"DONE", _job_uuid, result])
 
+        log.info("[%s] shutting down", self.identity)
         self.collector.close()

bonnie-0.1.tar.gz/bonnie/dealer/__init__.py Changed

@@ -20,10 +20,12 @@
 # USA.
 #
 
+import json
 import outputs
 
 import bonnie
 conf = bonnie.getConf()
+log = bonnie.getLogger('bonnie.dealer')
 
 class BonnieDealer(object):
     output_modules = {}
@@ -36,8 +38,25 @@
     def register_output(self, interests):
         self.output_interests = interests
 
+    def accept_notification(self, notification):
+        parsed = json.loads(notification)
+        event = parsed['event']
+        user = parsed['user'] if parsed.has_key('user') else None
+
+        blacklist_events = conf.get('dealer', 'blacklist_events').split(',')
+        blacklist_users  = conf.get('dealer', 'blacklist_users').split(',')
+
+        # ignore blacklisted events for blacklisted users
+        if event in blacklist_events and user is not None and user in blacklist_users:
+            return False
+
+        return True
+
     def run(self, notification):
-        output_modules = conf.get('dealer', 'output_modules')
-        for _output in self.output_modules.keys():
-            if _output.name() == output_modules:
-                _output.run(notification)
+        if self.accept_notification(notification):
+            output_modules = conf.get('dealer', 'output_modules')
+            for _output in self.output_modules.keys():
+                if _output.name() == output_modules:
+                    _output.run(notification)
+        else:
+            log.info("Ignoring notification %s", notification)

bonnie-0.1.tar.gz/bonnie/dealer/outputs/zmq_output.py Changed

bonnie-0.1.tar.gz/bonnie/utils.py Added

@@ -0,0 +1,224 @@
+import os
+import urllib
+import urlparse
+import datetime
+import subprocess
+
+from dateutil.parser import parse as parse_date
+from dateutil.tz import tzutc
+
+from email import message_from_string
+from email.header import decode_header
+from email.utils import getaddresses
+
+import bonnie
+conf = bonnie.getConf()
+log = bonnie.getLogger('bonnie.utils')
+
+
+def expand_uidset(uidset):
+    """
+        Expand the given UID set string into a complete set of values
+        Examples: 1,2,5 => [1,2,5] or 1:4 => [1,2,3,4]
+    """
+    _uids = []
+    for _uid in uidset.split(','):
+        if len(_uid.split(':')) > 1:
+            for __uid in range((int)(_uid.split(':')[0]), (int)(_uid.split(':')[1])+1):
+                _uids.append("%d" % (__uid))
+        else:
+            _uids.append(str(_uid))
+
+    return _uids
+
+
+def parse_imap_uri(uri):
+    """
+        Split the given URI string into its components
+    """
+    split_uri = urlparse.urlsplit(uri)
+
+    if len(split_uri.netloc.split('@')) == 3:
+        (username, domain, server) = split_uri.netloc.split('@')
+    elif len(split_uri.netloc.split('@')) == 2:
+        (username, server) = split_uri.netloc.split('@')
+        domain = None
+    elif len(split_uri.netloc.split('@')) == 1:
+        username = None
+        domain = None
+        server = split_uri.netloc
+
+    result = dict(user=username, domain=domain, host=server)
+
+    # First, .path == '/Calendar/Personal%20Calendar;UIDVALIDITY=$x[/;UID=$y]
+    # Take everything after the first slash, and omit any INBOX/ stuff.
+    path_str = '/'.join([x for x in split_uri.path.split('/') if not x == 'INBOX'][1:])
+    path_arr = path_str.split(';')
+    result['path'] = urllib.unquote(path_arr[0])
+
+    # parse the path/query parameters into a dict
+    param = dict()
+    for p in path_arr[1:]:
+        if '=' in p:
+            (key,val) = p.split('=', 2)
+            result[key] = urllib.unquote(val)
+
+    return result
+
+
+def decode_message_headers(message):
+    """
+        Copy headers from the given message object into a dict
+        structure and normalize the values into UTF-8 strings
+    """
+    headers = dict(message.items())
+
+    # split recipient headers into lists
+    for h in ['From','To','Cc','Bcc']:
+        if headers.has_key(h):
+            headers[h] = ['%s <%s>' % (name,address) for name,address in getaddresses(message.get_all(h))]
+
+    # decode header values into UTF-8
+    for k,value in headers.iteritems():
+        if isinstance(value, list):
+            headers[k] = [_decode_mime_header(x) for x in value]
+        else:
+            headers[k] = _decode_mime_header(value)
+
+    # replace content-type with the normalized value (good for searching)
+    headers['Content-Type'] = message.get_content_type()
+
+    # convert Date into UTC
+    if headers.has_key('Date'):
+        try:
+            date = parse_date(headers['Date']).astimezone(tzutc())
+            headers['Date'] = datetime.datetime.strftime(date, "%Y-%m-%dT%H:%M:%SZ")
+        except:
+            pass
+
+    return headers
+
+def _decode_mime_header(value):
+    """
+        Helper method to decode a single header string
+    """
+    return ' '.join(unicode(raw, charset) if charset is not None else raw for raw,charset in decode_header(value))
+
+
+def mail_message2dict(data):
+    """
+        Parse the given MIME message and return its contents as a dict
+    """
+    try:
+        message = message_from_string(data)
+    except Exception, e:
+        log.warning("Failed to parse MIME message: %r", e)
+        return dict(_data=data)
+
+    result = decode_message_headers(message)
+
+    if message.is_multipart():
+        for mime_part in message.walk():
+            # skip top level part
+            if not result.has_key('@parts'):
+                result['@parts'] = []
+                continue
+
+            part = dict(mime_part.items())
+            part['@body'] = mime_part.get_payload()
+
+            # convert payload into actual string data
+            if isinstance(part['@body'], list):
+                part['@body'] = (str(x) for x in part['@body'])
+
+            result['@parts'].append(part)
+    else:
+        result['@body'] = message.get_payload()
+
+    return result
+
+
+def imap_folder_path(uri):
+    """
+        Translate the folder name into a fully qualified folder path such as it
+        would be used by a cyrus administrator.
+    """
+    if isinstance(uri, str):
+        uri = parse_imap_uri(uri)
+
+    username = uri['user']
+    domain = uri['domain']
+    folder_name = uri['path']
+
+    # FIXME: prefix for shared folders; should be read from IMAP NAMESPACE command
+    shared_prefix = conf.get('imap', 'sharedprefix')
+
+    # Through filesystem
+    # To get the mailbox path, use:
+    # TODO: Assumption #1 is we are using virtual domains, and this domain does
+    # TODO: Assumption #2 is the mailbox in question is a user mailbox
+    # TODO: Assumption #3 is we use the unix hierarchy separator
+
+    # Translate the folder name in to a fully qualified folder path such as it
+    # would be used by a cyrus administrator.
+    #
+    # Other Users (covered, the netloc has the username the suffix is the
+    # original folder name).
+    if not username == None:
+        if folder_name == "INBOX":
+            folder_path = os.path.join('user', '%s@%s' % (username, domain))
+        else:
+            folder_path = os.path.join('user', username, '%s@%s' % (folder_name, domain))
+    # Shared Folders
+    else:
+        if folder_name.startswith(shared_prefix):
+            folder_path = folder_name[len(shared_prefix):]
+        else:
+            folder_path = folder_name
+
+    return folder_path
+
+
+def imap_mailbox_fs_path(uri):
+    """
+        Translate the given folder URI into a filesystem path where this mailbox is stored.
+    """
+    if isinstance(uri, str):
+        uri = parse_imap_uri(uri)
+
+    folder_name = uri['path']
+
+    # Translate the folder name in to a fully qualified folder path such as it
+    # would be used by a cyrus administrator.
+    folder_path = imap_folder_path(uri)
+
+    # Through filesystem
+    # To get the mailbox path, use:
+    # TODO: Assumption #1 is we are using virtual domains, and this domain does
+    # TODO: Assumption #2 is the mailbox in question is a user mailbox
+    # TODO: Assumption #3 is we use the unix hierarchy separator

bonnie-0.1.tar.gz/bonnie/worker/__init__.py Changed

@@ -67,19 +67,28 @@
 
         for _class in handlers.list_classes():
             __class = _class()
-            self.handler_modules[__class] = __class.register(callback=self.register_handler)
+            __class.register(callback=self.register_handler)
+            self.handler_modules[_class] = __class
 
         for _class in inputs.list_classes():
             __class = _class()
-            self.input_modules[__class] = __class.register(callback=self.register_input)
+            __class.register(callback=self.register_input)
+            self.input_modules[_class] = __class
 
+        output_modules = conf.get('worker', 'output_modules').split(',')
         for _class in outputs.list_classes():
-            __class = _class()
-            self.output_modules[__class] = __class.register(callback=self.register_output)
+            _output = _class()
+            if _output.name() in output_modules:
+                _output.register(callback=self.register_output)
+                self.output_modules[_class] = _output
 
+        storage_modules = conf.get('worker', 'storage_modules').split(',')
         for _class in storage.list_classes():
-            __class = _class()
-            self.storage_modules[__class] = __class.register(callback=self.register_storage)
+            _storage = _class()
+            if _storage.name() in storage_modules:
+                _storage.register(callback=self.register_storage)
+                self.storage_modules[_class] = _storage
+                self.storage = _storage
 
     def event_notification(self, notification):
         """
@@ -98,19 +107,42 @@
 
         if self.handler_interests.has_key(event):
             for interest in self.handler_interests[event]:
-                (notification, _jobs) = interest['callback'](notification=notification)
+                (notification, _jobs) = self.interest_callback(interest, notification)
                 jobs.extend(_jobs)
 
         for interest in self.handler_interests['_all']:
-            (notification, _jobs) = interest['callback'](notification=notification)
+            (notification, _jobs) = self.interest_callback(interest, notification)
             jobs.extend(_jobs)
 
+        # trigger storage modules which registered interest in particular notification properties
         if len(jobs) == 0:
-            for interest in self.output_interests['_all']:
-                (notification, _jobs) = interest['callback'](notification=notification)
-                jobs.extend(_jobs)
-
-        return notification, jobs
+            for prop,storage_interests in self.storage_interests.iteritems():
+                if notification.has_key(prop):
+                    for interest in storage_interests:
+                        (notification, _jobs) = self.interest_callback(interest, notification)
+                        jobs.extend(_jobs)
+
+        # finally send notification to output handlers if no jobs remaining
+        if len(jobs) == 0 and not notification.has_key('_suppress_output'):
+            if self.output_interests.has_key(event):
+                for interest in self.output_interests[event]:
+                    (notification, _jobs) = self.interest_callback(interest, notification)
+                    jobs.extend(_jobs)
+
+            if not notification.has_key('_suppress_output') and self.output_interests.has_key('_all'):
+                for interest in self.output_interests['_all']:
+                    (notification, _jobs) = self.interest_callback(interest, notification)
+                    jobs.extend(_jobs)
+
+        return notification, list(set(jobs))
+
+    def interest_callback(self, interest, notification):
+        """
+            Helper method to call an interest callback
+        """
+        kw = interest['kw'] if interest.has_key('kw') else {}
+        kw['notification'] = notification
+        return interest['callback'](**kw)
 
     def register_handler(self, interests={}):
         """
@@ -124,15 +156,18 @@
 
                 { 'MessageAppend': { 'callback': self.run } }
         """
-
+        print 'register_handler', interests
         for interest,how in interests.iteritems():
             if not self.handler_interests.has_key(interest):
                 self.handler_interests[interest] = []
 
             self.handler_interests[interest].append(how)
 
+        return self
+
     def register_input(self, interests):
         self.input_interests = interests
+        return self
 
     def register_output(self, interests):
         for interest,how in interests.iteritems():
@@ -141,12 +176,26 @@
 
             self.output_interests[interest].append(how)
 
+        return self
+
     def register_storage(self, interests):
-        self.storage_interests = interests
+        for interest,how in interests.iteritems():
+            if not self.storage_interests.has_key(interest):
+                self.storage_interests[interest] = []
+
+            self.storage_interests[interest].append(how)
+
+        return self
 
     def run(self):
-        input_modules = conf.get('worker', 'input_modules')
-        for _input in self.input_modules.keys():
-            if _input.name() == input_modules:
+        input_modules = conf.get('worker', 'input_modules').split(',')
+        for _input in self.input_modules.values():
+            if _input.name() in input_modules:
                 _input.run(callback=self.event_notification)
 
+    def terminate(self, *args, **kw):
+        for _input in self.input_modules.values():
+            if hasattr(_input, 'terminate'):
+                _input.terminate()
+            else:
+                _input.running = False
\ No newline at end of file

bonnie-0.1.tar.gz/bonnie/worker/handlers/__init__.py Changed

bonnie-0.1.tar.gz/bonnie/worker/handlers/base.py Changed

bonnie-0.1.tar.gz/bonnie/worker/handlers/changelog.py Added

@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Thomas Bruederli <bruederli at kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later version
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+#
+
+"""
+    Changelog handler for groupware objects
+"""
+import re
+import time
+import bonnie
+
+from dateutil.tz import tzutc
+from bonnie.worker.handlers import HandlerBase
+
+log = bonnie.getLogger('bonnie.worker.changelog')
+
+# timestamp (* 100) at the year 2010
+REVBASE = 94668480000
+
+class ChangelogHandler(HandlerBase):
+    events = ['MessageAppend','vnd.cmu.MessageMove']
+
+    def __init__(self, *args, **kw):
+        HandlerBase.__init__(self, *args, **kw)
+
+    def register(self, callback):
+        kw = { 'callback': self.run }
+        interests = dict((event,kw) for event in self.events)
+        self.worker = callback(interests)
+
+    def run(self, notification):
+        # message notifications require message headers
+        if not notification.has_key('messageHeaders'):
+            return (notification, [ b"FETCH" ] if notification['event'] == 'MessageAppend' else [ b"HEADER" ])
+
+        # check if this is a groupware object
+        object_type = None
+        msguid = notification['uidset'] if notification.has_key('uidset') else None
+        headers = notification['messageHeaders'][msguid] if notification['messageHeaders'].has_key(msguid) else None
+        if headers and headers.has_key('X-Kolab-Type') and headers.has_key('Subject'):
+            match = re.match(r"application/x-vnd.kolab.(\w+)", headers['X-Kolab-Type'])
+            if match:
+                object_type = match.group(1)
+
+        # assign a revision number based on the current time
+        if object_type is not None:
+            notification['revision'] = int(round(time.time() * 100 - REVBASE))
+            # TODO: save object type and UUID in separate fields?
+            # These are translated into headers.X-Kolab-Type and headers.Subject by the output module
+
+        log.debug("Object type %r detected in event %r" % (object_type, notification['event']), level=8)
+
+        # TODO: suppress MessageTrash/MessageExpunge events when followed by a MessageAppend for the same object
+
+        return (notification, [])
+

bonnie-0.1.tar.gz/bonnie/worker/handlers/logout.py Changed

@@ -21,6 +21,11 @@
     Base handler for an event notification of type 'Logout'
 """
 
+import time
+import datetime
+
+from dateutil.tz import tzutc
+from dateutil.parser import parse
 from bonnie.worker.handlers import HandlerBase
 
 class LogoutHandler(HandlerBase):
@@ -28,3 +33,48 @@
 
     def __init__(self, *args, **kw):
         HandlerBase.__init__(self, *args, **kw)
+
+    def run(self, notification):
+        # lookup corresponding Login event and update that record with logout_time
+        # and suppress separate logging of this event with notification['_suppress_output'] = True
+        if notification.has_key('vnd.cmu.sessionId'):
+            now = datetime.datetime.now(tzutc())
+            attempts = 5
+            while attempts > 0:
+                results = self.worker.storage.select(
+                    query=[
+                        ('event', '=', 'Login'),
+                        ('session_id', '=', notification['vnd.cmu.sessionId'])
+                    ],
+                    index='logstash-*',
+                    doctype='logs',
+                    fields='user,@timestamp',
+                    limit=1
+                )
+                if results['total'] > 0:
+                    login_event = results['hits'][0]
+
+                    try:
+                        timestamp = parse(login_event['@timestamp'])
+                    except:
+                        timestamp = now
+
+                    delta = now - timestamp
+
+                    # update Login event record
+                    self.worker.storage.set(
+                        key=login_event['_id'],
+                        index=login_event['_index'],
+                        doctype=login_event['_doctype'],
+                        value={
+                            'logout_time': datetime.datetime.strftime(now, "%Y-%m-%dT%H:%M:%S.%fZ"),
+                            'duration': delta.days * 24 * 3600 + delta.seconds + delta.microseconds / 1e6
+                        }
+                    )
+                    notification['_suppress_output'] = True
+                    return (notification, [])
+
+                attempts -= 1
+                time.sleep(1) # wait for storage and try again
+
+        return super(LogoutHandler, self).run(notification)
\ No newline at end of file

bonnie-0.1.tar.gz/bonnie/worker/handlers/mailboxcreate.py Changed

@@ -21,6 +21,7 @@
     Base handler for an event notification of type 'MailboxCreate'
 """
 
+import bonnie
 from bonnie.worker.handlers import HandlerBase
 
 class MailboxCreateHandler(HandlerBase):
@@ -28,3 +29,20 @@
 
     def __init__(self, *args, **kw):
         HandlerBase.__init__(self, *args, **kw)
+        self.log = bonnie.getLogger('bonnie.worker.' + self.event)
+
+    def run(self, notification):
+        # call super for some basic notification processing
+        (notification, jobs) = super(MailboxCreateHandler, self).run(notification)
+
+        # mailbox notifications require metadata
+        if not notification.has_key('metadata'):
+            self.log.debug("Adding GETMETADATA job for " + self.event, level=8)
+            jobs.append(b"GETMETADATA")
+            return (notification, jobs)
+
+        # extract uniqueid from metadata -> triggers the storage module
+        if notification['metadata'].has_key('/shared/vendor/cmu/cyrus-imapd/uniqueid'):
+            notification['folder_uniqueid'] = notification['metadata']['/shared/vendor/cmu/cyrus-imapd/uniqueid']
+
+        return (notification, jobs)

bonnie-0.1.tar.gz/bonnie/worker/handlers/mailboxrename.py Changed

bonnie-0.1.tar.gz/bonnie/worker/handlers/messageappend.py Changed

@@ -21,27 +21,21 @@
     Base handler for an event notification of type 'MessageAppend'
 """
 
-from bonnie.worker.handlers import HandlerBase
+from bonnie.worker.handlers import MessageHandlerBase
 
-class MessageAppendHandler(HandlerBase):
+class MessageAppendHandler(MessageHandlerBase):
     event = 'MessageAppend'
 
     def __init__(self, *args, **kw):
-        HandlerBase.__init__(self, *args, **kw)
-
-    def register(self, callback):
-        interests = {
-                self.event: {
-                        'callback': self.run,
-                    }
-            }
-
-        callback(interests)
+        MessageHandlerBase.__init__(self, *args, **kw)
 
     def run(self, notification):
-        if notification.has_key('messageContent') and not notification['messageContent'] in [None, ""]:
-            return (notification, [])
+        # call super for some basic notification processing
+        (notification, jobs) = super(MessageHandlerBase, self).run(notification)
+
+        if not notification.has_key('messageContent') or notification['messageContent'] in [None, ""]:
+            self.log.debug("Adding FETCH job for " + self.event, level=8)
+            return (notification, [ b"FETCH" ])
 
-        print "adding a job for messageappend"
-        jobs = [ b"FETCH" ]
         return (notification, jobs)
+

bonnie-0.1.tar.gz/bonnie/worker/handlers/messagebase.py Added

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Thomas Bruederli <bruederli at kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later version
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+#
+
+"""
+    Base handler for an event notification of type 'Message*'
+"""
+
+import bonnie
+from bonnie.worker.handlers import HandlerBase
+
+class MessageHandlerBase(HandlerBase):
+    event = None
+
+    def __init__(self, *args, **kw):
+        HandlerBase.__init__(self, *args, **kw)
+        self.log = bonnie.getLogger('bonnie.worker.' + self.event)
+
+    def run(self, notification):
+        # call super for some basic notification processing
+        (notification, jobs) = super(MessageHandlerBase, self).run(notification)
+
+        # message notifications require message headers
+        if not notification.has_key('messageHeaders'):
+            self.log.debug("Adding HEADER job for " + self.event, level=8)
+            jobs.append(b"HEADER")
+            return (notification, jobs)
+
+        return (notification, jobs)
+

bonnie-0.1.tar.gz/bonnie/worker/handlers/messagecopy.py Changed

bonnie-0.1.tar.gz/bonnie/worker/handlers/messagemove.py Changed

bonnie-0.1.tar.gz/bonnie/worker/handlers/messagenew.py Changed

bonnie-0.1.tar.gz/bonnie/worker/handlers/messagetrash.py Changed

bonnie-0.1.tar.gz/bonnie/worker/inputs/zmq_input.py Changed

@@ -33,9 +33,11 @@
 
 import bonnie
 conf = bonnie.getConf()
-log = bonnie.getLogger('worker.ZMQInput')
+log = bonnie.getLogger('bonnie.worker.ZMQInput')
 
 class ZMQInput(object):
+    running = False
+
     def __init__(self, *args, **kw):
         self.identity = u"Worker-%s-%d" % (socket.getfqdn(),os.getpid())
 
@@ -81,10 +83,15 @@
     def run(self, callback=None):
         log.info("[%s] starting", self.identity)
 
+        self.running = True
         self.report_state()
 
-        while True:
-            sockets = dict(self.poller.poll(1000))
+        while self.running:
+            try:
+                sockets = dict(self.poller.poll(1000))
+            except Exception, e:
+                log.error("zmq.Poller error: %r", e)
+                sockets = dict()
 
             if self.report_timestamp < (time.time() - 60):
                 self.report_state()
@@ -126,6 +133,7 @@
 
                         self.set_state_ready()
 
+        log.info("[%s] shutting down", self.identity)
         self.worker.close()
 
     def set_state_busy(self):

bonnie-0.1.tar.gz/bonnie/worker/outputs/elasticsearch_output.py Changed

@@ -23,8 +23,11 @@
                 host=elasticsearch_output_address
             )
 
+    def name(self):
+        return 'elasticsearch_output'
+
     def register(self, callback):
-        callback({'_all': { 'callback': self.run }})
+        self.worker = callback({'_all': { 'callback': self.run }})
 
     def notification2log(self, notification):
         """
@@ -40,11 +43,15 @@
             'aclSubject':   'acl_subject',
             'mailboxID':    'mailbox_id',
             'messageSize':  'message_size',
-            'messageHeaders': 'message_headers',
-            'messageContent': 'message',
+            'messageHeaders': None,
+            'messageContent': None,
+            'bodyStructure':  None,
+            'metadata':       None,
+            'acl':            None,
             'flagNames':    'flag_names',
             'diskUsed':     'disk_used',
-            'vnd.cmu.oldUidset': 'uidset',
+            'vnd.cmu.oldUidset': 'olduidset',
+            'vnd.cmu.sessionId': 'session_id',
         }
         log = { '@version': bonnie.API_VERSION }
         for key,val in notification.iteritems():
@@ -68,10 +75,27 @@
             timestamp = datetime.datetime.now(tzutc())
 
         notification['@timestamp'] = datetime.datetime.strftime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ")
+        index = 'logstash-%s' % (datetime.datetime.strftime(timestamp, "%Y-%m-%d"))
+
+        # for notifications concerning multiple messages, create entries for each message
+        if notification.has_key('messageHeaders') and isinstance(notification['messageHeaders'], dict) and len(notification['messageHeaders']) > 0:
+            for uid,headers in notification['messageHeaders'].iteritems():
+                notification['uidset']  = uid
+                notification['headers'] = headers
+                notification['message'] = None
+                if notification.has_key('messageContent') and notification['messageContent'].has_key(uid):
+                    notification['message'] = notification['messageContent'][uid]
 
-        self.es.create(
-                index='logstash-%s' % (datetime.datetime.strftime(timestamp, "%Y-%m-%d")),
+                self.es.create(
+                    index=index,
+                    doc_type='logs',
+                    body=self.notification2log(notification)
+                )
+        else:
+            self.es.create(
+                index=index,
                 doc_type='logs',
                 body=self.notification2log(notification)
             )
+
         return (notification, [])

bonnie-0.1.tar.gz/bonnie/worker/storage/elasticsearch_storage.py Changed

@@ -1,8 +1,387 @@
-__package__ = 'bonnie.worker.storage'
+# -*- coding: utf-8 -*-
+#
+# Copyright 2010-2014 Kolab Systems AG (http://www.kolabsys.com)
+#
+# Jeroen van Meeuwen (Kolab Systems) <vanmeeuwen a kolabsys.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3 or, at your option, any later
+# version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
+# USA.
+#
+
+import re
+import json
+import urllib
+import hashlib
+import datetime
+import elasticsearch
+
+from dateutil.tz import tzutc
+from bonnie.utils import parse_imap_uri
+
+import bonnie
+conf = bonnie.getConf()
+log = bonnie.getLogger('bonnie.worker.ElasticSearchStorage')
 
 class ElasticSearchStorage(object):
+    """
+        Storage node writing object data into Elasticsearch
+    """
+    default_index = 'objects'
+    default_doctype = 'object'
+    folders_index = 'objects'
+    folders_doctype = 'folder'
+
     def __init__(self, *args, **kw):
-        pass
+        elasticsearch_output_address = conf.get('worker', 'elasticsearch_storage_address')
+
+        if elasticsearch_output_address == None:
+            elasticsearch_output_address = 'localhost'
+
+        self.es = elasticsearch.Elasticsearch(
+            host=elasticsearch_output_address
+        )
+
+    def name(self):
+        return 'elasticsearch_storage'
+
+    def register(self, callback, **kw):
+        if callback is not None:
+            self.worker = callback(interests={
+                'uidset': { 'callback': self.resolve_folder_uri },
+                'folder_uniqueid': { 'callback': self.resolve_folder_uri },
+                'mailboxID': { 'callback': self.resolve_folder_uri, 'kw': { 'attrib': 'mailboxID' } }
+            })
+
+    def get(self, key, index=None, doctype=None, fields=None, **kw):
+        """
+            Standard API for accessing key/value storage
+        """
+        _index = index or self.default_index
+        _doctype = doctype or self.default_doctype
+        try:
+            res = self.es.get(
+                index=_index,
+                doc_type=_doctype,
+                id=key,
+                _source_include=fields or '*'
+            )
+            log.debug("ES get result for %s/%s/%s: %r" % (_index, _doctype, key, res), level=8)
+
+            if res['found']:
+                result = self._transform_result(res)
+            else:
+                result = None
+
+        except elasticsearch.exceptions.NotFoundError, e:
+            log.debug("ES entry not found for %s/%s/%s: %r" % (_index, _doctype, key, e))
+            result = None
+
+        except Exception, e:
+            log.warning("ES get exception: %r", e)
+            result = None
+
+        return result
+
+
+    def set(self, key, value, index=None, doctype=None, **kw):
+        """
+            Standard API for writing to key/value storage
+        """
+        _index = index or self.default_index
+        _doctype = doctype or self.default_doctype
+        try:
+            existing = self.es.get(
+                index=_index,
+                doc_type=_doctype,
+                id=key,
+                fields=None
+            )
+            log.debug("ES get result for %s/%s/%s: %r" % (_index, _doctype, key, existing), level=8)
+
+        except elasticsearch.exceptions.NotFoundError, e:
+            existing = None
+
+        except Exception, e:
+            log.warning("ES get exception: %r", e)
+            existing = None
+
+        if existing is None:
+            try:
+                ret = self.es.create(
+                    index=_index,
+                    doc_type=_doctype,
+                    id=key,
+                    body=value,
+                    consistency='one',
+                    replication='async'
+                )
+                log.debug("Created ES object for %s/%s/%s: %r" % (_index, _doctype, key, ret), level=8)
+
+            except Exception, e:
+                log.warning("ES create exception: %r", e)
+                ret = None
+        else:
+            try:
+                ret = self.es.update(
+                    index=_index,
+                    doc_type=_doctype,
+                    id=key,
+                    body={ 'doc': value },
+                    consistency='one',
+                    replication='async'
+                )
+                log.debug("Updated ES object for %s/%s/%s: %r" % (_index, _doctype, key, ret), level=8)
+
+            except Exception, e:
+                log.warning("ES update exception: %r", e)
+
+        return ret
+
+
+
+    def select(self, query, index=None, doctype=None, fields=None, sortby=None, limit=None, **kw):
+        """
+            Standard API for querying storage
+
+            @param query:   List of query parameters, each represented as a triplet of (<field> <op> <value>).
+                            combined to an AND list of search criterias. <value> can either be
+                             - a string for direct comparison
+                             - a list for "in" comparisons
+                             - a tuple with two values for range queries
+            @param index:   Index name (i.e. database name)
+            @param doctype: Document type (i.e. table name)
+            @param fields:  List of fields to retrieve (string, comma-separated)
+            @param sortby:  Fields to be used fort sorting the results (string, comma-separated)
+            @param limit:   Number of records to return
+        """
+        result = None
+        args = dict(index=index or self.default_index, doc_type=doctype or self.default_doctype, _source_include=fields or '*')
+
+        if isinstance(query, dict):
+            args['body'] = query
+        elif isinstance(query, list):
+            args['q'] = self._build_query(query)
+        else:
+            args['q'] = query
+
+        if sortby is not None:
+            args['sort'] = sortby
+        if limit is not None:
+            args['size'] = int(limit)
+
+        try:
+            res = self.es.search(**args)
+            log.debug("ES select result for %r: %r" % (args['q'] or args['body'], res), level=8)
+
+        except elasticsearch.exceptions.NotFoundError, e:
+            log.debug("ES entry not found for key %s: %r", key, e)
+            res = None
+
+        except Exception, e:
+            log.warning("ES get exception: %r", e)
+            res = None
+
+        if res is not None and res.has_key('hits'):
+            result = dict(total=res['hits']['total'])

bonnie-0.1.tar.gz/broker.py Changed

bonnie-0.1.tar.gz/collector.py Changed

bonnie-0.1.tar.gz/conf/bonnie.conf Changed

bonnie-0.1.tar.gz/tests Added

bonnie-0.1.tar.gz/tests/__init__.py Added

bonnie-0.1.tar.gz/tests/unit Added

bonnie-0.1.tar.gz/tests/unit/__init__.py Added

bonnie-0.1.tar.gz/tests/unit/resources Added

bonnie-0.1.tar.gz/tests/unit/resources/3. Added

bonnie-0.1.tar.gz/tests/unit/resources/event_mime_message.eml Added

@@ -0,0 +1,95 @@
+MIME-Version: 1.0
+Content-Type: multipart/mixed;
+ boundary="=_ddac4015cd2f0b6b232ed6f53d8bd55b"
+From: john.doe@example.org
+To: john.doe@example.org
+Date: Wed, 24 Sep 2014 06:52:00 +0200
+X-Kolab-Type: application/x-vnd.kolab.event
+X-Kolab-Mime-Version: 3.0
+Subject: 253E800C973E9FB99D174669001DB19B-FCBB6C4091F28CA0
+User-Agent: Kolab 3.1/Roundcube 1.1-git
+
+--=_ddac4015cd2f0b6b232ed6f53d8bd55b
+Content-Transfer-Encoding: quoted-printable
+Content-Type: text/plain; charset=ISO-8859-1
+
+This is a Kolab Groupware object. To view this object you will need an emai=
+l client that understands the Kolab Groupware format. For a list of such em=
+ail clients please visit http://www.kolab.org/
+
+
+--=_ddac4015cd2f0b6b232ed6f53d8bd55b
+Content-Transfer-Encoding: 8bit
+Content-Type: application/calendar+xml; charset=UTF-8; name=kolab.xml
+Content-Disposition: attachment;
+ filename=kolab.xml;
+ size=1864
+
+<?xml version="1.0" encoding="UTF-8" standalone="no" ?>
+<icalendar xmlns="urn:ietf:params:xml:ns:icalendar-2.0">
+
+  <vcalendar>
+    <properties>
+      <prodid>
+        <text>Roundcube-libkolab-1.1 Libkolabxml-1.1</text>
+      </prodid>
+      <version>
+        <text>2.0</text>
+      </version>
+      <x-kolab-version>
+        <text>3.1.0</text>
+      </x-kolab-version>
+    </properties>
+    <components>
+      <vevent>
+        <properties>
+          <uid>
+            <text>253E800C973E9FB99D174669001DB19B-FCBB6C4091F28CA0</text>
+          </uid>
+          <created>
+            <date-time>2014-09-23T23:31:23Z</date-time>
+          </created>
+          <dtstamp>
+            <date-time>2014-09-24T04:52:00Z</date-time>
+          </dtstamp>
+          <sequence>
+            <integer>3</integer>
+          </sequence>
+          <class>
+            <text>PUBLIC</text>
+          </class>
+          <dtstart>
+            <parameters>
+              <tzid>
+                <text>/kolab.org/Europe/Berlin</text>
+              </tzid>
+            </parameters>
+            <date-time>2014-09-26T14:00:00</date-time>
+          </dtstart>
+          <dtend>
+            <parameters>
+              <tzid>
+                <text>/kolab.org/Europe/Berlin</text>
+              </tzid>
+            </parameters>
+            <date-time>2014-09-26T15:00:00</date-time>
+          </dtend>
+          <summary>
+            <text>Test</text>
+          </summary>
+          <organizer>
+            <parameters>
+              <cn>
+                <text>Doe, John</text>
+              </cn>
+            </parameters>
+            <cal-address>mailto:%3Cjohn.doe%40example.org%3E</cal-address>
+          </organizer>
+        </properties>
+      </vevent>
+    </components>
+  </vcalendar>
+
+</icalendar>
+
+--=_ddac4015cd2f0b6b232ed6f53d8bd55b--
\ No newline at end of file

bonnie-0.1.tar.gz/tests/unit/test-001-utils.py Added

@@ -0,0 +1,78 @@
+import os
+import json
+from bonnie.utils import expand_uidset
+from bonnie.utils import parse_imap_uri
+from bonnie.utils import mail_message2dict
+from bonnie.utils import decode_message_headers
+from bonnie.utils import imap_folder_path
+from bonnie.utils import imap_mailbox_fs_path
+from email import message_from_string
+from twisted.trial import unittest
+
+
+class TestBonnieUtils(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def _get_resource(self, filename):
+        pwd = os.path.dirname(__file__)
+        filepath = os.path.join(pwd, 'resources', filename)
+        fp = open(filepath, 'r')
+        data = fp.read()
+        fp.close()
+        return data
+
+    def test_expand_uidset(self):
+        self.assertEqual(expand_uidset('3'), ['3'])
+        self.assertEqual(expand_uidset('3,5'), ['3','5'])
+        self.assertEqual(expand_uidset('3:5'), ['3','4','5'])
+
+    def test_parse_imap_uri(self):
+        url = parse_imap_uri("imap://john.doe@example.org@kolab33.example.org/Calendar/Personal%20Calendar;UIDVALIDITY=1411487702/;UID=3")
+        self.assertEqual(url['host'],   'kolab33.example.org')
+        self.assertEqual(url['user'],   'john.doe')
+        self.assertEqual(url['domain'], 'example.org')
+        self.assertEqual(url['path'],   'Calendar/Personal Calendar')
+        self.assertEqual(url['UID'],    '3')
+
+    def test_decode_message_headers(self):
+        message = message_from_string(self._get_resource('3.'))
+        headers = decode_message_headers(message)
+
+        self.assertEqual(len(headers['From']), 1)
+        self.assertEqual(len(headers['To']), 2)
+        self.assertEqual(headers['To'][0], u'Br\u00fcderli, Thomas <thomas.bruederli@example.org>')
+        self.assertEqual(headers['Content-Type'], 'text/plain')
+        self.assertEqual(headers['Date'], '2014-09-24T04:52:00Z')
+        self.assertEqual(headers['Subject'], 'Test')
+
+    def test_mail_message2dict(self):
+        message = mail_message2dict(self._get_resource('event_mime_message.eml'))
+
+        self.assertIsInstance(message, dict)
+        self.assertEqual(message['Subject'], '253E800C973E9FB99D174669001DB19B-FCBB6C4091F28CA0')
+        self.assertEqual(message['X-Kolab-Type'], 'application/x-vnd.kolab.event')
+        self.assertEqual(len(message['@parts']), 2)
+
+        xmlpart = message['@parts'][1]
+        self.assertEqual(xmlpart['Content-Type'], 'application/calendar+xml; charset=UTF-8; name=kolab.xml')
+
+        message2 = mail_message2dict("FOO")
+        self.assertIsInstance(message2, dict)
+        self.assertEqual(message2['@body'], "FOO")
+
+    def test_imap_folder_path(self):
+        p1 = imap_folder_path("imap://john.doe@example.org@kolab.example.org/Calendar;UID=3")
+        self.assertEqual(p1, "user/john.doe/Calendar@example.org")
+
+        p2 = imap_folder_path("imap://john.doe@example.org@kolab.example.org/INBOX;UIDVALIDITY=1411487702")
+        self.assertEqual(p2, "user/john.doe@example.org")
+
+        # test shared folders (but how are they referred in the uri?)
+        p3 = imap_folder_path("imap://kolab33.example.org/Shared%20Folders/shared/Project-X%40example.org;UIDVALIDITY=1412093781/;UID=2")
+        self.assertEqual(p3, "shared/Project-X@example.org")
+
+    def test_imap_mailbox_fs_path(self):
+        path = imap_mailbox_fs_path("imap://john.doe@example.org@kolab.example.org/Calendar/Personal%20Calendar;UID=3")
+        self.assertEqual(path, "/var/spool/imap/domain/e/example.org/j/user/john^doe/Calendar/Personal Calendar")

bonnie-0.1.tar.gz/tests/unit/test-002-collector.py Added

@@ -0,0 +1,84 @@
+import os
+import json
+import bonnie
+from bonnie.collector import BonnieCollector
+from bonnie.collector.handlers import MessageDataHandler, IMAPDataHandler
+from twisted.trial import unittest
+
+
+class TestBonnieCollector(unittest.TestCase):
+
+    def setUp(self):
+        # patch bonnie.utils.imap_mailbox_fs_path() to point to local resources folder
+        self.patch(bonnie.utils, 'imap_mailbox_fs_path', self._imap_mailbox_fs_path)
+
+    def _imap_mailbox_fs_path(self, uri):
+        pwd = os.path.dirname(__file__)
+        return os.path.join(pwd, 'resources')
+
+    def test_retrieve_headers_for_messages(self):
+        coll = MessageDataHandler()
+        notification = { 'uri': 'imap://john.doe@example.org@kolab.example.org/Calendar', 'uidset': '3' }
+        notification = json.loads(coll.retrieve_headers_for_messages(json.dumps(notification)))
+
+        self.assertTrue(notification.has_key('messageHeaders'))
+        self.assertIsInstance(notification['messageHeaders'], dict)
+        self.assertTrue(notification['messageHeaders'].has_key('3'))
+
+        headers = notification['messageHeaders']['3']
+        self.assertIsInstance(headers, dict)
+        self.assertEqual(headers['Subject'], 'Test')
+        self.assertEqual(headers['From'][0], 'John Doe <john.doe@example.org>')
+
+    def test_retrieve_contents_for_messages(self):
+        coll = MessageDataHandler()
+        notification = { 'uri': 'imap://john.doe@example.org@kolab.example.org/Calendar', 'uidset': '3' }
+        notification = json.loads(coll.retrieve_contents_for_messages(json.dumps(notification)))
+
+        self.assertTrue(notification.has_key('messageContent'))
+        self.assertIsInstance(notification['messageContent'], dict)
+        self.assertTrue(notification['messageContent'].has_key('3'))
+        self.assertIsInstance(notification['messageContent']['3'], unicode)
+
+        self.assertTrue(notification.has_key('messageHeaders'))
+        self.assertIsInstance(notification['messageHeaders'], dict)
+        self.assertTrue(notification['messageHeaders'].has_key('3'))
+        self.assertIsInstance(notification['messageHeaders']['3'], dict)
+
+    def test_get_imap_folder_metadata(self):
+        coll = IMAPDataHandler()
+        notification = { 'uri': 'imap://john.doe@example.org@kolab.example.org/Calendar' }
+        notification = json.loads(coll.get_imap_folder_metadata(json.dumps(notification)))
+
+        self.assertTrue(notification.has_key('metadata'))
+        self.assertIsInstance(notification['metadata'], dict)
+        self.assertEqual(notification['metadata']['/shared/vendor/kolab/folder-type'], 'event')
+
+    def test_get_imap_folder_acl(self):
+        coll = IMAPDataHandler()
+        notification = { 'uri': 'imap://john.doe@example.org@kolab.example.org/Calendar' }
+        notification = json.loads(coll.get_imap_folder_acl(json.dumps(notification)))
+
+        self.assertTrue(notification.has_key('acl'))
+        self.assertIsInstance(notification['acl'], dict)
+        self.assertEqual(notification['acl']['john.doe@example.org'], 'lrswipkxtecdan')
+
+    def test_zz_execute(self):
+        coll = BonnieCollector()
+
+        notification = { 'uri': 'imap://john.doe@example.org@kolab.example.org/Calendar;UID=3' }
+        result = json.loads(coll.execute('FETCH', json.dumps(notification)))
+
+        self.assertTrue(result.has_key('messageContent'))
+        self.assertTrue(result.has_key('uidset'))
+        self.assertEqual(result['uidset'], '3')
+
+        notification['vnd.cmu.oldUidset'] = '3'
+        result = json.loads(coll.execute('HEADER', json.dumps(notification)))
+        self.assertTrue(result.has_key('messageHeaders'))
+
+        result = json.loads(coll.execute('GETMETADATA', json.dumps(notification)))
+        self.assertTrue(result.has_key('metadata'))
+
+        result = json.loads(coll.execute('GETACL', json.dumps(notification)))
+        self.assertTrue(result.has_key('acl'))

bonnie-0.1.tar.gz/tests/unit/test-003-dealer.py Added

@@ -0,0 +1,48 @@
+import os
+import json
+import bonnie
+from bonnie.dealer import BonnieDealer
+from twisted.trial import unittest
+
+class BonnieDealerOutput(object):
+    """
+        Mockup class to catch event notifications forwarded by BonnieDealer
+    """
+    output_calls = []
+
+    def __init__(self, *args, **kw):
+        pass
+
+    def name(self):
+        return 'zmq_output'
+
+    def register(self, *args, **kw):
+        pass
+
+    def run(self, notification):
+        notification = json.loads(notification)
+        self.output_calls.append(notification['event'])
+
+class TestBonnieDealer(unittest.TestCase):
+
+    def setUp(self):
+        # patch bonnie.outputs.list_classes() to register a dummy output handler
+        self.patch(bonnie.dealer.outputs, 'list_classes', self._outputs_list_classes)
+
+    def _outputs_list_classes(self):
+        return [ BonnieDealerOutput ]
+
+    def test_run(self):
+        dealer = BonnieDealer()
+        dealer.run(json.dumps({ 'event':'Login', 'user':'john.doe@example.org' }))
+        self.assertIn('Login', BonnieDealerOutput.output_calls)
+
+        # don't trigger blacklisted events
+        BonnieDealerOutput.output_calls = []
+        dealer.run(json.dumps({ 'event':'Login', 'user':'cyrus-admin' }))
+        self.assertEqual(len(BonnieDealerOutput.output_calls), 0)
+
+        dealer.run(json.dumps({ 'event':'MessageAppend', 'user':'cyrus-admin' }))
+        self.assertEqual(len(BonnieDealerOutput.output_calls), 1)
+        self.assertIn('MessageAppend', BonnieDealerOutput.output_calls)
+        
\ No newline at end of file

bonnie-0.1.tar.gz/tests/unit/test-004-worker.py Added

@@ -0,0 +1,78 @@
+import json
+import elasticsearch
+import bonnie
+
+from bonnie.worker.storage import ElasticSearchStorage
+from twisted.trial import unittest
+
+class MockupElasticsearch(object):
+    """
+        Dummy mockup to avoid real connections to ES
+    """
+
+    def __init__(self, *args, **kw):
+        pass
+
+    def search(self, *args, **kw):
+        pass
+
+    def create(self, *args, **kw):
+        pass
+
+    def update(self, *args, **kw):
+        pass
+
+
+class TestBonnieWorker(unittest.TestCase):
+
+    def setUp(self):
+        self.patch(elasticsearch, 'Elasticsearch', MockupElasticsearch)
+
+    def test_001_notificaton2folder(self):
+        notification = {
+            'event': 'MailboxCreate',
+            'uri': 'imap://john.doe@example.org@kolab.example.org/Calendar/Subcal;UIDVALIDITY=12345',
+            'metadata': {
+                '/shared/vendor/cmu/cyrus-imapd/partition':          'default',
+                '/shared/vendor/cmu/cyrus-imapd/lastupdate':         '23-Sep-2014 11:55:15 -0400',
+                '/shared/vendor/cmu/cyrus-imapd/duplicatedeliver':   'false',
+                '/shared/vendor/cmu/cyrus-imapd/pop3newuidl':        'true',
+                '/shared/vendor/cmu/cyrus-imapd/size':               '0',
+                '/shared/vendor/cmu/cyrus-imapd/sharedseen':         'false',
+                '/shared/vendor/kolab/folder-type':                  'event',
+                '/shared/vendor/cmu/cyrus-imapd/uniqueid':           '28614de9-3614-422d-be5d-5716605ef0fc',
+            },
+            'acl': {
+                'john.doe@example.org': "lrswipkxtecdan",
+                'anyone': 'lrs',
+            }
+        }
+        storage = ElasticSearchStorage()
+        folder = storage.notificaton2folder(notification)
+        folder_id = folder['id']
+
+        self.assertTrue(folder.has_key('body'))
+        self.assertEqual(folder['body']['uniqueid'], notification['folder_uniqueid'])
+        self.assertEqual(folder['body']['server'], 'kolab.example.org')
+        self.assertEqual(folder['body']['owner'], 'john.doe@example.org')
+        self.assertEqual(folder['body']['uri'], 'imap://john.doe@example.org@kolab.example.org/Calendar/Subcal')
+
+        # check changes in ignored metadata
+        notification['metadata']['/shared/vendor/cmu/cyrus-imapd/lastupdate'] = '24-Sep-2014 16:21:19 -0400'
+        notification['metadata']['/shared/vendor/cmu/cyrus-imapd/size'] = '88'
+        notification['acl'] = {
+            'anyone': 'lrs',
+            'john.doe@example.org': "lrswipkxtecdan",
+        }
+        notification['uri'] = 'imap://john.doe@example.org@kolab.example.org/Calendar/RENAMED;UIDVALIDITY=123456'
+        folder = storage.notificaton2folder(notification)
+        self.assertEqual(folder['id'], folder_id)
+
+        # detect ACL changes
+        notification['acl'] = {
+            'john.doe@example.org': "lrswipkxtecdan",
+            'anyone': 'lrswd',
+        }
+        folder = storage.notificaton2folder(notification)
+        self.assertNotEqual(folder['id'], folder_id)
+        
\ No newline at end of file

bonnie-0.1.tar.gz/tests/unit/test-005-persistence.py Added

@@ -0,0 +1,94 @@
+import os
+import json
+import bonnie
+from bonnie.broker import persistence
+from bonnie.broker.persistence import db, PersistentBase
+from bonnie.broker.brokers.zmq_broker.job import Job
+from twisted.trial import unittest
+
+class PlistItem(PersistentBase):
+    __tablename__ = 'plisttest'
+    id = db.Column(db.Integer, primary_key=True)
+    value = db.Column(db.String)
+
+    def __init__(self, value, id=None):
+        self.value = value
+        self.id = id
+
+    def __repr__(self):
+        return '<PlistItem %s:%r>' % (self.id, self.value)
+
+
+class TestBonniePersistence(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        # TODO: clear database
+        # PersistentBase.metadata.drop_all(bonnie.broker.persistence.engine)
+        pass
+
+    def test_001_base_list(self):
+        plist = persistence.List('base', PlistItem)
+        plist.append(PlistItem("One"))
+        plist.append(PlistItem("Two"))
+        item3 = PlistItem("Three")
+        plist.append(item3)
+
+        self.assertEqual(len(plist), 3)
+        self.assertEqual(plist[2], item3)
+        self.assertEqual(2, plist.index(item3))
+        self.assertTrue(item3 in plist)
+
+        plist.append(PlistItem("Five"))
+        plist.append(PlistItem("Six"))
+
+        plist[4] = PlistItem("Five.5")
+        self.assertEqual(plist[4].value, "Five.5")
+
+        del plist[plist.index(item3)]
+        plist.pop(3)
+        plist.pop()
+        self.assertEqual(len(plist), 2)
+
+        i = 0
+        for item in plist:
+            i += 1
+            self.assertTrue(isinstance(item, PlistItem))
+
+        self.assertEqual(i, 2)
+
+
+    def test_003_broker_jobs(self):
+        worker_jobs = persistence.List('worker', Job)
+        collector_jobs = persistence.List('collector', Job)
+        one = Job(state='PENDING', notification='{"state":"pending","event":"test"}', collector_id='C.1')
+        two = Job(state='PENDING', notification='{"state":"pending","event":"other"}', collector_id='C.1')
+        done = Job(state='DONE', notification='{"state":"done","event":"passed"}', collector_id='C.1')
+        worker_jobs.append(one)
+        worker_jobs.append(two)
+        worker_jobs.append(done)
+
+        self.assertEqual(len(worker_jobs), 3)
+        self.assertEqual(len(collector_jobs), 0)
+        self.assertTrue(one.uuid in [x.uuid for x in worker_jobs])
+        pending = [x for x in worker_jobs if x.state == 'PENDING' and x.collector_id == 'C.1']
+        self.assertEqual(len(pending), 2)
+        job = pending.pop()
+
+        self.assertEqual(job, two)
+        self.assertEqual(len(pending), 1)
+        self.assertEqual(len(worker_jobs), 3)
+
+        # move job to another list
+        collector_jobs.append(one)
+        self.assertTrue(one in collector_jobs)
+        self.assertTrue(one in worker_jobs)
+
+        worker_jobs.remove(one)
+        self.assertFalse(one in worker_jobs)
+
+        self.assertEqual(len(worker_jobs), 2)
+        self.assertEqual(len(collector_jobs), 1)
+        
\ No newline at end of file

bonnie-0.1.tar.gz/worker.py Changed

Changes of Revision 20