openstack/aodh
Code Issues Proposed changes

Merge "Fix order of arguments in assertEquals"

Browse Source
This commit is contained in:
Jenkins 2014-03-21 07:03:16 +00:00 committed by Gerrit Code Review
commit 67989e73f1
10 changed files with 249 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
ceilometer/tests/api/v2/test_event_scenarios.py
View File

@ -93,28 +93,28 @@ class TestTraitAPI(EventTestBase):
path = self.PATH % "NO_SUCH_EVENT_TYPE" path = self.PATH % "NO_SUCH_EVENT_TYPE"
data = self.get_json(path, headers=headers) data = self.get_json(path, headers=headers)
self.assertEqual(data, []) self.assertEqual([], data)
def test_get_trait_data_for_event(self): def test_get_trait_data_for_event(self):
path = (self.PATH % "Foo") + "/trait_A" path = (self.PATH % "Foo") + "/trait_A"
data = self.get_json(path, headers=headers) data = self.get_json(path, headers=headers)
self.assertEqual(len(data), 1) self.assertEqual(1, len(data))
trait = data[0] trait = data[0]
self.assertEqual(trait['name'], "trait_A") self.assertEqual("trait_A", trait['name'])
def test_get_trait_data_for_non_existent_event(self): def test_get_trait_data_for_non_existent_event(self):
path = (self.PATH % "NO_SUCH_EVENT") + "/trait_A" path = (self.PATH % "NO_SUCH_EVENT") + "/trait_A"
data = self.get_json(path, headers=headers) data = self.get_json(path, headers=headers)
self.assertEqual(data, []) self.assertEqual([], data)
def test_get_trait_data_for_non_existent_trait(self): def test_get_trait_data_for_non_existent_trait(self):
path = (self.PATH % "Foo") + "/no_such_trait" path = (self.PATH % "Foo") + "/no_such_trait"
data = self.get_json(path, headers=headers) data = self.get_json(path, headers=headers)
self.assertEqual(data, []) self.assertEqual([], data)
class TestEventAPI(EventTestBase): class TestEventAPI(EventTestBase):
@ -123,7 +123,7 @@ class TestEventAPI(EventTestBase):
def test_get_events(self): def test_get_events(self):
data = self.get_json(self.PATH, headers=headers) data = self.get_json(self.PATH, headers=headers)
self.assertEqual(len(data), 3) self.assertEqual(3, len(data))
# We expect to get native UTC generated time back # We expect to get native UTC generated time back
expected_generated = timeutils.strtime( expected_generated = timeutils.strtime(
at=timeutils.normalize_time(self.trait_time), at=timeutils.normalize_time(self.trait_time),
@ -131,7 +131,7 @@ class TestEventAPI(EventTestBase):
for event in data: for event in data:
self.assertTrue(event['event_type'] in ['Foo', 'Bar', 'Zoo']) self.assertTrue(event['event_type'] in ['Foo', 'Bar', 'Zoo'])
self.assertEqual(4, len(event['traits'])) self.assertEqual(4, len(event['traits']))
self.assertEqual(event['generated'], expected_generated) self.assertEqual(expected_generated, event['generated'])
for trait_name in ['trait_A', 'trait_B', for trait_name in ['trait_A', 'trait_B',
'trait_C', 'trait_D']: 'trait_C', 'trait_D']:
self.assertTrue(trait_name in map(lambda x: x['name'], self.assertTrue(trait_name in map(lambda x: x['name'],
@ -173,7 +173,7 @@ class TestEventAPI(EventTestBase):
'value': 'my_Foo_text', 'value': 'my_Foo_text',
'type': 'string'}]) 'type': 'string'}])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(data[0]['event_type'], 'Foo') self.assertEqual('Foo', data[0]['event_type'])
def test_get_events_filter_int_trait(self): def test_get_events_filter_int_trait(self):
data = self.get_json(self.PATH, headers=headers, data = self.get_json(self.PATH, headers=headers,
@ -181,14 +181,12 @@ class TestEventAPI(EventTestBase):
'value': '101', 'value': '101',
'type': 'integer'}]) 'type': 'integer'}])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(data[0]['event_type'], 'Bar') self.assertEqual('Bar', data[0]['event_type'])
traits = filter(lambda x: x['name'] == 'trait_B', data[0]['traits']) traits = filter(lambda x: x['name'] == 'trait_B', data[0]['traits'])
self.assertEqual(1, len(traits)) self.assertEqual(1, len(traits))
self.assertEqual(traits[0]['type'], self.assertEqual('integer', traits[0]['type'])
'integer') self.assertEqual('101', traits[0]['value'])
self.assertEqual(traits[0]['value'],
'101')
def test_get_events_filter_float_trait(self): def test_get_events_filter_float_trait(self):
data = self.get_json(self.PATH, headers=headers, data = self.get_json(self.PATH, headers=headers,
@ -196,14 +194,12 @@ class TestEventAPI(EventTestBase):
'value': '200.123456', 'value': '200.123456',
'type': 'float'}]) 'type': 'float'}])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(data[0]['event_type'], 'Zoo') self.assertEqual('Zoo', data[0]['event_type'])
traits = filter(lambda x: x['name'] == 'trait_C', data[0]['traits']) traits = filter(lambda x: x['name'] == 'trait_C', data[0]['traits'])
self.assertEqual(1, len(traits)) self.assertEqual(1, len(traits))
self.assertEqual(traits[0]['type'], self.assertEqual('float', traits[0]['type'])
'float') self.assertEqual('200.123456', traits[0]['value'])
self.assertEqual(traits[0]['value'],
'200.123456')
def test_get_events_filter_datetime_trait(self): def test_get_events_filter_datetime_trait(self):
data = self.get_json(self.PATH, headers=headers, data = self.get_json(self.PATH, headers=headers,
@ -213,10 +209,8 @@ class TestEventAPI(EventTestBase):
self.assertEqual(3, len(data)) self.assertEqual(3, len(data))
traits = filter(lambda x: x['name'] == 'trait_D', data[0]['traits']) traits = filter(lambda x: x['name'] == 'trait_D', data[0]['traits'])
self.assertEqual(1, len(traits)) self.assertEqual(1, len(traits))
self.assertEqual(traits[0]['type'], self.assertEqual('datetime', traits[0]['type'])
'datetime') self.assertEqual(self.trait_time.isoformat(), traits[0]['value'])
self.assertEqual(traits[0]['value'],
self.trait_time.isoformat())
def test_get_events_multiple_filters(self): def test_get_events_multiple_filters(self):
data = self.get_json(self.PATH, headers=headers, data = self.get_json(self.PATH, headers=headers,
@ -227,7 +221,7 @@ class TestEventAPI(EventTestBase):
'value': 'my_Foo_text', 'value': 'my_Foo_text',
'type': 'string'}]) 'type': 'string'}])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(data[0]['event_type'], 'Foo') self.assertEqual('Foo', data[0]['event_type'])
def test_get_events_multiple_filters_no_matches(self): def test_get_events_multiple_filters_no_matches(self):
data = self.get_json(self.PATH, headers=headers, data = self.get_json(self.PATH, headers=headers,

8
ceilometer/tests/api/v2/test_list_events_scenarios.py
View File

@ -88,8 +88,8 @@ class TestListEvents(FunctionalTest,
data = self.get_json('/meters/instance') data = self.get_json('/meters/instance')
self.assertEqual(2, len(data)) self.assertEqual(2, len(data))
for s in data: for s in data:
self.assertEqual(s['recorded_at'], self.assertEqual(timeutils.utcnow.override_time.isoformat(),
timeutils.utcnow.override_time.isoformat()) s['recorded_at'])
def test_all_trailing_slash(self): def test_all_trailing_slash(self):
data = self.get_json('/meters/instance/') data = self.get_json('/meters/instance/')
@ -172,9 +172,9 @@ class TestListEvents(FunctionalTest,
sample = data[0] sample = data[0]
self.assertIn('resource_metadata', sample) self.assertIn('resource_metadata', sample)
self.assertEqual( self.assertEqual(
list(sorted(sample['resource_metadata'].iteritems())),
[('dict_properties.key', 'value'), [('dict_properties.key', 'value'),
('display_name', 'test-server'), ('display_name', 'test-server'),
('not_ignored_list', "['returned']"), ('not_ignored_list', "['returned']"),
('tag', 'self.sample'), ('tag', 'self.sample'),
]) ],
list(sorted(sample['resource_metadata'].iteritems())))

174
ceilometer/tests/api/v2/test_list_meters_scenarios.py
View File

@ -164,15 +164,15 @@ class TestListMeters(FunctionalTest,
def test_list_meters(self): def test_list_meters(self):
data = self.get_json('/meters') data = self.get_json('/meters')
self.assertEqual(4, len(data)) self.assertEqual(4, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id',
set(['resource-id',
'resource-id2', 'resource-id2',
'resource-id3', 'resource-id3',
'resource-id4'])) 'resource-id4']),
self.assertEqual(set(r['name'] for r in data), set(r['resource_id'] for r in data))
set(['meter.test', 'meter.mine'])) self.assertEqual(set(['meter.test', 'meter.mine']),
self.assertEqual(set(r['source'] for r in data), set(r['name'] for r in data))
set(['test_source', 'test_source1'])) self.assertEqual(set(['test_source', 'test_source1']),
set(r['source'] for r in data))
def test_meters_query_with_timestamp(self): def test_meters_query_with_timestamp(self):
date_time = datetime.datetime(2012, 7, 2, 10, 41) date_time = datetime.datetime(2012, 7, 2, 10, 41)
@ -182,11 +182,11 @@ class TestListMeters(FunctionalTest,
'op': 'gt', 'op': 'gt',
'value': isotime}], 'value': isotime}],
expect_errors=True) expect_errors=True)
self.assertEqual(resp.status_code, 400) self.assertEqual(400, resp.status_code)
self.assertEqual(jsonutils.loads(resp.body)['error_message'] self.assertEqual('Unknown argument: "timestamp": '
['faultstring'], 'not valid for this resource',
'Unknown argument: "timestamp": ' jsonutils.loads(resp.body)['error_message']
'not valid for this resource') ['faultstring'])
def test_list_samples(self): def test_list_samples(self):
resp = self.get_json('/samples', resp = self.get_json('/samples',
@ -194,12 +194,12 @@ class TestListMeters(FunctionalTest,
'op': 'eq', 'op': 'eq',
'value': 42}], 'value': 42}],
expect_errors=True) expect_errors=True)
self.assertEqual(resp.status_code, 400) self.assertEqual(400, resp.status_code)
self.assertEqual(jsonutils.loads(resp.body)['error_message'] self.assertEqual("Invalid input for field/attribute field. "
['faultstring'],
"Invalid input for field/attribute field. "
"Value: 'search_offset'. " "Value: 'search_offset'. "
"search_offset cannot be used without timestamp") "search_offset cannot be used without timestamp",
jsonutils.loads(resp.body)['error_message']
['faultstring'])
def test_query_samples_with_search_offset(self): def test_query_samples_with_search_offset(self):
data = self.get_json('/samples') data = self.get_json('/samples')
@ -224,7 +224,7 @@ class TestListMeters(FunctionalTest,
data = self.get_json('/samples/%s' % sample_id) data = self.get_json('/samples/%s' % sample_id)
self.assertIn('id', data) self.assertIn('id', data)
del data['recorded_at'] del data['recorded_at']
self.assertEqual(data, { self.assertEqual({
u'id': sample_id, u'id': sample_id,
u'metadata': {u'display_name': u'test-server', u'metadata': {u'display_name': u'test-server',
u'is_public': u'False', u'is_public': u'False',
@ -239,14 +239,14 @@ class TestListMeters(FunctionalTest,
u'unit': u'', u'unit': u'',
u'source': 'test_source', u'source': 'test_source',
u'user_id': u'user-id', u'user_id': u'user-id',
u'volume': 3.0}) u'volume': 3.0}, data)
def test_get_not_existing_sample(self): def test_get_not_existing_sample(self):
resp = self.get_json('/samples/not_exists', expect_errors=True, resp = self.get_json('/samples/not_exists', expect_errors=True,
status=404) status=404)
self.assertEqual(jsonutils.loads(resp.body)['error_message'] self.assertEqual("Sample not_exists Not Found",
['faultstring'], jsonutils.loads(resp.body)['error_message']
"Sample not_exists Not Found") ['faultstring'])
def test_list_samples_with_dict_metadata(self): def test_list_samples_with_dict_metadata(self):
data = self.get_json('/samples', data = self.get_json('/samples',
@ -258,7 +258,7 @@ class TestListMeters(FunctionalTest,
self.assertIn('id', data[0]) self.assertIn('id', data[0])
del data[0]['id'] # Randomly generated del data[0]['id'] # Randomly generated
del data[0]['recorded_at'] del data[0]['recorded_at']
self.assertEqual(data, [{ self.assertEqual([{
u'user_id': u'user-id4', u'user_id': u'user-id4',
u'resource_id': u'resource-id4', u'resource_id': u'resource-id4',
u'timestamp': u'2012-07-02T10:43:00', u'timestamp': u'2012-07-02T10:43:00',
@ -275,7 +275,7 @@ class TestListMeters(FunctionalTest,
u'properties.prop_1': u'prop_value', u'properties.prop_1': u'prop_value',
u'is_public': u'True', u'is_public': u'True',
u'size': u'0'} u'size': u'0'}
}]) }], data)
def test_list_meters_metadata_query(self): def test_list_meters_metadata_query(self):
data = self.get_json('/meters/meter.test', data = self.get_json('/meters/meter.test',
@ -284,10 +284,10 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample1', 'value': 'self.sample1',
}],) }],)
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id']),
set(['resource-id'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
def test_list_meters_resource_metadata_query(self): def test_list_meters_resource_metadata_query(self):
# NOTE(jd) Same test as above, but with the alias resource_metadata # NOTE(jd) Same test as above, but with the alias resource_metadata
@ -298,10 +298,10 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample1', 'value': 'self.sample1',
}],) }],)
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id']),
set(['resource-id'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
def test_list_meters_multi_metadata_query(self): def test_list_meters_multi_metadata_query(self):
data = self.get_json('/meters/meter.test', data = self.get_json('/meters/meter.test',
@ -314,10 +314,10 @@ class TestListMeters(FunctionalTest,
'value': 'test-server', 'value': 'test-server',
}],) }],)
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id']),
set(['resource-id'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
def test_list_meters_query_integer_metadata(self): def test_list_meters_query_integer_metadata(self):
data = self.get_json('/meters/meter.test', data = self.get_json('/meters/meter.test',
@ -327,13 +327,13 @@ class TestListMeters(FunctionalTest,
'type': 'integer'}] 'type': 'integer'}]
) )
self.assertEqual(2, len(data)) self.assertEqual(2, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id',
set(['resource-id', 'resource-id3']),
'resource-id3'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
self.assertEqual(set(r['resource_metadata']['size'] for r in data), self.assertEqual(set(['0']),
set(['0'])) set(r['resource_metadata']['size'] for r in data))
def test_list_meters_query_float_metadata(self): def test_list_meters_query_float_metadata(self):
data = self.get_json('/meters/meter.test', data = self.get_json('/meters/meter.test',
@ -343,13 +343,13 @@ class TestListMeters(FunctionalTest,
'type': 'float'}] 'type': 'float'}]
) )
self.assertEqual(2, len(data)) self.assertEqual(2, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id',
set(['resource-id', 'resource-id3']),
'resource-id3'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
self.assertEqual(set(r['resource_metadata']['util'] for r in data), self.assertEqual(set(['0.75']),
set(['0.75'])) set(r['resource_metadata']['util'] for r in data))
def test_list_meters_query_boolean_metadata(self): def test_list_meters_query_boolean_metadata(self):
data = self.get_json('/meters/meter.mine', data = self.get_json('/meters/meter.mine',
@ -359,12 +359,13 @@ class TestListMeters(FunctionalTest,
'type': 'boolean'}] 'type': 'boolean'}]
) )
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id2']),
set(['resource-id2'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.mine']),
set(['meter.mine'])) set(r['counter_name'] for r in data))
self.assertEqual(set(r['resource_metadata']['is_public'] for r self.assertEqual(set(['False']),
in data), set(['False'])) set(r['resource_metadata']['is_public']
for r in data))
def test_list_meters_query_string_metadata(self): def test_list_meters_query_string_metadata(self):
data = self.get_json('/meters/meter.test', data = self.get_json('/meters/meter.test',
@ -373,12 +374,12 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample'}] 'value': 'self.sample'}]
) )
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id']),
set(['resource-id'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
self.assertEqual(set(r['resource_metadata']['tag'] for r in data), self.assertEqual(set(['self.sample']),
set(['self.sample'])) set(r['resource_metadata']['tag'] for r in data))
def test_list_meters_query_integer_float_metadata_without_type(self): def test_list_meters_query_integer_float_metadata_without_type(self):
data = self.get_json('/meters/meter.test', data = self.get_json('/meters/meter.test',
@ -390,14 +391,14 @@ class TestListMeters(FunctionalTest,
'value': '0.75'}] 'value': '0.75'}]
) )
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id3']),
set(['resource-id3'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
self.assertEqual(set(r['resource_metadata']['size'] for r in data), self.assertEqual(set(['0']),
set(['0'])) set(r['resource_metadata']['size'] for r in data))
self.assertEqual(set(r['resource_metadata']['util'] for r in data), self.assertEqual(set(['0.75']),
set(['0.75'])) set(r['resource_metadata']['util'] for r in data))
def test_with_resource(self): def test_with_resource(self):
data = self.get_json('/meters', q=[{'field': 'resource_id', data = self.get_json('/meters', q=[{'field': 'resource_id',
@ -433,10 +434,10 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample2', 'value': 'self.sample2',
}]) }])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data), self.assertEqual(set(['resource-id2']),
set(['resource-id2'])) set(r['resource_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.mine']),
set(['meter.mine'])) set(r['counter_name'] for r in data))
def test_with_source(self): def test_with_source(self):
data = self.get_json('/meters', q=[{'field': 'source', data = self.get_json('/meters', q=[{'field': 'source',
@ -461,9 +462,10 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample2', 'value': 'self.sample2',
}]) }])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['source'] for r in data), set(['test_source'])) self.assertEqual(set(['test_source']),
self.assertEqual(set(r['counter_name'] for r in data), set(r['source'] for r in data))
set(['meter.mine'])) self.assertEqual(set(['meter.mine']),
set(r['counter_name'] for r in data))
def test_with_source_non_existent(self): def test_with_source_non_existent(self):
data = self.get_json('/meters', data = self.get_json('/meters',
@ -519,9 +521,9 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample1', 'value': 'self.sample1',
}]) }])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['user_id'] for r in data), set(['user-id'])) self.assertEqual(set(['user-id']), set(r['user_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
def test_with_user_non_existent(self): def test_with_user_non_existent(self):
data = self.get_json('/meters', data = self.get_json('/meters',
@ -529,7 +531,7 @@ class TestListMeters(FunctionalTest,
'value': 'user-id-foobar123', 'value': 'user-id-foobar123',
}], }],
) )
self.assertEqual(data, []) self.assertEqual([], data)
def test_with_project(self): def test_with_project(self):
data = self.get_json('/meters', data = self.get_json('/meters',
@ -569,10 +571,10 @@ class TestListMeters(FunctionalTest,
'value': 'self.sample1', 'value': 'self.sample1',
}]) }])
self.assertEqual(1, len(data)) self.assertEqual(1, len(data))
self.assertEqual(set(r['project_id'] for r in data), self.assertEqual(set(['project-id']),
set(['project-id'])) set(r['project_id'] for r in data))
self.assertEqual(set(r['counter_name'] for r in data), self.assertEqual(set(['meter.test']),
set(['meter.test'])) set(r['counter_name'] for r in data))
def test_with_project_non_existent(self): def test_with_project_non_existent(self):
data = self.get_json('/meters', data = self.get_json('/meters',
@ -580,7 +582,7 @@ class TestListMeters(FunctionalTest,
'value': 'jd-was-here', 'value': 'jd-was-here',
}], }],
) )
self.assertEqual(data, []) self.assertEqual([], data)
def test_list_meters_meter_id(self): def test_list_meters_meter_id(self):
data = self.get_json('/meters') data = self.get_json('/meters')

34
ceilometer/tests/api/v2/test_list_resources_scenarios.py
View File

@ -49,9 +49,9 @@ class TestListResources(FunctionalTest,
def _verify_sample_timestamps(self, res, first, last): def _verify_sample_timestamps(self, res, first, last):
self.assertTrue('first_sample_timestamp' in res) self.assertTrue('first_sample_timestamp' in res)
self.assertEqual(res['first_sample_timestamp'], self._isotime(first)) self.assertEqual(self._isotime(first), res['first_sample_timestamp'])
self.assertTrue('last_sample_timestamp' in res) self.assertTrue('last_sample_timestamp' in res)
self.assertEqual(res['last_sample_timestamp'], self._isotime(last)) self.assertEqual(self._isotime(last), res['last_sample_timestamp'])
def test_instance_no_metadata(self): def test_instance_no_metadata(self):
timestamp = datetime.datetime(2012, 7, 2, 10, 40) timestamp = datetime.datetime(2012, 7, 2, 10, 40)
@ -294,16 +294,16 @@ class TestListResources(FunctionalTest,
self.conn.record_metering_data(msg2) self.conn.record_metering_data(msg2)
resp1 = self.get_json('/resources/resource-id-1') resp1 = self.get_json('/resources/resource-id-1')
self.assertEqual(resp1["resource_id"], "resource-id-1") self.assertEqual("resource-id-1", resp1["resource_id"])
resp2 = self.get_json('/resources/resource-id-2') resp2 = self.get_json('/resources/resource-id-2')
self.assertEqual(resp2["resource_id"], "resource-id-2") self.assertEqual("resource-id-2", resp2["resource_id"])
resp3 = self.get_json('/resources/resource-id-3', expect_errors=True) resp3 = self.get_json('/resources/resource-id-3', expect_errors=True)
self.assertEqual(resp3.status_code, 404) self.assertEqual(404, resp3.status_code)
self.assertEqual(json.loads(resp3.body)['error_message'] self.assertEqual("Resource resource-id-3 Not Found",
['faultstring'], json.loads(resp3.body)['error_message']
"Resource resource-id-3 Not Found") ['faultstring'])
def test_with_user(self): def test_with_user(self):
sample1 = sample.Sample( sample1 = sample.Sample(
@ -478,13 +478,11 @@ class TestListResources(FunctionalTest,
data = self.get_json('/resources') data = self.get_json('/resources')
metadata = data[0]['metadata'] metadata = data[0]['metadata']
self.assertEqual( self.assertEqual([(u'dict_properties.key', u'value'),
[(u'dict_properties.key', u'value'), (u'display_name', u'test-server'),
(u'display_name', u'test-server'), (u'not_ignored_list', u"['returned']"),
(u'not_ignored_list', u"['returned']"), (u'tag', u'self.sample')],
(u'tag', u'self.sample') list(sorted(metadata.iteritems())))
],
list(sorted(metadata.iteritems())))
def test_resource_meter_links(self): def test_resource_meter_links(self):
sample1 = sample.Sample( sample1 = sample.Sample(
@ -509,11 +507,11 @@ class TestListResources(FunctionalTest,
data = self.get_json('/resources') data = self.get_json('/resources')
links = data[0]['links'] links = data[0]['links']
self.assertEqual(len(links), 2) self.assertEqual(2, len(links))
self.assertEqual(links[0]['rel'], 'self') self.assertEqual('self', links[0]['rel'])
self.assertTrue((self.PATH_PREFIX + '/resources/resource-id') self.assertTrue((self.PATH_PREFIX + '/resources/resource-id')
in links[0]['href']) in links[0]['href'])
self.assertEqual(links[1]['rel'], 'instance') self.assertEqual('instance', links[1]['rel'])
self.assertTrue((self.PATH_PREFIX + '/meters/instance?' self.assertTrue((self.PATH_PREFIX + '/meters/instance?'
'q.field=resource_id&q.value=resource-id') 'q.field=resource_id&q.value=resource-id')
in links[1]['href']) in links[1]['href'])

20
ceilometer/tests/api/v2/test_post_samples_scenarios.py
View File

@ -84,8 +84,8 @@ class TestPostSamples(FunctionalTest,
data = self.post_json('/meters/my_counter_name/', s1, data = self.post_json('/meters/my_counter_name/', s1,
expect_errors=True) expect_errors=True)
self.assertEqual(data.status_int, 400) self.assertEqual(400, data.status_int)
self.assertEqual(len(self.published), 0) self.assertEqual(0, len(self.published))
def test_messsage_id_provided(self): def test_messsage_id_provided(self):
"""Do not accept sample with message_id.""" """Do not accept sample with message_id."""
@ -104,8 +104,8 @@ class TestPostSamples(FunctionalTest,
data = self.post_json('/meters/my_counter_name/', s1, data = self.post_json('/meters/my_counter_name/', s1,
expect_errors=True) expect_errors=True)
self.assertEqual(data.status_int, 400) self.assertEqual(400, data.status_int)
self.assertEqual(len(self.published), 0) self.assertEqual(0, len(self.published))
def test_wrong_project_id(self): def test_wrong_project_id(self):
"""Do not accept cross posting samples to different projects.""" """Do not accept cross posting samples to different projects."""
@ -129,8 +129,8 @@ class TestPostSamples(FunctionalTest,
"bc23a9d531064583ace8f67dad60f6bb", "bc23a9d531064583ace8f67dad60f6bb",
}) })
self.assertEqual(data.status_int, 400) self.assertEqual(400, data.status_int)
self.assertEqual(len(self.published), 0) self.assertEqual(0, len(self.published))
def test_multiple_samples(self): def test_multiple_samples(self):
"""Send multiple samples. """Send multiple samples.
@ -197,7 +197,7 @@ class TestPostSamples(FunctionalTest,
print('posting without %s' % m) print('posting without %s' % m)
data = self.post_json('/meters/my_counter_name', s_broke, data = self.post_json('/meters/my_counter_name', s_broke,
expect_errors=True) expect_errors=True)
self.assertEqual(data.status_int, 400) self.assertEqual(400, data.status_int)
def test_multiple_project_id_and_admin(self): def test_multiple_project_id_and_admin(self):
"""Allow admin is allowed to set multiple project_id.""" """Allow admin is allowed to set multiple project_id."""
@ -223,7 +223,7 @@ class TestPostSamples(FunctionalTest,
data = self.post_json('/meters/my_counter_name/', s1, data = self.post_json('/meters/my_counter_name/', s1,
headers={"X-Roles": "admin"}) headers={"X-Roles": "admin"})
self.assertEqual(data.status_int, 200) self.assertEqual(200, data.status_int)
for x, s in enumerate(s1): for x, s in enumerate(s1):
# source is modified to include the project_id. # source is modified to include the project_id.
s['source'] = '%s:%s' % (s['project_id'], s['source'] = '%s:%s' % (s['project_id'],
@ -269,7 +269,7 @@ class TestPostSamples(FunctionalTest,
'name2': 'value2'}}] 'name2': 'value2'}}]
data = self.post_json('/meters/my_counter_name/', s1, data = self.post_json('/meters/my_counter_name/', s1,
expect_errors=True) expect_errors=True)
self.assertEqual(data.status_int, 200) self.assertEqual(200, data.status_int)
for x, s in enumerate(s1): for x, s in enumerate(s1):
# source is modified to include the project_id. # source is modified to include the project_id.
s['source'] = '%s:%s' % ( s['source'] = '%s:%s' % (
@ -306,7 +306,7 @@ class TestPostSamples(FunctionalTest,
'X-User-Id': user_id, 'X-User-Id': user_id,
}) })
self.assertEqual(data.status_int, 200) self.assertEqual(200, data.status_int)
for x, s in enumerate(s1): for x, s in enumerate(s1):
# source is modified to include the project_id. # source is modified to include the project_id.
s['source'] = '%s:%s' % (project_id, s['source'] = '%s:%s' % (project_id,

62
ceilometer/tests/api/v2/test_query.py
View File

@ -40,7 +40,7 @@ class TestQuery(test.BaseTestCase):
value='123', value='123',
type='integer') type='integer')
expected = 123 expected = 123
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_float(self): def test_get_value_as_type_with_float(self):
query = api.Query(field='metadata.size', query = api.Query(field='metadata.size',
@ -48,7 +48,7 @@ class TestQuery(test.BaseTestCase):
value='123.456', value='123.456',
type='float') type='float')
expected = 123.456 expected = 123.456
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_boolean(self): def test_get_value_as_type_with_boolean(self):
query = api.Query(field='metadata.is_public', query = api.Query(field='metadata.is_public',
@ -56,7 +56,7 @@ class TestQuery(test.BaseTestCase):
value='True', value='True',
type='boolean') type='boolean')
expected = True expected = True
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_string(self): def test_get_value_as_type_with_string(self):
query = api.Query(field='metadata.name', query = api.Query(field='metadata.name',
@ -64,35 +64,35 @@ class TestQuery(test.BaseTestCase):
value='linux', value='linux',
type='string') type='string')
expected = 'linux' expected = 'linux'
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_integer_without_type(self): def test_get_value_as_type_with_integer_without_type(self):
query = api.Query(field='metadata.size', query = api.Query(field='metadata.size',
op='eq', op='eq',
value='123') value='123')
expected = 123 expected = 123
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_float_without_type(self): def test_get_value_as_type_with_float_without_type(self):
query = api.Query(field='metadata.size', query = api.Query(field='metadata.size',
op='eq', op='eq',
value='123.456') value='123.456')
expected = 123.456 expected = 123.456
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_boolean_without_type(self): def test_get_value_as_type_with_boolean_without_type(self):
query = api.Query(field='metadata.is_public', query = api.Query(field='metadata.is_public',
op='eq', op='eq',
value='True') value='True')
expected = True expected = True
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_string_without_type(self): def test_get_value_as_type_with_string_without_type(self):
query = api.Query(field='metadata.name', query = api.Query(field='metadata.name',
op='eq', op='eq',
value='linux') value='linux')
expected = 'linux' expected = 'linux'
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_bad_type(self): def test_get_value_as_type_with_bad_type(self):
query = api.Query(field='metadata.size', query = api.Query(field='metadata.size',
@ -114,7 +114,7 @@ class TestQuery(test.BaseTestCase):
op='eq', op='eq',
value='123-1') value='123-1')
expected = '123-1' expected = '123-1'
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_boolean_expression_without_type(self): def test_get_value_as_type_boolean_expression_without_type(self):
# bug 1221736 # bug 1221736
@ -122,7 +122,7 @@ class TestQuery(test.BaseTestCase):
op='eq', op='eq',
value='True or False') value='True or False')
expected = 'True or False' expected = 'True or False'
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_syntax_error(self): def test_get_value_as_type_with_syntax_error(self):
# bug 1221736 # bug 1221736
@ -131,7 +131,7 @@ class TestQuery(test.BaseTestCase):
op='eq', op='eq',
value=value) value=value)
expected = value expected = value
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
def test_get_value_as_type_with_syntax_error_colons(self): def test_get_value_as_type_with_syntax_error_colons(self):
# bug 1221736 # bug 1221736
@ -140,20 +140,20 @@ class TestQuery(test.BaseTestCase):
op='eq', op='eq',
value=value) value=value)
expected = value expected = value
self.assertEqual(query._get_value_as_type(), expected) self.assertEqual(expected, query._get_value_as_type())
class TestValidateGroupByFields(test.BaseTestCase): class TestValidateGroupByFields(test.BaseTestCase):
def test_valid_field(self): def test_valid_field(self):
result = api._validate_groupby_fields(['user_id']) result = api._validate_groupby_fields(['user_id'])
self.assertEqual(result, ['user_id']) self.assertEqual(['user_id'], result)
def test_valid_fields_multiple(self): def test_valid_fields_multiple(self):
result = set( result = set(
api._validate_groupby_fields(['user_id', 'project_id', 'source']) api._validate_groupby_fields(['user_id', 'project_id', 'source'])
) )
self.assertEqual(result, set(['user_id', 'project_id', 'source'])) self.assertEqual(set(['user_id', 'project_id', 'source']), result)
def test_invalid_field(self): def test_invalid_field(self):
self.assertRaises(wsme.exc.UnknownArgument, self.assertRaises(wsme.exc.UnknownArgument,
@ -169,7 +169,7 @@ class TestValidateGroupByFields(test.BaseTestCase):
result = set( result = set(
api._validate_groupby_fields(['user_id', 'source', 'user_id']) api._validate_groupby_fields(['user_id', 'source', 'user_id'])
) )
self.assertEqual(result, set(['user_id', 'source'])) self.assertEqual(set(['user_id', 'source']), result)
class TestQueryToKwArgs(tests_base.BaseTestCase): class TestQueryToKwArgs(tests_base.BaseTestCase):
@ -186,8 +186,8 @@ class TestQueryToKwArgs(tests_base.BaseTestCase):
value='uid')] value='uid')]
kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__) kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__)
self.assertIn('user', kwargs) self.assertIn('user', kwargs)
self.assertEqual(len(kwargs), 1) self.assertEqual(1, len(kwargs))
self.assertEqual(kwargs['user'], 'uid') self.assertEqual('uid', kwargs['user'])
def test_sample_filter_multi(self): def test_sample_filter_multi(self):
q = [api.Query(field='user_id', q = [api.Query(field='user_id',
@ -206,12 +206,12 @@ class TestQueryToKwArgs(tests_base.BaseTestCase):
op='eq', op='eq',
value='meter_name')] value='meter_name')]
kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__) kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__)
self.assertEqual(len(kwargs), 5) self.assertEqual(5, len(kwargs))
self.assertEqual(kwargs['user'], 'uid') self.assertEqual('uid', kwargs['user'])
self.assertEqual(kwargs['project'], 'pid') self.assertEqual('pid', kwargs['project'])
self.assertEqual(kwargs['resource'], 'rid') self.assertEqual('rid', kwargs['resource'])
self.assertEqual(kwargs['source'], 'source_name') self.assertEqual('source_name', kwargs['source'])
self.assertEqual(kwargs['meter'], 'meter_name') self.assertEqual('meter_name', kwargs['meter'])
def test_sample_filter_timestamp(self): def test_sample_filter_timestamp(self):
ts_start = timeutils.utcnow() ts_start = timeutils.utcnow()
@ -223,11 +223,11 @@ class TestQueryToKwArgs(tests_base.BaseTestCase):
op='gt', op='gt',
value=str(ts_start))] value=str(ts_start))]
kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__) kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__)
self.assertEqual(len(kwargs), 4) self.assertEqual(4, len(kwargs))
self.assertTimestampEqual(kwargs['start'], ts_start) self.assertTimestampEqual(kwargs['start'], ts_start)
self.assertTimestampEqual(kwargs['end'], ts_end) self.assertTimestampEqual(kwargs['end'], ts_end)
self.assertEqual(kwargs['start_timestamp_op'], 'gt') self.assertEqual('gt', kwargs['start_timestamp_op'])
self.assertEqual(kwargs['end_timestamp_op'], 'lt') self.assertEqual('lt', kwargs['end_timestamp_op'])
def test_sample_filter_meta(self): def test_sample_filter_meta(self):
q = [api.Query(field='metadata.size', q = [api.Query(field='metadata.size',
@ -237,10 +237,10 @@ class TestQueryToKwArgs(tests_base.BaseTestCase):
op='eq', op='eq',
value='meta_id')] value='meta_id')]
kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__) kwargs = api._query_to_kwargs(q, storage.SampleFilter.__init__)
self.assertEqual(len(kwargs), 1) self.assertEqual(1, len(kwargs))
self.assertEqual(len(kwargs['metaquery']), 2) self.assertEqual(2, len(kwargs['metaquery']))
self.assertEqual(kwargs['metaquery']['metadata.size'], 20) self.assertEqual(20, kwargs['metaquery']['metadata.size'])
self.assertEqual(kwargs['metaquery']['metadata.id'], 'meta_id') self.assertEqual('meta_id', kwargs['metaquery']['metadata.id'])
def test_sample_filter_non_equality_on_metadata(self): def test_sample_filter_non_equality_on_metadata(self):
queries = [api.Query(field='resource_metadata.image_id', queries = [api.Query(field='resource_metadata.image_id',
@ -320,4 +320,4 @@ class TestQueryToKwArgs(tests_base.BaseTestCase):
kwargs = api._query_to_kwargs(queries, kwargs = api._query_to_kwargs(queries,
storage.SampleFilter.__init__) storage.SampleFilter.__init__)
for o in ['user', 'project', 'resource']: for o in ['user', 'project', 'resource']:
self.assertEqual(kwargs.get(o), 'fake_%s_id' % o) self.assertEqual('fake_%s_id' % o, kwargs.get(o))

28
ceilometer/tests/api/v2/test_statistics.py
View File

@ -60,9 +60,9 @@ class TestStatisticsDuration(test.BaseTestCase):
start_timestamp=self.start, start_timestamp=self.start,
end_timestamp=self.end, end_timestamp=self.end,
) )
self.assertEqual(s.duration_start, self.start) self.assertEqual(self.start, s.duration_start)
self.assertEqual(s.duration_end, self.middle1) self.assertEqual(self.middle1, s.duration_end)
self.assertEqual(s.duration, 8 * 60 * 60) self.assertEqual(8 * 60 * 60, s.duration)
def test_within_range(self): def test_within_range(self):
s = v2.Statistics(duration_start=self.middle1, s = v2.Statistics(duration_start=self.middle1,
@ -70,9 +70,9 @@ class TestStatisticsDuration(test.BaseTestCase):
start_timestamp=self.start, start_timestamp=self.start,
end_timestamp=self.end, end_timestamp=self.end,
) )
self.assertEqual(s.duration_start, self.middle1) self.assertEqual(self.middle1, s.duration_start)
self.assertEqual(s.duration_end, self.middle2) self.assertEqual(self.middle2, s.duration_end)
self.assertEqual(s.duration, 10 * 60 * 60) self.assertEqual(10 * 60 * 60, s.duration)
def test_within_range_zero_duration(self): def test_within_range_zero_duration(self):
s = v2.Statistics(duration_start=self.middle1, s = v2.Statistics(duration_start=self.middle1,
@ -80,9 +80,9 @@ class TestStatisticsDuration(test.BaseTestCase):
start_timestamp=self.start, start_timestamp=self.start,
end_timestamp=self.end, end_timestamp=self.end,
) )
self.assertEqual(s.duration_start, self.middle1) self.assertEqual(self.middle1, s.duration_start)
self.assertEqual(s.duration_end, self.middle1) self.assertEqual(self.middle1, s.duration_end)
self.assertEqual(s.duration, 0) self.assertEqual(0, s.duration)
def test_overlap_range_end(self): def test_overlap_range_end(self):
s = v2.Statistics(duration_start=self.middle2, s = v2.Statistics(duration_start=self.middle2,
@ -90,9 +90,9 @@ class TestStatisticsDuration(test.BaseTestCase):
start_timestamp=self.start, start_timestamp=self.start,
end_timestamp=self.end, end_timestamp=self.end,
) )
self.assertEqual(s.duration_start, self.middle2) self.assertEqual(self.middle2, s.duration_start)
self.assertEqual(s.duration_end, self.end) self.assertEqual(self.end, s.duration_end)
self.assertEqual(s.duration, ((6 * 60) - 1) * 60) self.assertEqual(((6 * 60) - 1) * 60, s.duration)
def test_after_range(self): def test_after_range(self):
s = v2.Statistics(duration_start=self.late1, s = v2.Statistics(duration_start=self.late1,
@ -110,5 +110,5 @@ class TestStatisticsDuration(test.BaseTestCase):
start_timestamp=None, start_timestamp=None,
end_timestamp=None, end_timestamp=None,
) )
self.assertEqual(s.duration_start, self.late1) self.assertEqual(self.late1, s.duration_start)
self.assertEqual(s.duration_end, self.late2) self.assertEqual(self.late2, s.duration_end)

4
ceilometer/tests/api/v2/test_wsme_custom_type.py
View File

@ -29,9 +29,9 @@ class TestWsmeCustomType(test.BaseTestCase):
ae = v2.AdvEnum("name", str, "one", "other", default="other") ae = v2.AdvEnum("name", str, "one", "other", default="other")
obj = dummybase() obj = dummybase()
self.assertEqual(obj.ae, "other") self.assertEqual("other", obj.ae)
obj = dummybase(ae="one") obj = dummybase(ae="one")
self.assertEqual(obj.ae, "one") self.assertEqual("one", obj.ae)
self.assertRaises(ValueError, dummybase, ae="not exists") self.assertRaises(ValueError, dummybase, ae="not exists")

44
ceilometer/tests/compute/notifications/test_cpu.py
View File

@ -94,7 +94,7 @@ class TestMetricsNotifications(test.BaseTestCase):
self.assertIn(METRICS_UPDATE['event_type'], self.assertIn(METRICS_UPDATE['event_type'],
ic.event_types) ic.event_types)
samples = list(ic.process_notification(METRICS_UPDATE)) samples = list(ic.process_notification(METRICS_UPDATE))
self.assertEqual(samples[0].resource_id, RES_ID) self.assertEqual(RES_ID, samples[0].resource_id)
return samples[0] return samples[0]
def test_compute_metrics(self): def test_compute_metrics(self):
@ -104,56 +104,56 @@ class TestMetricsNotifications(test.BaseTestCase):
info = ic._get_sample(METRICS_UPDATE, 'cpu.frequency') info = ic._get_sample(METRICS_UPDATE, 'cpu.frequency')
info_none = ic._get_sample(METRICS_UPDATE, 'abc.efg') info_none = ic._get_sample(METRICS_UPDATE, 'abc.efg')
info_error = ic._get_sample(ERROR_METRICS, 'cpu.frequency') info_error = ic._get_sample(ERROR_METRICS, 'cpu.frequency')
self.assertEqual(info['payload']['name'], 'cpu.frequency') self.assertEqual('cpu.frequency', info['payload']['name'])
self.assertIsNone(info_none) self.assertIsNone(info_none)
self.assertIsNone(info_error) self.assertIsNone(info_error)
def test_compute_cpu_frequency(self): def test_compute_cpu_frequency(self):
c = self._process_notification(cpu.CpuFrequency()) c = self._process_notification(cpu.CpuFrequency())
self.assertEqual(c.name, 'compute.node.cpu.frequency') self.assertEqual('compute.node.cpu.frequency', c.name)
self.assertEqual(c.volume, 1600) self.assertEqual(1600, c.volume)
def test_compute_cpu_user_time(self): def test_compute_cpu_user_time(self):
c = self._process_notification(cpu.CpuUserTime()) c = self._process_notification(cpu.CpuUserTime())
self.assertEqual(c.name, 'compute.node.cpu.user.time') self.assertEqual('compute.node.cpu.user.time', c.name)
self.assertEqual(c.volume, 17421440000000L) self.assertEqual(17421440000000L, c.volume)
def test_compute_cpu_kernel_time(self): def test_compute_cpu_kernel_time(self):
c = self._process_notification(cpu.CpuKernelTime()) c = self._process_notification(cpu.CpuKernelTime())
self.assertEqual(c.name, 'compute.node.cpu.kernel.time') self.assertEqual('compute.node.cpu.kernel.time', c.name)
self.assertEqual(c.volume, 7852600000000L) self.assertEqual(7852600000000L, c.volume)
def test_compute_cpu_idle_time(self): def test_compute_cpu_idle_time(self):
c = self._process_notification(cpu.CpuIdleTime()) c = self._process_notification(cpu.CpuIdleTime())
self.assertEqual(c.name, 'compute.node.cpu.idle.time') self.assertEqual('compute.node.cpu.idle.time', c.name)
self.assertEqual(c.volume, 1307374400000000L) self.assertEqual(1307374400000000L, c.volume)
def test_compute_cpu_iowait_time(self): def test_compute_cpu_iowait_time(self):
c = self._process_notification(cpu.CpuIowaitTime()) c = self._process_notification(cpu.CpuIowaitTime())
self.assertEqual(c.name, 'compute.node.cpu.iowait.time') self.assertEqual('compute.node.cpu.iowait.time', c.name)
self.assertEqual(c.volume, 11697470000000L) self.assertEqual(11697470000000L, c.volume)
def test_compute_cpu_kernel_percent(self): def test_compute_cpu_kernel_percent(self):
c = self._process_notification(cpu.CpuKernelPercent()) c = self._process_notification(cpu.CpuKernelPercent())
self.assertEqual(c.name, 'compute.node.cpu.kernel.percent') self.assertEqual('compute.node.cpu.kernel.percent', c.name)
self.assertEqual(c.volume, 0.5841204961898534) self.assertEqual(0.5841204961898534, c.volume)
def test_compute_cpu_idle_percent(self): def test_compute_cpu_idle_percent(self):
c = self._process_notification(cpu.CpuIdlePercent()) c = self._process_notification(cpu.CpuIdlePercent())
self.assertEqual(c.name, 'compute.node.cpu.idle.percent') self.assertEqual('compute.node.cpu.idle.percent', c.name)
self.assertEqual(c.volume, 97.24985141658965) self.assertEqual(97.24985141658965, c.volume)
def test_compute_cpu_user_percent(self): def test_compute_cpu_user_percent(self):
c = self._process_notification(cpu.CpuUserPercent()) c = self._process_notification(cpu.CpuUserPercent())
self.assertEqual(c.name, 'compute.node.cpu.user.percent') self.assertEqual('compute.node.cpu.user.percent', c.name)
self.assertEqual(c.volume, 1.2959045637294348) self.assertEqual(1.2959045637294348, c.volume)
def test_compute_cpu_iowait_percent(self): def test_compute_cpu_iowait_percent(self):
c = self._process_notification(cpu.CpuIowaitPercent()) c = self._process_notification(cpu.CpuIowaitPercent())
self.assertEqual(c.name, 'compute.node.cpu.iowait.percent') self.assertEqual('compute.node.cpu.iowait.percent', c.name)
self.assertEqual(c.volume, 0.8701235234910634) self.assertEqual(0.8701235234910634, c.volume)
def test_compute_cpu_percent(self): def test_compute_cpu_percent(self):
c = self._process_notification(cpu.CpuPercent()) c = self._process_notification(cpu.CpuPercent())
self.assertEqual(c.name, 'compute.node.cpu.percent') self.assertEqual('compute.node.cpu.percent', c.name)
self.assertEqual(c.volume, 2.7501485834103515) self.assertEqual(2.7501485834103515, c.volume)

93
ceilometer/tests/compute/notifications/test_instance.py
View File

@ -565,7 +565,7 @@ class TestNotifications(test.BaseTestCase):
('host', info.resource_metadata['host'], ('host', info.resource_metadata['host'],
INSTANCE_CREATE_END['publisher_id']), INSTANCE_CREATE_END['publisher_id']),
]: ]:
self.assertEqual(actual, expected, name) self.assertEqual(expected, actual, name)
@staticmethod @staticmethod
def _find_counter(counters, name): def _find_counter(counters, name):
@ -582,142 +582,142 @@ class TestNotifications(test.BaseTestCase):
def test_instance_create_instance(self): def test_instance_create_instance(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_CREATE_END)) counters = list(ic.process_notification(INSTANCE_CREATE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, 1) self.assertEqual(1, c.volume)
def test_instance_create_flavor(self): def test_instance_create_flavor(self):
ic = instance.InstanceFlavor() ic = instance.InstanceFlavor()
counters = list(ic.process_notification(INSTANCE_CREATE_END)) counters = list(ic.process_notification(INSTANCE_CREATE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, 1) self.assertEqual(1, c.volume)
def test_instance_create_memory(self): def test_instance_create_memory(self):
ic = instance.Memory() ic = instance.Memory()
counters = list(ic.process_notification(INSTANCE_CREATE_END)) counters = list(ic.process_notification(INSTANCE_CREATE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, INSTANCE_CREATE_END['payload']['memory_mb']) self.assertEqual(INSTANCE_CREATE_END['payload']['memory_mb'], c.volume)
def test_instance_create_vcpus(self): def test_instance_create_vcpus(self):
ic = instance.VCpus() ic = instance.VCpus()
counters = list(ic.process_notification(INSTANCE_CREATE_END)) counters = list(ic.process_notification(INSTANCE_CREATE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, INSTANCE_CREATE_END['payload']['vcpus']) self.assertEqual(INSTANCE_CREATE_END['payload']['vcpus'], c.volume)
def test_instance_create_root_disk_size(self): def test_instance_create_root_disk_size(self):
ic = instance.RootDiskSize() ic = instance.RootDiskSize()
counters = list(ic.process_notification(INSTANCE_CREATE_END)) counters = list(ic.process_notification(INSTANCE_CREATE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, INSTANCE_CREATE_END['payload']['root_gb']) self.assertEqual(INSTANCE_CREATE_END['payload']['root_gb'], c.volume)
def test_instance_create_ephemeral_disk_size(self): def test_instance_create_ephemeral_disk_size(self):
ic = instance.EphemeralDiskSize() ic = instance.EphemeralDiskSize()
counters = list(ic.process_notification(INSTANCE_CREATE_END)) counters = list(ic.process_notification(INSTANCE_CREATE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, self.assertEqual(INSTANCE_CREATE_END['payload']['ephemeral_gb'],
INSTANCE_CREATE_END['payload']['ephemeral_gb']) c.volume)
def test_instance_exists_instance(self): def test_instance_exists_instance(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_EXISTS)) counters = list(ic.process_notification(INSTANCE_EXISTS))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
def test_instance_exists_flavor(self): def test_instance_exists_flavor(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_EXISTS)) counters = list(ic.process_notification(INSTANCE_EXISTS))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
def test_instance_delete_instance(self): def test_instance_delete_instance(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_DELETE_START)) counters = list(ic.process_notification(INSTANCE_DELETE_START))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
def test_instance_delete_flavor(self): def test_instance_delete_flavor(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_DELETE_START)) counters = list(ic.process_notification(INSTANCE_DELETE_START))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
def test_instance_finish_resize_instance(self): def test_instance_finish_resize_instance(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END)) counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, 1) self.assertEqual(1, c.volume)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_finish_resize_flavor(self): def test_instance_finish_resize_flavor(self):
ic = instance.InstanceFlavor() ic = instance.InstanceFlavor()
counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END)) counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, 1) self.assertEqual(1, c.volume)
self.assertEqual(c.name, 'instance:m1.small') self.assertEqual('instance:m1.small', c.name)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_finish_resize_memory(self): def test_instance_finish_resize_memory(self):
ic = instance.Memory() ic = instance.Memory()
counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END)) counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, self.assertEqual(INSTANCE_FINISH_RESIZE_END['payload']['memory_mb'],
INSTANCE_FINISH_RESIZE_END['payload']['memory_mb']) c.volume)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_finish_resize_vcpus(self): def test_instance_finish_resize_vcpus(self):
ic = instance.VCpus() ic = instance.VCpus()
counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END)) counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, self.assertEqual(INSTANCE_FINISH_RESIZE_END['payload']['vcpus'],
INSTANCE_FINISH_RESIZE_END['payload']['vcpus']) c.volume)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_resize_finish_instance(self): def test_instance_resize_finish_instance(self):
ic = instance.Instance() ic = instance.Instance()
counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END)) counters = list(ic.process_notification(INSTANCE_FINISH_RESIZE_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, 1) self.assertEqual(1, c.volume)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_resize_finish_flavor(self): def test_instance_resize_finish_flavor(self):
ic = instance.InstanceFlavor() ic = instance.InstanceFlavor()
counters = list(ic.process_notification(INSTANCE_RESIZE_REVERT_END)) counters = list(ic.process_notification(INSTANCE_RESIZE_REVERT_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, 1) self.assertEqual(1, c.volume)
self.assertEqual(c.name, 'instance:m1.tiny') self.assertEqual('instance:m1.tiny', c.name)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_resize_finish_memory(self): def test_instance_resize_finish_memory(self):
ic = instance.Memory() ic = instance.Memory()
counters = list(ic.process_notification(INSTANCE_RESIZE_REVERT_END)) counters = list(ic.process_notification(INSTANCE_RESIZE_REVERT_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, self.assertEqual(INSTANCE_RESIZE_REVERT_END['payload']['memory_mb'],
INSTANCE_RESIZE_REVERT_END['payload']['memory_mb']) c.volume)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_resize_finish_vcpus(self): def test_instance_resize_finish_vcpus(self):
ic = instance.VCpus() ic = instance.VCpus()
counters = list(ic.process_notification(INSTANCE_RESIZE_REVERT_END)) counters = list(ic.process_notification(INSTANCE_RESIZE_REVERT_END))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
c = counters[0] c = counters[0]
self.assertEqual(c.volume, self.assertEqual(INSTANCE_RESIZE_REVERT_END['payload']['vcpus'],
INSTANCE_RESIZE_REVERT_END['payload']['vcpus']) c.volume)
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
def test_instance_delete_samples(self): def test_instance_delete_samples(self):
ic = instance.InstanceDelete() ic = instance.InstanceDelete()
counters = list(ic.process_notification(INSTANCE_DELETE_SAMPLES)) counters = list(ic.process_notification(INSTANCE_DELETE_SAMPLES))
self.assertEqual(len(counters), 2) self.assertEqual(2, len(counters))
names = [c.name for c in counters] names = [c.name for c in counters]
self.assertEqual(names, ['sample-name1', 'sample-name2']) self.assertEqual(['sample-name1', 'sample-name2'], names)
c = counters[0] c = counters[0]
self._verify_user_metadata(c.resource_metadata) self._verify_user_metadata(c.resource_metadata)
@ -728,11 +728,8 @@ class TestNotifications(test.BaseTestCase):
ic.event_types) ic.event_types)
counters = list(ic.process_notification(INSTANCE_SCHEDULED)) counters = list(ic.process_notification(INSTANCE_SCHEDULED))
self.assertEqual(len(counters), 1) self.assertEqual(1, len(counters))
names = [c.name for c in counters] names = [c.name for c in counters]
self.assertEqual(names, ['instance.scheduled']) self.assertEqual(['instance.scheduled'], names)
rid = [c.resource_id for c in counters] rid = [c.resource_id for c in counters]
self.assertEqual(rid, ['fake-uuid1-1']) self.assertEqual(['fake-uuid1-1'], rid)
c = counters[0]
metadata = c.resource_metadata['request_spec']['instance_properties']
self._verify_user_metadata(metadata)