diff --git a/ceilometer/tests/base.py b/ceilometer/tests/base.py new file mode 100644 index 0000000000..8613d459c2 --- /dev/null +++ b/ceilometer/tests/base.py @@ -0,0 +1,40 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +# +# Copyright © 2012 New Dream Network (DreamHost) +# +# Author: Doug Hellmann +# +# 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 unittest + +import mox +import stubout + + +class TestCase(unittest.TestCase): + + def setUp(self): + super(TestCase, self).setUp() + self.mox = mox.Mox() + self.stubs = stubout.StubOutForTesting() + + def tearDown(self): + self.mox.UnsetStubs() + self.stubs.UnsetAll() + self.stubs.SmartUnsetAll() + self.mox.VerifyAll() + super(TestCase, self).tearDown() diff --git a/ceilometer/tests/skip.py b/ceilometer/tests/skip.py new file mode 100644 index 0000000000..643a7906ef --- /dev/null +++ b/ceilometer/tests/skip.py @@ -0,0 +1,86 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +"""Base classes for our unit tests. + +Allows overriding of flags for use of fakes, and some black magic for +inline callbacks. + +""" + +import functools +import unittest + +import nose.plugins.skip + + +class skip_test(object): + """Decorator that skips a test.""" + # TODO(tr3buchet): remember forever what comstud did here + def __init__(self, msg): + self.message = msg + + def __call__(self, func): + @functools.wraps(func) + def _skipper(*args, **kw): + """Wrapped skipper function.""" + raise nose.SkipTest(self.message) + return _skipper + + +class skip_if(object): + """Decorator that skips a test if condition is true.""" + def __init__(self, condition, msg): + self.condition = condition + self.message = msg + + def __call__(self, func): + @functools.wraps(func) + def _skipper(*args, **kw): + """Wrapped skipper function.""" + if self.condition: + raise nose.SkipTest(self.message) + func(*args, **kw) + return _skipper + + +class skip_unless(object): + """Decorator that skips a test if condition is not true.""" + def __init__(self, condition, msg): + self.condition = condition + self.message = msg + + def __call__(self, func): + @functools.wraps(func) + def _skipper(*args, **kw): + """Wrapped skipper function.""" + if not self.condition: + raise nose.SkipTest(self.message) + func(*args, **kw) + return _skipper + + +def skip_if_fake(func): + """Decorator that skips a test if running in fake mode.""" + def _skipper(*args, **kw): + """Wrapped skipper function.""" + if FLAGS.fake_tests: + raise unittest.SkipTest('Test cannot be run in fake mode') + else: + return func(*args, **kw) + return _skipper diff --git a/tests/collector/__init__.py b/tests/collector/__init__.py index 70bd003562..e69de29bb2 100644 --- a/tests/collector/__init__.py +++ b/tests/collector/__init__.py @@ -1 +0,0 @@ -from nova.tests import * diff --git a/tests/collector/test_manager.py b/tests/collector/test_manager.py index 721e2769e0..e257acc755 100644 --- a/tests/collector/test_manager.py +++ b/tests/collector/test_manager.py @@ -20,20 +20,18 @@ import datetime -from nova import context -from nova import test - from ceilometer import meter from ceilometer.collector import manager from ceilometer.storage import base +from ceilometer.tests import base as tests_base -class TestCollectorManager(test.TestCase): +class TestCollectorManager(tests_base.TestCase): def setUp(self): super(TestCollectorManager, self).setUp() self.mgr = manager.CollectorManager() - self.ctx = context.RequestContext("user", "project") + self.ctx = None def test_valid_message(self): msg = {'counter_name': 'test', @@ -88,4 +86,3 @@ class TestCollectorManager(test.TestCase): self.mgr.record_metering_data(self.ctx, msg) self.mox.VerifyAll() - diff --git a/tests/compute/__init__.py b/tests/compute/__init__.py index 70bd003562..e69de29bb2 100644 --- a/tests/compute/__init__.py +++ b/tests/compute/__init__.py @@ -1 +0,0 @@ -from nova.tests import * diff --git a/tests/compute/test_instance.py b/tests/compute/test_instance.py index 5e0d321cd3..bf6ed44981 100644 --- a/tests/compute/test_instance.py +++ b/tests/compute/test_instance.py @@ -18,15 +18,31 @@ """Tests for ceilometer.compute.instance """ -from nova import context -from nova import test -from nova import db +import unittest + +import mock from ceilometer.compute import instance from ceilometer.compute import manager -class TestLocationMetadata(test.TestCase): +class FauxInstance(object): + + def __init__(self, **kwds): + for name, value in kwds.items(): + setattr(self, name, value) + + def __getitem__(self, key): + return getattr(self, key) + + def get(self, key, default): + try: + return getattr(self, key) + except AttributeError: + return default + + +class TestLocationMetadata(unittest.TestCase): INSTANCE_PROPERTIES = {'display_name': 'display name', 'reservation_id': 'reservation id', @@ -45,13 +61,18 @@ class TestLocationMetadata(test.TestCase): } def setUp(self): - self.context = context.RequestContext('admin', 'admin', is_admin=True) self.manager = manager.AgentManager() super(TestLocationMetadata, self).setUp() - self.instance = db.instance_create(self.context, - self.INSTANCE_PROPERTIES) + self.instance = FauxInstance(**self.INSTANCE_PROPERTIES) + self.instance.host = 'made-up-hostname' + m = mock.MagicMock() + m.flavorid = 1 + self.instance.instance_type = m def test_metadata(self): md = instance.get_metadata_from_dbobject(self.instance) for name in self.INSTANCE_PROPERTIES.keys(): - assert md[name] == self.INSTANCE_PROPERTIES[name] + actual = md[name] + print 'checking', name, actual + expected = self.INSTANCE_PROPERTIES[name] + assert actual == expected diff --git a/tests/compute/test_libvirt.py b/tests/compute/test_libvirt.py index 27cb2ce585..eabba08ffe 100644 --- a/tests/compute/test_libvirt.py +++ b/tests/compute/test_libvirt.py @@ -19,6 +19,8 @@ """Tests for manager. """ +import unittest + try: import libvirt as ignored_libvirt except ImportError: @@ -26,45 +28,47 @@ except ImportError: else: libvirt_missing = False -from nova import context +import mock + from nova import flags -from nova import test -from nova import db from ceilometer.compute import libvirt from ceilometer.compute import manager +from ceilometer.tests import skip -class TestDiskIOPollster(test.TestCase): +class TestDiskIOPollster(unittest.TestCase): def setUp(self): - self.context = context.RequestContext('admin', 'admin', is_admin=True) self.manager = manager.AgentManager() self.pollster = libvirt.DiskIOPollster() super(TestDiskIOPollster, self).setUp() - self.instance = db.instance_create(self.context, {}) + self.instance = mock.MagicMock() + self.instance.name = 'instance-00000001' + self.instance.id = 1 flags.FLAGS.compute_driver = 'libvirt.LibvirtDriver' + flags.FLAGS.connection_type = 'libvirt' - @test.skip_if(libvirt_missing, 'Test requires libvirt') + @skip.skip_if(libvirt_missing, 'Test requires libvirt') def test_fetch_diskio(self): - counters = list(self.pollster.get_counters(self.manager, - self.instance)) + list(self.pollster.get_counters(self.manager, self.instance)) #assert counters # FIXME(dhellmann): The CI environment doesn't produce # a response when the fake driver asks for the disks, so # we do not get any counters in response. - @test.skip_if(libvirt_missing, 'Test requires libvirt') + @skip.skip_if(libvirt_missing, 'Test requires libvirt') def test_fetch_diskio_not_libvirt(self): flags.FLAGS.compute_driver = 'fake.FakeDriver' + flags.FLAGS.connection_type = 'fake' counters = list(self.pollster.get_counters(self.manager, self.instance)) assert not counters - @test.skip_if(libvirt_missing, 'Test requires libvirt') + @skip.skip_if(libvirt_missing, 'Test requires libvirt') def test_fetch_diskio_with_libvirt_non_existent_instance(self): - print 'ID:', self.instance.id - inst = db.instance_get(self.context, self.instance.id) - inst.id = 999 # change the id so the driver cannot find the instance - counters = list(self.pollster.get_counters(self.manager, inst)) + instance = mock.MagicMock() + instance.name = 'instance-00000999' + instance.id = 999 + counters = list(self.pollster.get_counters(self.manager, instance)) assert not counters diff --git a/tests/compute/test_manager.py b/tests/compute/test_manager.py index 261c1482a1..2756ce3fd7 100644 --- a/tests/compute/test_manager.py +++ b/tests/compute/test_manager.py @@ -20,12 +20,10 @@ import datetime -from nova import context -from nova import test - from ceilometer.compute import manager from ceilometer import counter from ceilometer import publish +from ceilometer.tests import base def test_load_plugins(): @@ -35,7 +33,7 @@ def test_load_plugins(): return -class TestRunTasks(test.TestCase): +class TestRunTasks(base.TestCase): class Pollster: counters = [] @@ -66,20 +64,19 @@ class TestRunTasks(test.TestCase): self.stubs.Set(publish, 'publish_counter', self.faux_notify) self.mgr = manager.AgentManager() self.mgr.pollsters = [('test', self.Pollster())] - self.ctx = context.RequestContext("user", "project") # Set up a fake instance value to be returned by # instance_get_all_by_host() so when the manager gets the list # of instances to poll we can control the results. self.instance = 'faux instance' self.mox.StubOutWithMock(self.mgr.db, 'instance_get_all_by_host') self.mgr.db.instance_get_all_by_host( - self.ctx, + None, self.mgr.host, ).AndReturn([self.instance]) self.mox.ReplayAll() # Invoke the periodic tasks to call the pollsters. - self.mgr.periodic_tasks(self.ctx) + self.mgr.periodic_tasks(None) def test_message(self): assert self.Pollster.counters[0][1] is self.instance diff --git a/tests/network/__init__.py b/tests/network/__init__.py index 70bd003562..e69de29bb2 100644 --- a/tests/network/__init__.py +++ b/tests/network/__init__.py @@ -1 +0,0 @@ -from nova.tests import * diff --git a/tests/network/test_floatingip.py b/tests/network/test_floatingip.py index 7878bb6a21..b8b64da7f2 100644 --- a/tests/network/test_floatingip.py +++ b/tests/network/test_floatingip.py @@ -17,46 +17,48 @@ # License for the specific language governing permissions and limitations # under the License. +import mock + from nova import context from nova import db -from nova import exception -from nova import test from ceilometer.network import floatingip from ceilometer.central import manager +from ceilometer.tests import base -class TestFloatingIPPollster(test.TestCase): +class TestFloatingIPPollster(base.TestCase): def setUp(self): + super(TestFloatingIPPollster, self).setUp() self.context = context.RequestContext('admin', 'admin', is_admin=True) self.manager = manager.AgentManager() self.pollster = floatingip.FloatingIPPollster() - super(TestFloatingIPPollster, self).setUp() + self.stubs.Set(db, 'floating_ip_get_all', self.faux_get_ips) - def test_get_counters(self): - try: - list(self.pollster.get_counters(self.manager, - self.context) - ) - except exception.NoFloatingIpsDefined: - pass - else: - assert False, 'Should have seen an error' + def faux_get_ips(self, context): + ips = [] + for i in range(1, 4): + ip = mock.MagicMock() + ip.address = '1.1.1.%d' % i + ip.host = self.manager.host + ips.append(ip) + return ips + + # FIXME(dhellmann): Is there a useful way to define this + # test without a database? + # + # def test_get_counters_none_defined(self): + # try: + # list(self.pollster.get_counters(self.manager, + # self.context) + # ) + # except exception.NoFloatingIpsDefined: + # pass + # else: + # assert False, 'Should have seen an error' def test_get_counters_not_empty(self): - db.floating_ip_create(self.context, - {'address': '1.1.1.1', - 'host': self.manager.host, - }) - db.floating_ip_create(self.context, - {'address': '1.1.1.2', - 'host': self.manager.host + "randomstring", - }) - db.floating_ip_create(self.context, - {'address': '1.1.1.3', - 'host': self.manager.host + "randomstring", - }) counters = list(self.pollster.get_counters(self.manager, self.context)) self.assertEqual(len(counters), 3) addresses = [c.resource_metadata['address'] diff --git a/tests/storage/test_register_opts.py b/tests/storage/test_register_opts.py index 129f04362d..080ab2b22c 100644 --- a/tests/storage/test_register_opts.py +++ b/tests/storage/test_register_opts.py @@ -18,14 +18,13 @@ """Tests for ceilometer/storage/ """ -from nova import test - from ceilometer import storage from ceilometer.storage import base from ceilometer.openstack.common import cfg +from ceilometer.tests import base as test_base -class RegisterOpts(test.TestCase): +class RegisterOpts(test_base.TestCase): def faux_get_engine(self, conf): return self._faux_engine diff --git a/tests/test_publish.py b/tests/test_publish.py index 278ba32703..ff65fb006d 100644 --- a/tests/test_publish.py +++ b/tests/test_publish.py @@ -19,16 +19,16 @@ """ import datetime +import unittest -from nova import context from ceilometer.openstack.common import rpc -from nova import test +from ceilometer.tests import base from ceilometer import counter from ceilometer import publish -class TestPublish(test.TestCase): +class TestPublish(base.TestCase): test_data = counter.Counter( source='test', @@ -51,8 +51,7 @@ class TestPublish(test.TestCase): super(TestPublish, self).setUp() self.notifications = [] self.stubs.Set(rpc, 'cast', self.faux_notify) - self.ctx = context.RequestContext("user", "project") - publish.publish_counter(self.ctx, self.test_data) + publish.publish_counter(None, self.test_data) def test_notify(self): assert len(self.notifications) == 2 diff --git a/tools/test-requires b/tools/test-requires index 287e1800a3..95f6b542d0 100644 --- a/tools/test-requires +++ b/tools/test-requires @@ -1,6 +1,7 @@ nose coverage pep8>=1.0 +mock mox glance>=2011.3.1 python-glanceclient