Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# 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.
"""Tests for ceilometer/agent/manager.py
"""
import mock
from ceilometer.compute import manager
from ceilometer import nova_client
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import test
from tests import agentbase
class TestManager(test.BaseTestCase):
@mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
def test_load_plugins(self):
mgr = manager.AgentManager()
self.assertIsNotNone(list(mgr.pollster_manager))
class TestRunTasks(agentbase.BaseAgentManagerTestCase):
def _fake_instance(self, name, state):
instance = mock.MagicMock()
instance.name = name
setattr(instance, 'OS-EXT-STS:vm_state', state)
return instance
def _raise_exception(self):
'x-account-container-count': 7,
}),
('tenant-001', {'x-account-object-count': 34,
'x-account-bytes-used': 9898989898,
'x-account-container-count': 17,
})]
class TestManager(manager.AgentManager):
def __init__(self):
super(TestManager, self).__init__()
self.keystone = mock.MagicMock()
class TestSwiftPollster(test.BaseTestCase):
# Define scenarios to run all of the tests against all of the
# pollsters.
scenarios = [
('storage.objects',
{'factory': swift.ObjectsPollster}),
('storage.objects.size',
{'factory': swift.ObjectsSizePollster}),
('storage.objects.containers',
{'factory': swift.ObjectsContainersPollster}),
]
@staticmethod
def fake_ks_service_catalog_url_for(*args, **kwargs):
raise exceptions.EndpointNotFound("Fake keystone exception")
# 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.
from ceilometer.openstack.common import test
from ceilometer.storage import models
class FakeModel(models.Model):
def __init__(self, arg1, arg2):
models.Model.__init__(self, arg1=arg1, arg2=arg2)
class ModelTest(test.BaseTestCase):
def test_create_attributes(self):
m = FakeModel(1, 2)
self.assertEqual(m.arg1, 1)
self.assertEqual(m.arg2, 2)
def test_as_dict(self):
m = FakeModel(1, 2)
d = m.as_dict()
self.assertEqual(d, {'arg1': 1, 'arg2': 2})
def test_as_dict_recursive(self):
m = FakeModel(1, FakeModel('a', 'b'))
d = m.as_dict()
self.assertEqual(d, {'arg1': 1,
'arg2': {'arg1': 'a',
return os.getenv('OS_TEST_DBAPI_CONNECTION', 'sqlite://')
def __init__(self, test):
super(DbFixture, self).__init__()
self.test = test
def setUp(self):
super(DbFixture, self).setUp()
self.test.engine = session.create_engine(self._get_uri())
self.test.sessionmaker = session.get_maker(self.test.engine)
self.addCleanup(self.test.engine.dispose)
class DbTestCase(test.BaseTestCase):
"""Base class for testing of DB code.
Using `DbFixture`. Intended to be the main database test case to use all
the tests on a given backend with user defined uri. Backend specific
tests should be decorated with `backend_specific` decorator.
"""
FIXTURE = DbFixture
def setUp(self):
super(DbTestCase, self).setUp()
self.useFixture(self.FIXTURE(self))
ALLOWED_DIALECTS = ['sqlite', 'mysql', 'postgresql']
class FakeApp(object):
def __init__(self, body=['This string is 28 bytes long']):
self.body = body
def __call__(self, env, start_response):
start_response('200 OK', [
('Content-Type', 'text/plain'),
('Content-Length', str(sum(map(len, self.body))))
])
while env['wsgi.input'].read(5):
pass
return self.body
class TestSwiftMiddleware(test.BaseTestCase):
class _faux_pipeline_manager(pipeline.PipelineManager):
class _faux_pipeline(object):
def __init__(self, pipeline_manager):
self.pipeline_manager = pipeline_manager
self.samples = []
def publish_samples(self, ctxt, samples):
self.samples.extend(samples)
def flush(self, context):
pass
def __init__(self):
self.pipelines = [self._faux_pipeline(self)]
# 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.
"""Tests for libvirt inspector.
"""
from ceilometer.compute.virt import inspector as virt_inspector
from ceilometer.compute.virt.libvirt import inspector as libvirt_inspector
import fixtures
from ceilometer.openstack.common import test
from ceilometer.openstack.common.fixture import moxstubout
class TestLibvirtInspection(test.BaseTestCase):
def setUp(self):
super(TestLibvirtInspection, self).setUp()
self.instance_name = 'instance-00000001'
self.inspector = libvirt_inspector.LibvirtInspector()
self.mox = self.useFixture(moxstubout.MoxStubout()).mox
self.inspector.connection = self.mox.CreateMockAnything()
self.domain = self.mox.CreateMockAnything()
def test_inspect_instances(self):
class FakeDomain(object):
def name(self):
return 'fake_name'
def UUIDString(self):
return 'uuid'
# 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.
import mock
from ceilometer.openstack.common import test
class TestPollsterBase(test.BaseTestCase):
def setUp(self):
super(TestPollsterBase, self).setUp()
self.addCleanup(mock.patch.stopall)
self.inspector = mock.Mock()
self.instance = mock.MagicMock()
self.instance.name = 'instance-00000001'
setattr(self.instance, 'OS-EXT-SRV-ATTR:instance_name',
self.instance.name)
self.instance.id = 1
self.instance.flavor = {'name': 'm1.small', 'id': 2, 'vcpus': 1,
'ram': 512, 'disk': 20, 'ephemeral': 0}
patch_virt = mock.patch('ceilometer.compute.virt.inspector'
u'_context_remote_address': u'192.168.22.101',
u'publisher_id': u'volume.ubuntu-VirtualBox',
u'payload': {u'status': u'extending',
u'volume_type_id': None,
u'display_name': u'abc',
u'tenant_id': u'6c97f1ecf17047eab696786d56a0bff5',
u'created_at': u'2012-09-21 10:10:47',
u'snapshot_id': None,
u'volume_id': u'3b761164-84b4-4eb3-8fcb-1974c641d6ef',
u'user_id': u'4d2fa4b76a4a4ecab8c468c8dea42f89',
u'launched_at': u'2012-09-21 10:10:50',
u'size': 3},
u'priority': u'INFO'}
class TestNotifications(test.BaseTestCase):
def _verify_common_sample(self, s, name, notification):
self.assertFalse(s is None)
self.assertEqual(s.name, name)
self.assertEqual(s.resource_id, notification['payload']['volume_id'])
self.assertEqual(s.timestamp, notification['timestamp'])
metadata = s.resource_metadata
self.assertEqual(metadata.get('host'), notification['publisher_id'])
def test_volume_exists(self):
v = notifications.Volume()
samples = list(v.process_notification(NOTIFICATION_VOLUME_EXISTS))
self.assertEqual(len(samples), 1)
s = samples[0]
self._verify_common_sample(s, 'volume', NOTIFICATION_VOLUME_EXISTS)
self.assertEqual(s.volume, 1)
# 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.
"""class for tests in ceilometer/alarm/evaluator/__init__.py
"""
import mock
from ceilometer.alarm import evaluator
from ceilometer.openstack.common import test
class TestEvaluatorBaseClass(test.BaseTestCase):
def setUp(self):
super(TestEvaluatorBaseClass, self).setUp()
self.called = False
def _notify(self, alarm, previous, reason):
self.called = True
raise Exception('Boom!')
def test_base_refresh(self):
notifier = mock.MagicMock()
notifier.notify = self._notify
class EvaluatorSub(evaluator.Evaluator):
def evaluate(self, alarm):
pass
#
# 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.
import mock
import novaclient
from ceilometer import nova_client
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import test
class TestNovaClient(test.BaseTestCase):
def setUp(self):
super(TestNovaClient, self).setUp()
self.nv = nova_client.Client()
self.stubs = self.useFixture(moxstubout.MoxStubout()).stubs
self.stubs.Set(self.nv.nova_client.flavors, 'get',
self.fake_flavors_get)
self.stubs.Set(self.nv.nova_client.images, 'get', self.fake_images_get)
@staticmethod
def fake_flavors_get(*args, **kwargs):
a = mock.MagicMock()
a.id = args[0]
if a.id == 1:
a.name = 'm1.tiny'
elif a.id == 2: