Merge "Replace assert statements with assert methods"

ceilometer/tests/api/v2/test_compute_duration_by_resource_scenarios.py

@@ -54,8 +54,8 @@ class TestComputeDurationByResource(FunctionalTest,
 
     def _patch_get_interval(self, start, end):
         def get_interval(event_filter, period, groupby, aggregate):
-            assert event_filter.start
+            self.assertIsNotNone(event_filter.start)
-            assert event_filter.end
+            self.assertIsNotNone(event_filter.end)
             if (event_filter.start > end or event_filter.end < start):
                 return []
             duration_start = max(event_filter.start, start)

ceilometer/tests/api/v2/test_list_meters_scenarios.py

@@ -527,7 +527,7 @@ class TestListMeters(FunctionalTest,
                                  'value': 'test_source_doesnt_exist',
                                  }],
                              )
-        assert not data
+        self.assertIsEmpty(data)
 
     def test_with_user(self):
         data = self.get_json('/meters',

ceilometer/tests/base.py

@@ -41,6 +41,20 @@ class BaseTestCase(test.BaseTestCase):
             0.0,
             places=5)
 
+    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))
+
+    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))
+
     @staticmethod
     def path_get(project_file=None):
         root = os.path.abspath(os.path.join(os.path.dirname(__file__),

ceilometer/tests/compute/pollsters/base.py

@@ -20,10 +20,10 @@
 
 import mock
 
-from ceilometer.openstack.common import test
+import ceilometer.tests.base as base
 
 
-class TestPollsterBase(test.BaseTestCase):
+class TestPollsterBase(base.BaseTestCase):
 
     def setUp(self):
         super(TestPollsterBase, self).setUp()

ceilometer/tests/compute/pollsters/test_diskio.py

@@ -46,7 +46,7 @@ class TestDiskPollsters(base.TestPollsterBase):
         mgr = manager.AgentManager()
         cache = {}
         samples = list(pollster.get_samples(mgr, cache, [self.instance]))
-        assert samples
+        self.assertIsNotEmpty(samples)
         self.assertIn(pollster.CACHE_KEY_DISK, cache)
         self.assertIn(self.instance.name, cache[pollster.CACHE_KEY_DISK])
 
@@ -96,7 +96,7 @@ class TestDiskRatePollsters(base.TestPollsterBase):
         mgr = manager.AgentManager()
         cache = {}
         samples = list(pollster.get_samples(mgr, cache, [self.instance]))
-        assert samples
+        self.assertIsNotEmpty(samples)
         self.assertIsNotNone(samples)
         self.assertIn(pollster.CACHE_KEY_DISK_RATE, cache)
         self.assertIn(self.instance.id, cache[pollster.CACHE_KEY_DISK_RATE])

ceilometer/tests/dispatcher/test_db.py

@@ -68,8 +68,8 @@ class TestDispatcherDB(test.BaseTestCase):
 
         self.dispatcher.record_metering_data(msg)
 
-        assert not self.dispatcher.storage_conn.called, \
+        if self.dispatcher.storage_conn.called:
-            'Should not have called the storage connection'
+            self.fail('Should not have called the storage connection')
 
     def test_timestamp_conversion(self):
         msg = {'counter_name': 'test',

ceilometer/tests/network/test_notifications.py

@@ -19,7 +19,7 @@
 """
 
 from ceilometer.network import notifications
-from ceilometer.openstack.common import test
+from ceilometer.tests import base as test
 
 NOTIFICATION_NETWORK_CREATE = {
     u'_context_roles': [u'anotherrole',
@@ -307,17 +307,17 @@ class TestEventTypes(test.BaseTestCase):
     def test_network(self):
         v = notifications.Network()
         events = v.event_types
-        assert events
+        self.assertIsNotEmpty(events)
 
     def test_subnet(self):
         v = notifications.Subnet()
         events = v.event_types
-        assert events
+        self.assertIsNotEmpty(events)
 
     def test_port(self):
         v = notifications.Port()
         events = v.event_types
-        assert events
+        self.assertIsNotEmpty(events)
 
     def test_router(self):
         self.assertTrue(notifications.Router().event_types)

ceilometer/tests/storage/test_storage_scenarios.py

@@ -174,7 +174,7 @@ class ResourceTest(DBTestBase,
             self.assertEqual(resource.metadata['display_name'], 'test-server')
             break
         else:
-            assert False, 'Never found resource-id'
+            self.fail('Never found resource-id')
 
     def test_get_resources_start_timestamp(self):
         timestamp = datetime.datetime(2012, 7, 2, 10, 42)
@@ -425,7 +425,7 @@ class MeterTest(DBTestBase,
     def test_get_meters_by_metaquery(self):
         q = {'metadata.display_name': 'test-server'}
         results = list(self.conn.get_meters(metaquery=q))
-        assert results
+        self.assertIsNotEmpty(results)
         self.assertEqual(len(results), 9)
 
     def test_get_meters_by_empty_metaquery(self):
@@ -542,7 +542,7 @@ class RawSampleTest(DBTestBase,
     def test_get_samples_by_resource(self):
         f = storage.SampleFilter(user='user-id', resource='resource-id')
         results = list(self.conn.get_samples(f))
-        assert results
+        self.assertIsNotEmpty(results)
         meter = results[1]
         d = meter.as_dict()
         self.assertEqual(d['recorded_at'], timeutils.utcnow())
@@ -640,12 +640,12 @@ class RawSampleTest(DBTestBase,
     def test_get_samples_by_name(self):
         f = storage.SampleFilter(user='user-id', meter='no-such-meter')
         results = list(self.conn.get_samples(f))
-        assert not results
+        self.assertIsEmpty(results)
 
     def test_get_samples_by_name2(self):
         f = storage.SampleFilter(user='user-id', meter='instance')
         results = list(self.conn.get_samples(f))
-        assert results
+        self.assertIsNotEmpty(results)
 
     def test_get_samples_by_source(self):
         f = storage.SampleFilter(source='test-1')