ceilometer.tests.base

Source code for ceilometer.tests.base

# Copyright 2012 New Dream Network (DreamHost)
#
# 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.
"""Test base classes.
"""
import functools
import os
import tempfile

import fixtures
import oslo_messaging.conffixture
from oslo_utils import timeutils
from oslotest import base
import six
from testtools import testcase
import yaml

import ceilometer
from ceilometer import messaging


[docs]class BaseTestCase(base.BaseTestCase):
[docs] def setup_messaging(self, conf, exchange=None): self.useFixture(oslo_messaging.conffixture.ConfFixture(conf)) conf.set_override("notification_driver", ["messaging"]) if not exchange: exchange = 'ceilometer' conf.set_override("control_exchange", exchange) # NOTE(sileht): Ensure a new oslo.messaging driver is loaded # between each tests self.transport = messaging.get_transport(conf, "fake://", cache=False) self.useFixture(fixtures.MockPatch( 'ceilometer.messaging.get_transport', return_value=self.transport))
[docs] def cfg2file(self, data): cfgfile = tempfile.NamedTemporaryFile(mode='w', delete=False) self.addCleanup(os.remove, cfgfile.name) cfgfile.write(yaml.safe_dump(data)) cfgfile.close() return cfgfile.name
[docs] def assertTimestampEqual(self, first, second, msg=None): """Checks that two timestamps are equals. This relies on assertAlmostEqual to avoid rounding problem, and only checks up the first microsecond values. """ return self.assertAlmostEqual( timeutils.delta_seconds(first, second), 0.0, places=5)
[docs] def assertIsEmpty(self, obj): try: if len(obj) != 0: self.fail("%s is not empty" % type(obj)) except (TypeError, AttributeError): self.fail("%s doesn't have length" % type(obj))
[docs] def assertIsNotEmpty(self, obj): try: if len(obj) == 0: self.fail("%s is empty" % type(obj)) except (TypeError, AttributeError): self.fail("%s doesn't have length" % type(obj))
[docs] @staticmethod def path_get(project_file=None): root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', ) ) if project_file: return os.path.join(root, project_file) else: return root
def _skip_decorator(func): @functools.wraps(func) def skip_if_not_implemented(*args, **kwargs): try: return func(*args, **kwargs) except ceilometer.NotImplementedError as e: raise testcase.TestSkipped(six.text_type(e)) return skip_if_not_implemented
[docs]class SkipNotImplementedMeta(type): def __new__(cls, name, bases, local): for attr in local: value = local[attr] if callable(value) and ( attr.startswith('test_') or attr == 'setUp'): local[attr] = _skip_decorator(value) return type.__new__(cls, name, bases, local)
Creative Commons Attribution 3.0 License

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.