Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
u'size': 2048}),
]
class _BaseObject(object):
pass
class TestManager(manager.AgentManager):
def __init__(self):
super(TestManager, self).__init__()
self.keystone = None
class TestImagePollster(test.BaseTestCase):
def fake_get_glance_client(self, ksclient):
glanceclient = _BaseObject()
setattr(glanceclient, "images", _BaseObject())
setattr(glanceclient.images,
"list", lambda *args, **kwargs: iter(IMAGE_LIST))
return glanceclient
@mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
def setUp(self):
super(TestImagePollster, self).setUp()
self.stubs = self.useFixture(moxstubout.MoxStubout()).stubs
self.context = context.get_admin_context()
self.manager = TestManager()
self.stubs.Set(glance._Base, 'get_glance_client',
self.fake_get_glance_client)
u'ephemeral_gb': 0,
u'host': u'compute-host-name',
u'availability_zone': u'1e3ce043029547f1a61c1996d1a531a4',
u'os_type': u'linux?',
u'architecture': u'x86',
u'image_ref': u'UUID',
u'kernel_id': u'1e3ce043029547f1a61c1996d1a531a5',
u'ramdisk_id': u'1e3ce043029547f1a61c1996d1a531a6',
},
u'priority': u'INFO',
u'publisher_id': u'compute.vagrant-precise',
u'timestamp': u'2012-05-08 20:23:48.028195',
}
class NotificationBaseTestCase(test.BaseTestCase):
def test_handle_event_type(self):
self.assertFalse(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['compute']))
self.assertFalse(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['compute.*.foobar']))
self.assertFalse(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['compute.*.*.foobar']))
self.assertTrue(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['compute.*']))
self.assertTrue(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['*']))
self.assertTrue(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['compute.*.start']))
self.assertTrue(plugin.NotificationBase._handle_event_type(
'compute.instance.start', ['*.start']))
self.assertTrue(plugin.NotificationBase._handle_event_type(
# 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.
"""Base class for tests in ceilometer/alarm/evaluator/
"""
import mock
from ceilometer.openstack.common import test
from ceilometer.openstack.common import timeutils
class TestEvaluatorBase(test.BaseTestCase):
def setUp(self):
super(TestEvaluatorBase, self).setUp()
self.api_client = mock.Mock()
self.notifier = mock.MagicMock()
self.evaluator = self.EVALUATOR(self.notifier)
self.prepare_alarms()
def tearDown(self):
super(TestEvaluatorBase, self).tearDown()
timeutils.utcnow.override_time = None
@staticmethod
def prepare_alarms(self):
self.alarms = []
def _evaluate_all_alarms(self):
from ceilometer.alarm import service
from ceilometer.openstack.common import context
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common import test
DATA_JSON = ('{"current": "ALARM", "alarm_id": "foobar",'
' "reason": "what ?", "previous": "OK"}')
NOTIFICATION = dict(alarm_id='foobar',
condition=dict(threshold=42),
reason='what ?',
previous='OK',
current='ALARM')
class TestAlarmNotifier(test.BaseTestCase):
def setUp(self):
super(TestAlarmNotifier, self).setUp()
self.CONF = self.useFixture(config.Config()).conf
self.service = service.AlarmNotifierService('somehost', 'sometopic')
@mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
def test_init_host(self):
# If we try to create a real RPC connection, init_host() never
# returns. Mock it out so we can establish the service
# configuration.
with mock.patch('ceilometer.openstack.common.rpc.create_connection'):
self.service.start()
def test_notify_alarm(self):
data = {
from stevedore import extension
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import test
from ceilometer.openstack.common import timeutils
from ceilometer import pipeline
from ceilometer import publisher
from ceilometer.publisher import test as test_publisher
from ceilometer import sample
from ceilometer import transformer
from ceilometer.transformer import accumulator
from ceilometer.transformer import conversions
class TestTransformerAccumulator(test.BaseTestCase):
def test_handle_sample(self):
test_sample = sample.Sample(
name='a',
type=sample.TYPE_GAUGE,
volume=1,
unit='B',
user_id="test_user",
project_id="test_proj",
resource_id="test_resource",
timestamp=timeutils.utcnow().isoformat(),
resource_metadata={}
)
# Test when size is set to less than 1.
tf = accumulator.TransformerAccumulator(size=0)
"""Test the methods related to query."""
import datetime
import mock
import wsme
from ceilometer.api.controllers import v2 as api
from ceilometer.api.controllers.v2 import Query
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import test
from ceilometer.openstack.common import timeutils
from ceilometer import storage
from ceilometer.tests import base as tests_base
class TestQuery(test.BaseTestCase):
def test_get_value_as_type_with_integer(self):
query = Query(field='metadata.size',
op='eq',
value='123',
type='integer')
expected = 123
self.assertEqual(query._get_value_as_type(), expected)
def test_get_value_as_type_with_float(self):
query = Query(field='metadata.size',
op='eq',
value='123.456',
type='float')
expected = 123.456
self.assertEqual(query._get_value_as_type(), expected)
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for ceilometer/collector/dispatcher/database.py
"""
from datetime import datetime
from ceilometer.collector.dispatcher import database
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import test
from ceilometer.publisher import rpc
from ceilometer.storage import base
class TestDispatcherDB(test.BaseTestCase):
def setUp(self):
super(TestDispatcherDB, self).setUp()
self.CONF = self.useFixture(config.Config()).conf
self.dispatcher = database.DatabaseDispatcher(self.CONF)
self.ctx = None
self.mox = self.useFixture(moxstubout.MoxStubout()).mox
def test_valid_message(self):
msg = {'counter_name': 'test',
'resource_id': self.id(),
'counter_volume': 1,
}
msg['message_signature'] = rpc.compute_signature(
msg,
self.CONF.publisher_rpc.metering_secret,