Remove dependency on nova test modules

Replace all nova test module calls with stubs or mocks to avoid
issues caused by changes in the nova test framework or nova itself.

Change-Id: I07248b64cc5c30c90c5e68df09ae8dfc2875c279
Signed-off-by: Doug Hellmann <doug.hellmann@dreamhost.com>

ceilometer/tests/base.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+#
+# Copyright © 2012 New Dream Network (DreamHost)
+#
+# Author: Doug Hellmann <doug.hellmann@dreamhost.com>
+#
+# 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()

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

tests/collector/__init__.py

@@ -1 +0,0 @@
-from nova.tests import *

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()
-

tests/compute/__init__.py

@@ -1 +0,0 @@
-from nova.tests import *

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

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

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

tests/network/__init__.py

@@ -1 +0,0 @@
-from nova.tests import *

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']

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

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

tools/test-requires

@@ -1,6 +1,7 @@
 nose
 coverage
 pep8>=1.0
+mock
 mox
 glance>=2011.3.1
 python-glanceclient