# Path: monasca_notification/notification.py
# class Notification(object):
#     def __init__(self, id, type, name, address, period, retry_count, alarm):
#     def __eq__(self, other):
#     def __ne__(self, other):
#     def to_json(self):
# 
# Path: monasca_notification/processors/alarm_processor.py
# CONF = cfg.CONF
# class AlarmProcessor(object):
#     def __init__(self):
#     def _parse_alarm(alarm_data):
#     def _alarm_is_valid(self, alarm):
#     def _build_notification(self, alarm):
#     def to_notification(self, raw_alarm):
# 
# Path: tests/base.py
# class DisableStatsdFixture(fixtures.Fixture):
# class ConfigFixture(oo_cfg.Config):
# class BaseTestCase(oslotest_base.BaseTestCase):
# class PluginTestCase(BaseTestCase):
#     def setUp(self):
#     def __init__(self):
#     def setUp(self):
#     def _clean_config_loaded_flag():
#     def setUp(self):
#     def conf_override(**kw):
#     def conf_default(**kw):
#     def setUp(self, register_opts=None):


import collections
import json
import time
from unittest import mock
from monasca_common.kafka import legacy_kafka_message
from monasca_notification import notification as m_notification
from monasca_notification.processors import alarm_processor
from tests import base
        json_msg = json.dumps({'alarm-transitioned': message})
        msg_tuple = message_tuple(key, json_msg)
        return legacy_kafka_message.LegacyKafkaMessage([partition,
                                                        alarm_tuple(offset,
                                                                    msg_tuple)])

    @mock.patch('pymysql.connect')
    @mock.patch('monasca_notification.processors.alarm_processor.log')
    def _run_alarm_processor(self, alarm, sql_response, mock_log, mock_mysql):
        """Runs a mocked alarm processor reading from queue while running,
        returns (queue_message, log_message)
        """
        # Since the log runs in another thread I can mock it directly, instead
        # change the methods to put to a queue
        mock_log.warn = self.trap.append
        mock_log.error = self.trap.append
        mock_log.exception = self.trap.append

        # Setup the sql response
        if sql_response is not None:
            mock_mysql.return_value = mock_mysql
            mock_mysql.cursor.return_value = mock_mysql
            mock_mysql.__iter__.return_value = sql_response

        self.conf_override(group='mysql', ssl=None,
                           host='localhost', port='3306',
                           user='mysql_user', db='dbname',
                           passwd='mysql_passwd')
        self.conf_override(group='statsd', host='localhost',
                           port=8125)

# Path: monasca_notification/notification.py
# class Notification(object):
#     def __init__(self, id, type, name, address, period, retry_count, alarm):
#     def __eq__(self, other):
#     def __ne__(self, other):
#     def to_json(self):
# 
# Path: monasca_notification/processors/alarm_processor.py
# CONF = cfg.CONF
# class AlarmProcessor(object):
#     def __init__(self):
#     def _parse_alarm(alarm_data):
#     def _alarm_is_valid(self, alarm):
#     def _build_notification(self, alarm):
#     def to_notification(self, raw_alarm):
# 
# Path: tests/base.py
# class DisableStatsdFixture(fixtures.Fixture):
# class ConfigFixture(oo_cfg.Config):
# class BaseTestCase(oslotest_base.BaseTestCase):
# class PluginTestCase(BaseTestCase):
#     def setUp(self):
#     def __init__(self):
#     def setUp(self):
#     def _clean_config_loaded_flag():
#     def setUp(self):
#     def conf_override(**kw):
#     def conf_default(**kw):
#     def setUp(self, register_opts=None):


# Path: tests/test_alarm_processor.py
import collections
import json
import time
from unittest import mock
from monasca_common.kafka import legacy_kafka_message
from monasca_notification import notification as m_notification
from monasca_notification.processors import alarm_processor
from tests import base

        json_msg = json.dumps({'alarm-transitioned': message})
        msg_tuple = message_tuple(key, json_msg)
        return legacy_kafka_message.LegacyKafkaMessage([partition,
                                                        alarm_tuple(offset,
                                                                    msg_tuple)])

    @mock.patch('pymysql.connect')
    @mock.patch('monasca_notification.processors.alarm_processor.log')
    def _run_alarm_processor(self, alarm, sql_response, mock_log, mock_mysql):
        """Runs a mocked alarm processor reading from queue while running,
        returns (queue_message, log_message)
        """
        # Since the log runs in another thread I can mock it directly, instead
        # change the methods to put to a queue
        mock_log.warn = self.trap.append
        mock_log.error = self.trap.append
        mock_log.exception = self.trap.append

        # Setup the sql response
        if sql_response is not None:
            mock_mysql.return_value = mock_mysql
            mock_mysql.cursor.return_value = mock_mysql
            mock_mysql.__iter__.return_value = sql_response

        self.conf_override(group='mysql', ssl=None,
                           host='localhost', port='3306',
                           user='mysql_user', db='dbname',
                           passwd='mysql_passwd')
        self.conf_override(group='statsd', host='localhost',
                           port=8125)

        processor = alarm_processor.AlarmProcessor()