credentials.user = creds_dict['vim_user']
credentials.password = creds_dict['vim_password']
credentials.tenant_name = creds_dict['vim_tenant_name']
- if 'config' in creds_dict:
- credentials.config = json.dumps(creds_dict['config'])
+ if 'config' not in creds_dict:
+ creds_dict['config'] = {}
+ credentials.config = json.dumps(creds_dict['config'])
self.database_manager.save_credentials(credentials)
def get_credentials(self, vim_uuid):
- return self.database_manager.get_credentials(vim_uuid)
+ creds = self.database_manager.get_credentials(vim_uuid)
+ if creds.config is None:
+ creds.config = {}
+ return creds
def delete_auth_credentials(self, creds_dict):
credentials = self.get_credentials(creds_dict['_id'])
from peewee import *
from playhouse.sqlite_ext import SqliteExtDatabase
-from osm_mon.plugins.OpenStack.settings import Config
+from osm_mon.core.settings import Config
log = logging.getLogger(__name__)
cfg = Config.instance()
user = CharField()
password = CharField()
tenant_name = CharField()
- config = TextField(null=True)
+ config = TextField(default='{}')
class Alarm(BaseModel):
import yaml
+from osm_mon.core.settings import Config
+
logging.basicConfig(stream=sys.stdout,
format='%(asctime)s %(message)s',
datefmt='%m/%d/%Y %I:%M:%S %p',
level=logging.INFO)
log = logging.getLogger(__name__)
-
sys.path.append(os.path.abspath(os.path.join(os.path.realpath(__file__), '..', '..', '..', '..')))
from kafka import KafkaConsumer
from osm_mon.core.auth import AuthManager
from osm_mon.core.database import DatabaseManager
-# Initialize servers
-if "BROKER_URI" in os.environ:
- server = {'server': os.getenv("BROKER_URI")}
-else:
- server = {'server': 'localhost:9092'}
+cfg = Config.instance()
+cfg.read_environ()
# Initialize consumers for alarms and metrics
-common_consumer = KafkaConsumer(bootstrap_servers=server['server'],
+common_consumer = KafkaConsumer(bootstrap_servers=cfg.BROKER_URI,
key_deserializer=bytes.decode,
value_deserializer=bytes.decode,
group_id="mon-consumer")
try:
credentials = database_manager.get_credentials(vim_uuid)
return credentials.type
- except Exception as exc:
+ except Exception:
log.exception("Error getting vim_type: ")
return None
values = json.loads(message.value)
except ValueError:
values = yaml.safe_load(message.value)
- # Check the message topic
- if message.topic == "metric_request":
- # Check the vim desired by the message
- vim_type = get_vim_type(values['vim_uuid'])
-
- if vim_type == "openstack":
- log.info("This message is for the OpenStack plugin.")
- openstack_metrics.metric_calls(message)
- elif vim_type == "aws":
- log.info("This message is for the CloudWatch plugin.")
- aws_conn = aws_connection.setEnvironment()
- cloudwatch_metrics.metric_calls(message, aws_conn)
-
- elif vim_type == "vmware":
- log.info("This metric_request message is for the vROPs plugin.")
- vrops_rcvr.consume(message)
- else:
- log.debug("vim_type is misconfigured or unsupported; %s",
- vim_type)
-
- elif message.topic == "alarm_request":
- # Check the vim desired by the message
- vim_type = get_vim_type(values['vim_uuid'])
- if vim_type == "openstack":
- log.info("This message is for the OpenStack plugin.")
- openstack_alarms.alarming(message)
-
- elif vim_type == "aws":
- log.info("This message is for the CloudWatch plugin.")
- aws_conn = aws_connection.setEnvironment()
- cloudwatch_alarms.alarm_calls(message, aws_conn)
-
- elif vim_type == "vmware":
- log.info("This alarm_request message is for the vROPs plugin.")
- vrops_rcvr.consume(message)
-
- else:
- log.debug("vim_type is misconfigured or unsupported; %s",
- vim_type)
-
- elif message.topic == "vim_account":
+ if message.topic == "vim_account":
if message.key == "create" or message.key == "edit":
auth_manager.store_auth_credentials(values)
if message.key == "delete":
auth_manager.delete_auth_credentials(values)
- # TODO: Remove in the near future when all plugins support vim_uuid. Modify tests accordingly.
- elif message.topic == "access_credentials":
+ else:
# Check the vim desired by the message
vim_type = get_vim_type(values['vim_uuid'])
+ if vim_type == "openstack":
+ log.info("This message is for the OpenStack plugin.")
+ if message.topic == "metric_request":
+ openstack_metrics.metric_calls(message)
+ if message.topic == "alarm_request":
+ openstack_alarms.alarming(message)
- if vim_type == "aws":
+ elif vim_type == "aws":
log.info("This message is for the CloudWatch plugin.")
- aws_access_credentials.access_credential_calls(message)
+ aws_conn = aws_connection.setEnvironment()
+ if message.topic == "metric_request":
+ cloudwatch_metrics.metric_calls(message, aws_conn)
+ if message.topic == "alarm_request":
+ cloudwatch_alarms.alarm_calls(message, aws_conn)
+ if message.topic == "access_credentials":
+ aws_access_credentials.access_credential_calls(message)
elif vim_type == "vmware":
- log.info("This access_credentials message is for the vROPs plugin.")
+ log.info("This metric_request message is for the vROPs plugin.")
vrops_rcvr.consume(message)
else:
log.debug("vim_type is misconfigured or unsupported; %s",
vim_type)
- else:
- log.info("This topic is not relevant to any of the MON plugins.")
-
-
- except Exception as exc:
- log.exception("Exception: %s")
+ except Exception:
+ log.exception("Exception processing message: ")
--- /dev/null
+# Copyright 2017 Intel Research and Development Ireland Limited
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Intel Corporation
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: helena.mcgough@intel.com or adrian.hoban@intel.com
+##
+"""Global configuration managed by environment variables."""
+
+import logging
+import os
+
+from collections import namedtuple
+
+from osm_mon.plugins.OpenStack.singleton import Singleton
+
+import six
+
+__author__ = "Helena McGough"
+
+log = logging.getLogger(__name__)
+
+
+class BadConfigError(Exception):
+ """Configuration exception."""
+
+ pass
+
+
+class CfgParam(namedtuple('CfgParam', ['key', 'default', 'data_type'])):
+ """Configuration parameter definition."""
+
+ def value(self, data):
+ """Convert a string to the parameter type."""
+ try:
+ return self.data_type(data)
+ except (ValueError, TypeError):
+ raise BadConfigError(
+ 'Invalid value "%s" for configuration parameter "%s"' % (
+ data, self.key))
+
+
+@Singleton
+class Config(object):
+ """Configuration object."""
+
+ _configuration = [
+ CfgParam('BROKER_URI', "localhost:9092", six.text_type),
+ CfgParam('OS_NOTIFIER_URI', "http://localhost:8662", six.text_type),
+ CfgParam('OS_DEFAULT_GRANULARITY', "300", six.text_type),
+ ]
+
+ _config_dict = {cfg.key: cfg for cfg in _configuration}
+ _config_keys = _config_dict.keys()
+
+ def __init__(self):
+ """Set the default values."""
+ for cfg in self._configuration:
+ setattr(self, cfg.key, cfg.default)
+
+ def read_environ(self):
+ """Check the appropriate environment variables and update defaults."""
+ for key in self._config_keys:
+ try:
+ val = str(os.environ[key])
+ setattr(self, key, val)
+ except KeyError as exc:
+ log.warn("Failed to configure plugin: %s", exc)
+ return
import six
import yaml
+from osm_mon.core.auth import AuthManager
from osm_mon.core.database import DatabaseManager
from osm_mon.core.message_bus.producer import KafkaProducer
+from osm_mon.core.settings import Config
from osm_mon.plugins.OpenStack.common import Common
from osm_mon.plugins.OpenStack.response import OpenStack_Response
-from osm_mon.plugins.OpenStack.settings import Config
log = logging.getLogger(__name__)
config.read_environ()
self._database_manager = DatabaseManager()
+ self._auth_manager = AuthManager()
# Use the Response class to generate valid json response messages
self._response = OpenStack_Response()
# Initializer a producer to send responses back to SO
self._producer = KafkaProducer("alarm_response")
+ def configure_alarm(self, alarm_endpoint, metric_endpoint, auth_token, values, vim_config):
+ """Create requested alarm in Aodh."""
+ url = "{}/v2/alarms/".format(alarm_endpoint)
+
+ # Check if the desired alarm is supported
+ alarm_name = values['alarm_name'].lower()
+ metric_name = values['metric_name'].lower()
+ resource_id = values['resource_uuid']
+
+ if metric_name not in METRIC_MAPPINGS.keys():
+ log.warn("This metric is not supported.")
+ return None, False
+
+ # Check for the required metric
+ metric_id = self.check_for_metric(auth_token, metric_endpoint, metric_name, resource_id)
+
+ try:
+ if metric_id is not None:
+ # Create the alarm if metric is available
+ if 'granularity' in vim_config and 'granularity' not in values:
+ values['granularity'] = vim_config['granularity']
+ payload = self.check_payload(values, metric_name, resource_id,
+ alarm_name)
+ new_alarm = Common.perform_request(
+ url, auth_token, req_type="post", payload=payload)
+ return json.loads(new_alarm.text)['alarm_id'], True
+ else:
+ log.warn("The required Gnocchi metric does not exist.")
+ return None, False
+
+ except Exception as exc:
+ log.warn("Failed to create the alarm: %s", exc)
+ return None, False
+
def alarming(self, message):
"""Consume info from the message bus to manage alarms."""
try:
alarm_endpoint = Common.get_endpoint("alarming", vim_uuid)
metric_endpoint = Common.get_endpoint("metric", vim_uuid)
+ vim_account = self._auth_manager.get_credentials(vim_uuid)
+ vim_config = json.loads(vim_account.config)
+
if message.key == "create_alarm_request":
# Configure/Update an alarm
alarm_details = values['alarm_create_request']
alarm_id, alarm_status = self.configure_alarm(
- alarm_endpoint, metric_endpoint, auth_token, alarm_details)
+ alarm_endpoint, metric_endpoint, auth_token, alarm_details, vim_config)
# Generate a valid response message, send via producer
try:
self._producer.create_alarm_response(
'create_alarm_response', resp_message,
'alarm_response')
- except Exception as exc:
+ except Exception:
log.exception("Response creation failed:")
elif message.key == "list_alarm_request":
self._producer.list_alarm_response(
'list_alarm_response', resp_message,
'alarm_response')
- except Exception as exc:
+ except Exception:
log.exception("Failed to send a valid response back.")
elif message.key == "delete_alarm_request":
self._producer.delete_alarm_response(
'delete_alarm_response', resp_message,
'alarm_response')
- except Exception as exc:
- log.warn("Failed to create delete response:%s", exc)
+ except Exception:
+ log.exception("Failed to create delete response: ")
elif message.key == "acknowledge_alarm":
# Acknowledge that an alarm has been dealt with by the SO
alarm_details = values['alarm_update_request']
alarm_id, status = self.update_alarm(
- alarm_endpoint, auth_token, alarm_details)
+ alarm_endpoint, auth_token, alarm_details, vim_config)
# Generate a response for an update request
try:
self._producer.update_alarm_response(
'update_alarm_response', resp_message,
'alarm_response')
- except Exception as exc:
- log.warn("Failed to send an update response:%s", exc)
+ except Exception:
+ log.exception("Failed to send an update response: ")
else:
log.debug("Unknown key, no action will be performed")
return
- def configure_alarm(self, alarm_endpoint, metric_endpoint, auth_token, values):
- """Create requested alarm in Aodh."""
- url = "{}/v2/alarms/".format(alarm_endpoint)
-
- # Check if the desired alarm is supported
- alarm_name = values['alarm_name'].lower()
- metric_name = values['metric_name'].lower()
- resource_id = values['resource_uuid']
-
- if metric_name not in METRIC_MAPPINGS.keys():
- log.warn("This metric is not supported.")
- return None, False
-
- # Check for the required metric
- metric_id = self.check_for_metric(auth_token, metric_endpoint, metric_name, resource_id)
-
- try:
- if metric_id is not None:
- # Create the alarm if metric is available
- payload = self.check_payload(values, metric_name, resource_id,
- alarm_name)
- new_alarm = Common.perform_request(
- url, auth_token, req_type="post", payload=payload)
- return json.loads(new_alarm.text)['alarm_id'], True
- else:
- log.warn("The required Gnocchi metric does not exist.")
- return None, False
-
- except Exception as exc:
- log.warn("Failed to create the alarm: %s", exc)
- return None, False
-
def delete_alarm(self, endpoint, auth_token, alarm_id):
"""Delete alarm function."""
url = "{}/v2/alarms/%s".format(endpoint) % alarm_id
else:
return True
- except Exception as exc:
- log.warn("Failed to delete alarm: %s because %s.", alarm_id, exc)
+ except Exception:
+ log.exception("Failed to delete alarm %s :", alarm_id)
return False
def list_alarms(self, endpoint, auth_token, list_details):
Common.perform_request(
url, auth_token, req_type="put", payload=payload)
return True
- except Exception as exc:
- log.warn("Unable to update alarm state: %s", exc)
+ except Exception:
+ log.exception("Unable to update alarm state: ")
return False
- def update_alarm(self, endpoint, auth_token, values):
+ def update_alarm(self, endpoint, auth_token, values, vim_config):
"""Get alarm name for an alarm configuration update."""
# Get already existing alarm details
url = "{}/v2/alarms/%s".format(endpoint) % values['alarm_uuid']
return None, False
# Generates and check payload configuration for alarm update
+ if 'granularity' in vim_config and 'granularity' not in values:
+ values['granularity'] = vim_config['granularity']
payload = self.check_payload(values, metric_name, resource_id,
alarm_name, alarm_state=alarm_state)
statistic = values['statistic'].lower()
- granularity = '300'
+ granularity = cfg.OS_DEFAULT_GRANULARITY
if 'granularity' in values:
granularity = values['granularity']
"""A Webserver to send alarm notifications from Aodh to the SO."""
import json
import logging
+import os
import sys
import time
-import os
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
from six.moves.BaseHTTPServer import HTTPServer
from osm_mon.plugins.OpenStack.common import Common
from osm_mon.plugins.OpenStack.response import OpenStack_Response
-from osm_mon.plugins.OpenStack.settings import Config
+from osm_mon.core.settings import Config
class NotifierHandler(BaseHTTPRequestHandler):
import datetime
import json
import logging
-
import time
import six
import yaml
from osm_mon.core.message_bus.producer import KafkaProducer
+from osm_mon.core.settings import Config
from osm_mon.plugins.OpenStack.common import Common
-
from osm_mon.plugins.OpenStack.response import OpenStack_Response
-from osm_mon.plugins.OpenStack.settings import Config
log = logging.getLogger(__name__)
+++ /dev/null
-# Copyright 2017 Intel Research and Development Ireland Limited
-# *************************************************************
-
-# This file is part of OSM Monitoring module
-# All Rights Reserved to Intel Corporation
-
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-
-# http://www.apache.org/licenses/LICENSE-2.0
-
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-# For those usages not covered by the Apache License, Version 2.0 please
-# contact: helena.mcgough@intel.com or adrian.hoban@intel.com
-##
-"""Configurations for the OpenStack plugins."""
-
-import logging
-import os
-
-from collections import namedtuple
-
-from osm_mon.plugins.OpenStack.singleton import Singleton
-
-import six
-
-__author__ = "Helena McGough"
-
-log = logging.getLogger(__name__)
-
-
-class BadConfigError(Exception):
- """Configuration exception."""
-
- pass
-
-
-class CfgParam(namedtuple('CfgParam', ['key', 'default', 'data_type'])):
- """Configuration parameter definition."""
-
- def value(self, data):
- """Convert a string to the parameter type."""
- try:
- return self.data_type(data)
- except (ValueError, TypeError):
- raise BadConfigError(
- 'Invalid value "%s" for configuration parameter "%s"' % (
- data, self.key))
-
-
-@Singleton
-class Config(object):
- """Plugin confguration."""
-
- _configuration = [
- CfgParam('OS_NOTIFIER_URI', "http://localhost:8662", six.text_type),
- ]
-
- _config_dict = {cfg.key: cfg for cfg in _configuration}
- _config_keys = _config_dict.keys()
-
- def __init__(self):
- """Set the default values."""
- for cfg in self._configuration:
- setattr(self, cfg.key, cfg.default)
-
- def read_environ(self):
- """Check the appropriate environment variables and update defaults."""
- for key in self._config_keys:
- try:
- val = str(os.environ[key])
- setattr(self, key, val)
- except KeyError as exc:
- log.warn("Failed to configure plugin: %s", exc)
- return
import mock
+from osm_mon.core.auth import AuthManager
+from osm_mon.core.database import VimCredentials
from osm_mon.plugins.OpenStack.Aodh import alarming as alarm_req
from osm_mon.plugins.OpenStack.common import Common
log = logging.getLogger(__name__)
+mock_creds = VimCredentials()
+mock_creds.config = '{}'
+
class Message(object):
"""A class to mock a message object value for alarm requests."""
self.alarming = alarm_req.Alarming()
self.alarming.common = Common()
+ @mock.patch.object(AuthManager, 'get_credentials')
@mock.patch.object(Common, 'get_endpoint')
@mock.patch.object(Common, 'get_auth_token')
- def test_alarming_authentication(self, get_token, get_endpoint):
+ def test_alarming_authentication(self, get_token, get_endpoint, get_creds):
"""Test getting an auth_token and endpoint for alarm requests."""
# if auth_token is None environment variables are used to authenticate
message = Message()
+ get_creds.return_value = mock_creds
+
self.alarming.alarming(message)
get_token.assert_called_with('test_id')
@mock.patch.object(Common, 'get_endpoint', mock.Mock())
@mock.patch.object(Common, 'get_auth_token', mock.Mock())
+ @mock.patch.object(AuthManager, 'get_credentials')
@mock.patch.object(alarm_req.Alarming, 'delete_alarm')
- def test_delete_alarm_key(self, del_alarm):
+ def test_delete_alarm_key(self, del_alarm, get_creds):
"""Test the functionality for a create alarm request."""
# Mock a message value and key
message = Message()
'alarm_delete_request':
{'alarm_uuid': 'my_alarm_id'}})
+ get_creds.return_value = mock_creds
+
# Call the alarming functionality and check delete request
self.alarming.alarming(message)
del_alarm.assert_called_with(mock.ANY, mock.ANY, 'my_alarm_id')
@mock.patch.object(Common, 'get_endpoint', mock.Mock())
@mock.patch.object(Common, 'get_auth_token', mock.Mock())
+ @mock.patch.object(AuthManager, 'get_credentials')
@mock.patch.object(alarm_req.Alarming, 'list_alarms')
- def test_list_alarm_key(self, list_alarm):
+ def test_list_alarm_key(self, list_alarm, get_creds):
"""Test the functionality for a list alarm request."""
# Mock a message with list alarm key and value
message = Message()
message.key = 'list_alarm_request'
message.value = json.dumps({'vim_uuid': 'test_id', 'alarm_list_request': 'my_alarm_details'})
+ get_creds.return_value = mock_creds
+
# Call the alarming functionality and check list functionality
self.alarming.alarming(message)
list_alarm.assert_called_with(mock.ANY, mock.ANY, 'my_alarm_details')
@mock.patch.object(Common, 'get_auth_token', mock.Mock())
@mock.patch.object(Common, 'get_endpoint', mock.Mock())
+ @mock.patch.object(AuthManager, 'get_credentials')
@mock.patch.object(alarm_req.Alarming, 'update_alarm_state')
- def test_ack_alarm_key(self, ack_alarm):
+ def test_ack_alarm_key(self, ack_alarm, get_creds):
"""Test the functionality for an acknowledge alarm request."""
# Mock a message with acknowledge alarm key and value
message = Message()
'ack_details':
{'alarm_uuid': 'my_alarm_id'}})
+ get_creds.return_value = mock_creds
+
# Call alarming functionality and check acknowledge functionality
self.alarming.alarming(message)
ack_alarm.assert_called_with(mock.ANY, mock.ANY, 'my_alarm_id')
@mock.patch.object(Common, 'get_auth_token', mock.Mock())
@mock.patch.object(Common, 'get_endpoint', mock.Mock())
+ @mock.patch.object(AuthManager, 'get_credentials')
@mock.patch.object(alarm_req.Alarming, 'configure_alarm')
- def test_config_alarm_key(self, config_alarm):
+ def test_config_alarm_key(self, config_alarm, get_creds):
"""Test the functionality for a create alarm request."""
# Mock a message with config alarm key and value
message = Message()
message.key = 'create_alarm_request'
message.value = json.dumps({'vim_uuid': 'test_id', 'alarm_create_request': 'alarm_details'})
+ get_creds.return_value = mock_creds
+
# Call alarming functionality and check config alarm call
config_alarm.return_value = 'my_alarm_id', True
self.alarming.alarming(message)
- config_alarm.assert_called_with(mock.ANY, mock.ANY, mock.ANY, 'alarm_details')
+ config_alarm.assert_called_with(mock.ANY, mock.ANY, mock.ANY, 'alarm_details', {})
import mock
+from osm_mon.core.settings import Config
from osm_mon.plugins.OpenStack.Aodh import alarming as alarm_req
from osm_mon.plugins.OpenStack.common import Common
-from osm_mon.plugins.OpenStack.settings import Config
log = logging.getLogger(__name__)
values = {"alarm_name": "my_alarm",
"metric_name": "my_metric",
"resource_uuid": "my_r_id"}
- self.alarming.configure_alarm(alarm_endpoint, metric_endpoint, auth_token, values)
+ self.alarming.configure_alarm(alarm_endpoint, metric_endpoint, auth_token, values, {})
perf_req.assert_not_called()
perf_req.reset_mock()
check_metric.return_value = None
- self.alarming.configure_alarm(alarm_endpoint, metric_endpoint, auth_token, values)
+ self.alarming.configure_alarm(alarm_endpoint, metric_endpoint, auth_token, values, {})
perf_req.assert_not_called()
@mock.patch.object(alarm_req.Alarming, "check_payload")
check_metric.return_value = "my_metric_id"
check_pay.return_value = "my_payload"
- self.alarming.configure_alarm(alarm_endpoint, metric_endpoint, auth_token, values)
+ self.alarming.configure_alarm(alarm_endpoint, metric_endpoint, auth_token, values, {})
perf_req.assert_called_with(
"alarm_endpoint/v2/alarms/", auth_token,
req_type="post", payload="my_payload")
"""Test update alarm with invalid get response."""
values = {"alarm_uuid": "my_alarm_id"}
- self.alarming.update_alarm(alarm_endpoint, auth_token, values)
+ self.alarming.update_alarm(alarm_endpoint, auth_token, values, {})
perf_req.assert_called_with(mock.ANY, auth_token, req_type="get")
check_pay.assert_not_called()
check_pay.return_value = None
values = {"alarm_uuid": "my_alarm_id"}
- self.alarming.update_alarm(alarm_endpoint, auth_token, values)
+ self.alarming.update_alarm(alarm_endpoint, auth_token, values, {})
perf_req.assert_called_with(mock.ANY, auth_token, req_type="get")
self.assertEqual(perf_req.call_count, 1)
perf_req.return_value = resp
values = {"alarm_uuid": "my_alarm_id"}
- self.alarming.update_alarm(alarm_endpoint, auth_token, values)
+ self.alarming.update_alarm(alarm_endpoint, auth_token, values, {})
check_pay.assert_called_with(values, "disk_write_ops", "my_resource_id",
"my_alarm", alarm_state="alarm")
"""Tests for all common OpenStack methods."""
import json
-
import logging
-
import unittest
-from keystoneclient.v3 import client
-
import mock
+import requests
+from keystoneclient.v3 import client
from osm_mon.core.auth import AuthManager
from osm_mon.core.database import VimCredentials
+from osm_mon.core.settings import Config
from osm_mon.plugins.OpenStack.common import Common
-from osm_mon.plugins.OpenStack.settings import Config
-
-import requests
__author__ = "Helena McGough"
"""Tests for all common OpenStack methods."""
import json
-
import unittest
-
from BaseHTTPServer import BaseHTTPRequestHandler
import mock
from osm_mon.core.message_bus.producer import KafkaProducer
-
+from osm_mon.core.settings import Config
from osm_mon.plugins.OpenStack.Aodh.alarming import Alarming
from osm_mon.plugins.OpenStack.common import Common
from osm_mon.plugins.OpenStack.response import OpenStack_Response
-from osm_mon.plugins.OpenStack.settings import Config
# Mock data from post request
post_data = json.dumps({"severity": "critical",
"""Tests for settings for OpenStack plugins configurations."""
import logging
-
import os
-
import unittest
-import mock
-
-from osm_mon.plugins.OpenStack.settings import Config
+from osm_mon.core.settings import Config
log = logging.getLogger(__name__)
"""Tests for all common OpenStack methods."""
import json
-
import logging
-
import socket
import unittest
+from threading import Thread
+import mock
+import requests
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
from six.moves.BaseHTTPServer import HTTPServer
-from threading import Thread
-
from osm_mon.core.message_bus.producer import KafkaProducer
-
-import mock
-
+from osm_mon.core.settings import Config
from osm_mon.plugins.OpenStack.Aodh.alarming import Alarming
from osm_mon.plugins.OpenStack.common import Common
from osm_mon.plugins.OpenStack.response import OpenStack_Response
-from osm_mon.plugins.OpenStack.settings import Config
-
-import requests
log = logging.getLogger(__name__)
projects / osm / MON.git / commitdiff