def read_data_process(self, data_process_id=""):
     # Read DataProcess object with _id matching  id
     log.debug("Reading DataProcess object id: %s" % data_process_id)
     data_proc_obj = self.clients.resource_registry.read(data_process_id)
     if not data_proc_obj:
         raise NotFound("DataProcess %s does not exist" % data_process_id)
     return data_proc_obj

    def on_start(self):
        log.debug('EventAlertTransform.on_start()')
        super(EventAlertTransform, self).on_start()

        #- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # get the algorithm to use
        #- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

        self.timer_origin = self.CFG.get_safe('process.timer_origin', 'Interval Timer')
        self.instrument_origin = self.CFG.get_safe('process.instrument_origin', '')

        self.counter = 0
        self.event_times = []

        #-------------------------------------------------------------------------------------
        # Set up a listener for instrument events
        #-------------------------------------------------------------------------------------

        self.instrument_event_queue = gevent.queue.Queue()

        def instrument_event_received(message, headers):
            log.debug("EventAlertTransform received an instrument event here::: %s" % message)
            self.instrument_event_queue.put(message)

        self.instrument_event_subscriber = EventSubscriber(origin = self.instrument_origin,
            callback=instrument_event_received)

        self.instrument_event_subscriber.start()

        #-------------------------------------------------------------------------------------
        # Create the publisher that will publish the Alert message
        #-------------------------------------------------------------------------------------

        self.event_publisher = EventPublisher()

    def _link_resources_lowlevel(self, subject_id='', association_type='', object_id='', check_duplicates=True):
        """
        create an association
        @param subject_id the resource ID of the predefined type
        @param association_type the predicate
        @param object_id the resource ID of the type to be joined
        @param check_duplicates whether to check for an existing association of this exact subj-pred-obj
        @todo check for errors: does RR check for bogus ids?
        """

        assert(type("") == type(subject_id) == type(object_id))

        if check_duplicates:
            dups = self._resource_link_exists(subject_id, association_type, object_id)
            if dups:
                log.debug("Create %s Association from '%s': ALREADY EXISTS"
                          % (self._assn_name(association_type),
                             self._toplevel_call()))
                return dups

        associate_success = self.RR.create_association(subject_id,
                                                       association_type,
                                                       object_id)

        log.debug("Create %s Association from '%s': %s"
                  % (self._assn_name(association_type),
                     self._toplevel_call(),
                      str(associate_success)))

        return associate_success

    def find_events(self, origin='', type='', min_datetime=0, max_datetime=0, limit= -1, descending=False):
        """
        This method leverages couchdb view and simple filters. It does not use elastic search.

        Returns a list of events that match the specified search criteria. Will throw a not NotFound exception
        if no events exist for the given parameters.

        @param origin         str
        @param event_type     str
        @param min_datetime   int  seconds
        @param max_datetime   int  seconds
        @param limit          int         (integer limiting the number of results (0 means unlimited))
        @param descending     boolean     (if True, reverse order (of production time) is applied, e.g. most recent first)
        @retval event_list    []
        @throws NotFound    object with specified parameters does not exist
        @throws NotFound    object with specified parameters does not exist
        """
        event_tuples = []

        try:
            event_tuples = self.container.event_repository.find_events(event_type=type, origin=origin, start_ts=min_datetime, end_ts=max_datetime, limit=limit, descending=descending)
        except Exception as exc:
            log.warning("The UNS find_events operation for event origin = %s and type = %s failed. Error message = %s", origin, type, exc.message)

        events = [item[2] for item in event_tuples]
        log.debug("(find_events) UNS found the following relevant events: %s", events)

        return events

def send_email(event, msg_recipient, smtp_client, rr_client):
    """
    A common method to send email with formatting

    @param event              Event
    @param msg_recipient        str
    @param smtp_client          fake or real smtp client object

    """

    #log.debug("Got type of event to notify on: %s", event.type_)

    #------------------------------------------------------------------------------------
    # the 'from' email address for notification emails
    #------------------------------------------------------------------------------------

    ION_NOTIFICATION_EMAIL_ADDRESS = '[email protected]'
    smtp_sender = CFG.get_safe('server.smtp.sender', ION_NOTIFICATION_EMAIL_ADDRESS)

    msg = convert_events_to_email_message([event], rr_client)
    msg['From'] = smtp_sender
    msg['To'] = msg_recipient
    log.debug("UNS sending email from %s to %s for event type: %s", smtp_sender,msg_recipient, event.type_)
    #log.debug("UNS using the smtp client: %s", smtp_client)

    try:
        smtp_client.sendmail(smtp_sender, [msg_recipient], msg.as_string())
    except: # Can be due to a broken connection... try to create a connection
        smtp_client = setting_up_smtp_client()
        log.debug("Connect again...message received after ehlo exchange: %s", str(smtp_client.ehlo()))
        smtp_client.sendmail(smtp_sender, [msg_recipient], msg.as_string())

    def get_datastore(self, ds_name, profile=DataStore.DS_PROFILE.BASIC, config=None):
        """
        Factory method to get a datastore instance from given name, profile and config.
        @param ds_name  Logical name of datastore (will be scoped with sysname)
        @param profile  One of known constants determining the use of the store
        @param config  Override config to use
        """
        validate_true(ds_name, 'ds_name must be provided')
        if ds_name in self._datastores:
            log.debug("get_datastore(): Found instance of store '%s'" % ds_name)
            return self._datastores[ds_name]

        scoped_name = DatastoreManager.get_scoped_name(ds_name)

        # Create a datastore instance
        log.info("get_datastore(): Create instance of store '%s' as database=%s" % (ds_name, scoped_name))
        new_ds = DatastoreManager.get_datastore_instance(ds_name, profile)

        # Create store if not existing
        if not new_ds.datastore_exists(scoped_name):
            new_ds.create_datastore(scoped_name, create_indexes=True, profile=profile)
        else:
            # NOTE: This may be expensive if called more than once per container
            # If views exist and are dropped and recreated
            new_ds._define_views(profile=profile, keepviews=True)

        # Set a few standard datastore instance fields
        new_ds.local_name = ds_name
        new_ds.ds_profile = profile

        self._datastores[ds_name] = new_ds

        return new_ds

    def stop_consume_impl(self, consumer_tag):
        """
        Stops consuming by consumer tag.
        """
        #log.debug("AMQPTransport.stop_consume_impl(%s): %s", self._client.channel_number, consumer_tag)
        self._sync_call(self._client.basic_cancel, 'callback', consumer_tag)

        # PIKA 0.9.5 / GEVENT interaction problem here
        # we get called back too early, the basic_cancel hasn't really finished processing yet. we need
        # to wait until our consumer tag is removed from the pika channel's consumers dict.
        # See: https://gist.github.com/3751870

        attempts = 6
        while attempts > 0:
            if consumer_tag not in self._client._consumers:
                break
            else:
                log.debug("stop_consume_impl waiting for ctag to be removed from consumers, attempts rem: %s", attempts)

            attempts -= 1

            sleep(1) if attempts < 4 else sleep(0.1)        # Wait shorter the first few times

        if consumer_tag in self._client._consumers:
            raise TransportError("stop_consume_impl did not complete in the expected amount of time, transport may be compromised")

    def delete_doc(self, doc, datastore_name=""):
        if not datastore_name:
            datastore_name = self.datastore_name
        try:
            db = self.server[datastore_name]
        except ValueError:
            raise BadRequest("Data store name %s is invalid" % datastore_name)
        if type(doc) is str:
            log.info('Deleting object %s/%s' % (datastore_name, doc))
            if self._is_in_association(doc, datastore_name):
                obj = self.read(doc, datastore_name)
                log.warn("XXXXXXX Attempt to delete object %s that still has associations" % str(obj))
#                raise BadRequest("Object cannot be deleted until associations are broken")
            try:
                del db[doc]
            except ResourceNotFound:
                raise NotFound('Object with id %s does not exist.' % str(doc))
        else:
            log.info('Deleting object %s/%s' % (datastore_name, doc["_id"]))
            if self._is_in_association(doc["_id"], datastore_name):
                log.warn("XXXXXXX Attempt to delete object %s that still has associations" % str(doc))
#                raise BadRequest("Object cannot be deleted until associations are broken")
            try:
                res = db.delete(doc)
            except ResourceNotFound:
                raise NotFound('Object with id %s does not exist.' % str(doc["_id"]))
            log.debug('Delete result: %s' % str(res))

    def __init__(self, host=None, port=None, datastore_name='prototype', options="", profile=DataStore.DS_PROFILE.BASIC):
        log.debug('__init__(host=%s, port=%s, datastore_name=%s, options=%s' % (host, port, datastore_name, options))
        self.host = host or CFG.server.couchdb.host
        self.port = port or CFG.server.couchdb.port
        # The scoped name of the datastore
        self.datastore_name = datastore_name
        self.auth_str = ""
        try:
            if CFG.server.couchdb.username and CFG.server.couchdb.password:
                self.auth_str = "%s:%[email protected]" % (CFG.server.couchdb.username, CFG.server.couchdb.password)
                log.debug("Using username:password authentication to connect to datastore")
        except AttributeError:
            log.error("CouchDB username:password not configured correctly. Trying anonymous...")

        connection_str = "http://%s%s:%s" % (self.auth_str, self.host, self.port)
        #connection_str = "http://%s:%s" % (self.host, self.port)
        # TODO: Security risk to emit password into log. Remove later.
        log.info('Connecting to CouchDB server: %s' % connection_str)
        self.server = couchdb.Server(connection_str)

        # Datastore specialization
        self.profile = profile

        # serializers
        self._io_serializer     = IonObjectSerializer()
        self._io_deserializer   = IonObjectDeserializer(obj_registry=obj_registry)

 def test_comm_config_type_list(self):
     types = CommConfig.valid_type_list()
     log.debug( "types: %s" % types)
     
     known_types = ['ethernet']
     
     self.assertEqual(sorted(types), sorted(known_types))

    def create_doc(self, doc, object_id=None, datastore_name=""):
        if not datastore_name:
            datastore_name = self.datastore_name
        if '_id' in doc:
            raise BadRequest("Doc must not have '_id'")
        if '_rev' in doc:
            raise BadRequest("Doc must not have '_rev'")

        if object_id:
            try:
                self.read(object_id, '', datastore_name)
                raise BadRequest("Object with id %s already exist" % object_id)
            except NotFound:
                pass

        # Assign an id to doc (recommended in CouchDB documentation)
        doc["_id"] = object_id or uuid4().hex
        log.info('Creating new object %s/%s' % (datastore_name, doc["_id"]))
        log.debug('create doc contents: %s', doc)

        # Save doc.  CouchDB will assign version to doc.
        try:
            res = self.server[datastore_name].save(doc)
        except ResourceNotFound:
            raise BadRequest("Data store %s does not exist" % datastore_name)
        except ResourceConflict:
            raise BadRequest("Object with id %s already exist" % doc["_id"])
        except ValueError:
            raise BadRequest("Data store name %s invalid" % datastore_name)
        log.debug('Create result: %s' % str(res))
        id, version = res
        return (id, version)

    def launch(self):
        """
        Launches the simulator process as indicated by _COMMAND.

        @return (rsn_oms, uri) A pair with the CIOMSSimulator instance and the
                associated URI to establish connection with it.
        """
        log.debug("[OMSim] Launching: %s", _COMMAND)

        self._process = self._spawn(_COMMAND)

        if not self._process or not self.poll():
            msg = "[OMSim] Failed to launch simulator: %s" % _COMMAND
            log.error(msg)
            raise Exception(msg)

        log.debug("[OMSim] process started, pid: %s", self.getpid())

        # give it some time to start up
        sleep(5)

        # get URI:
        uri = None
        with open("logs/rsn_oms_simulator.yml", buffering=1) as f:
            # we expect one of the first few lines to be of the form:
            # rsn_oms_simulator_uri=xxxx
            # where xxxx is the uri -- see oms_simulator_server.
            while uri is None:
                line = f.readline()
                if line.index("rsn_oms_simulator_uri=") == 0:
                    uri = line[len("rsn_oms_simulator_uri="):].strip()

        self._rsn_oms = CIOMSClientFactory.create_instance(uri)
        return self._rsn_oms, uri

    def start(self):

        log.debug("GovernanceController starting ...")

        self._CFG = CFG

        self.enabled = CFG.get_safe('interceptor.interceptors.governance.config.enabled', False)

        log.info("GovernanceInterceptor enabled: %s" % str(self.enabled))

        self.policy_event_subscriber = None

        #containers default to not Org Boundary and ION Root Org
        self._is_container_org_boundary = CFG.get_safe('container.org_boundary',False)
        self._container_org_name = CFG.get_safe('container.org_name', CFG.get_safe('system.root_org', 'ION'))
        self._container_org_id = None
        self._system_root_org_name = CFG.get_safe('system.root_org', 'ION')

        self._is_root_org_container = (self._container_org_name == self._system_root_org_name)

        if self.enabled:

            config = CFG.get_safe('interceptor.interceptors.governance.config')

            self.initialize_from_config(config)

            self.policy_event_subscriber = EventSubscriber(event_type=OT.PolicyEvent, callback=self.policy_event_callback)
            self.policy_event_subscriber.start()

            self.rr_client = ResourceRegistryServiceProcessClient(node=self.container.node, process=self.container)
            self.policy_client = PolicyManagementServiceProcessClient(node=self.container.node, process=self.container)

    def __init__(self, container):
        log.debug("ExchangeManager initializing ...")
        self.container = container

        # Define the callables that can be added to Container public API
        # @TODO: remove
        self.container_api = [self.create_xs,
                              self.create_xp,
                              self.create_xn_service,
                              self.create_xn_process,
                              self.create_xn_queue]

        # Add the public callables to Container
        for call in self.container_api:
            setattr(self.container, call.__name__, call)

        self.default_xs         = ExchangeSpace(self, ION_ROOT_XS)
        self._xs_cache          = {}        # caching of xs names to RR objects
        self._default_xs_obj    = None      # default XS registry object
        self.org_id             = None

        # mappings
        self.xs_by_name = { ION_ROOT_XS: self.default_xs }      # friendly named XS to XSO
        self.xn_by_name = {}                                    # friendly named XN to XNO
        # xn by xs is a property

        self._chan = None

        # @TODO specify our own to_name here so we don't get auto-behavior - tricky chicken/egg
        self._ems_client    = ExchangeManagementServiceProcessClient(process=self.container)
        self._rr_client     = ResourceRegistryServiceProcessClient(process=self.container)

    def get_data_process_subscriptions_count(self, data_process_id=""):
        if not data_process_id:
            raise BadRequest("The data_process_definition_id parameter is empty")

        subscription_ids, _ = self.clients.resource_registry.find_objects(subject=data_process_id, predicate=PRED.hasSubscription, id_only=True)
        log.debug("get_data_process_subscriptions_count(id=%s): %s subscriptions", data_process_id, len(subscription_ids))
        return len(subscription_ids)

    def find_res_by_lcstate(self, lcstate, restype=None, id_only=False):
        log.debug("find_res_by_lcstate(lcstate=%s, restype=%s)" % (lcstate, restype))
        if type(id_only) is not bool:
            raise BadRequest('id_only must be type bool, not %s' % type(id_only))
        db = self.server[self.datastore_name]
        view = db.view(self._get_viewname("resource","by_lcstate"), include_docs=(not id_only))
        is_hierarchical = (lcstate in ResourceLifeCycleSM.STATE_ALIASES)
        # lcstate is a hiearachical state and we need to treat the view differently
        if is_hierarchical:
            key = [1, lcstate]
        else:
            key = [0, lcstate]
        if restype:
            key.append(restype)
        endkey = list(key)
        endkey.append(END_MARKER)
        rows = view[key:endkey]

        if is_hierarchical:
            res_assocs = [dict(lcstate=row['key'][3], type=row['key'][2], name=row['key'][4], id=row.id) for row in rows]
        else:
            res_assocs = [dict(lcstate=row['key'][1], type=row['key'][2], name=row['key'][3], id=row.id) for row in rows]

        log.debug("find_res_by_lcstate() found %s objects" % (len(res_assocs)))
        if id_only:
            res_ids = [row.id for row in rows]
            return (res_ids, res_assocs)
        else:
            res_docs = [self._persistence_dict_to_ion_object(row.doc) for row in rows]
            return (res_docs, res_assocs)

 def lookup(self, qualified_key='/'):
     """
     Read entry residing in directory at parent node level.
     """
     log.debug("Reading content at path %s" % qualified_key)
     direntry = self._safe_read(qualified_key)
     return direntry.attributes if direntry else None

    def find_res_by_alternative_id(self, alt_id=None, alt_id_ns=None, id_only=False, filter=None):
        log.debug("find_res_by_alternative_id(restype=%s, alt_id_ns=%s)", alt_id, alt_id_ns)
        if alt_id and type(alt_id) is not str:
            raise BadRequest('Argument alt_id illegal')
        if alt_id_ns and type(alt_id_ns) is not str:
            raise BadRequest('Argument alt_id_ns illegal')
        if type(id_only) is not bool:
            raise BadRequest('id_only must be type bool, not %s' % type(id_only))
        filter = filter if filter is not None else {}
        ds, datastore_name = self._get_datastore()
        key = []
        if alt_id:
            key.append(alt_id)
            if alt_id_ns is not None:
                key.append(alt_id_ns)

        endkey = self._get_endkey(key)
        rows = ds.view(self._get_view_name("resource", "by_altid"), include_docs=(not id_only), start_key=key, end_key=endkey, **filter)

        if alt_id_ns and not alt_id:
            res_assocs = [dict(alt_id=row['key'][0], alt_id_ns=row['key'][1], id=row['id']) for row in rows if row['key'][1] == alt_id_ns]
        else:
            res_assocs = [dict(alt_id=row['key'][0], alt_id_ns=row['key'][1], id=row['id']) for row in rows]
        log.debug("find_res_by_alternative_id() found %s objects", len(res_assocs))
        self._count(find_res_by_altid_call=1, find_res_by_altid_obj=len(res_assocs))
        if id_only:
            res_ids = [row['id'] for row in res_assocs]
            return (res_ids, res_assocs)
        else:
            if alt_id_ns and not alt_id:
                res_docs = [self._persistence_dict_to_ion_object(self._get_row_doc(row)) for row in rows if row['key'][1] == alt_id_ns]
            else:
                res_docs = [self._persistence_dict_to_ion_object(self._get_row_doc(row)) for row in rows]
            return (res_docs, res_assocs)

    def overlay(self, transport, *methods):
        for m in methods:
            self._methods[m] = getattr(transport, m)

        log.debug("ComposableTransport.overlay(%s) %s %s (%s)", self.channel_number, type(transport), transport, transport.channel_number)

        self._transports.append(transport)

def setting_up_smtp_client():
    '''
    Sets up the smtp client
    '''

    #------------------------------------------------------------------------------------
    # the default smtp server
    #------------------------------------------------------------------------------------

    ION_SMTP_SERVER = 'mail.oceanobservatories.org'

    smtp_host = CFG.get_safe('server.smtp.host', ION_SMTP_SERVER)
    smtp_port = CFG.get_safe('server.smtp.port', 25)
    smtp_sender = CFG.get_safe('server.smtp.sender')
    smtp_password = CFG.get_safe('server.smtp.password')

    if CFG.get_safe('system.smtp',False): #Default is False - use the fake_smtp
        log.debug('Using the real SMTP library to send email notifications!')

        smtp_client = smtplib.SMTP(smtp_host)
        smtp_client.ehlo()
        smtp_client.starttls()
        smtp_client.login(smtp_sender, smtp_password)

    else:
        log.debug('Using a fake SMTP library to simulate email notifications!')

        smtp_client = fake_smtplib.SMTP(smtp_host)

    return smtp_client