openstack/aodh
Code Issues Proposed changes

Merge "Fix order of arguments in assertEquals"

Browse Source
This commit is contained in:
Jenkins 2014-03-21 06:48:52 +00:00 committed by Gerrit Code Review
commit c88b619911
19 changed files with 351 additions and 367 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
ceilometer/tests/compute/pollsters/test_cpu.py
View File

@ -54,11 +54,10 @@ class TestCPUPollster(base.TestPollsterBase):
def _verify_cpu_metering(expected_time):
cache = {}
samples = list(pollster.get_samples(mgr, cache, [self.instance]))
self.assertEqual(len(samples), 1)
self.assertEqual(set([s.name for s in samples]),
set(['cpu']))
self.assertEqual(samples[0].volume, expected_time)
self.assertEqual(samples[0].resource_metadata.get('cpu_number'), 2)
self.assertEqual(1, len(samples))
self.assertEqual(set(['cpu']), set([s.name for s in samples]))
self.assertEqual(expected_time, samples[0].volume)
self.assertEqual(2, samples[0].resource_metadata.get('cpu_number'))
# ensure elapsed time between polling cycles is non-zero
time.sleep(0.001)
@ -76,9 +75,9 @@ class TestCPUPollster(base.TestPollsterBase):
cache = {}
samples = list(pollster.get_samples(mgr, cache, [self.instance]))
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].volume, 10 ** 6)
self.assertEqual(len(cache), 0)
self.assertEqual(1, len(samples))
self.assertEqual(10 ** 6, samples[0].volume)
self.assertEqual(0, len(cache))
class TestCPUUtilPollster(base.TestPollsterBase):

7
ceilometer/tests/compute/pollsters/test_diskio.py
View File

@ -50,13 +50,12 @@ class TestDiskPollsters(base.TestPollsterBase):
self.assertIn(pollster.CACHE_KEY_DISK, cache)
self.assertIn(self.instance.name, cache[pollster.CACHE_KEY_DISK])
self.assertEqual(set([s.name for s in samples]),
set([name]))
self.assertEqual(set([name]), set([s.name for s in samples]))
match = [s for s in samples if s.name == name]
self.assertEqual(len(match), 1, 'missing counter %s' % name)
self.assertEqual(match[0].volume, expected_volume)
self.assertEqual(match[0].type, 'cumulative')
self.assertEqual(expected_volume, match[0].volume)
self.assertEqual('cumulative', match[0].type)
def test_disk_read_requests(self):
self._check_get_samples(disk.ReadRequestsPollster,

18
ceilometer/tests/compute/pollsters/test_instance.py
View File

@ -35,18 +35,18 @@ class TestInstancePollster(base.TestPollsterBase):
mgr = manager.AgentManager()
pollster = pollsters_instance.InstancePollster()
samples = list(pollster.get_samples(mgr, {}, [self.instance]))
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].name, 'instance')
self.assertEqual(samples[0].resource_metadata['vcpus'], 1)
self.assertEqual(samples[0].resource_metadata['memory_mb'], 512)
self.assertEqual(samples[0].resource_metadata['disk_gb'], 20)
self.assertEqual(samples[0].resource_metadata['root_gb'], 20)
self.assertEqual(samples[0].resource_metadata['ephemeral_gb'], 0)
self.assertEqual(1, len(samples))
self.assertEqual('instance', samples[0].name)
self.assertEqual(1, samples[0].resource_metadata['vcpus'])
self.assertEqual(512, samples[0].resource_metadata['memory_mb'])
self.assertEqual(20, samples[0].resource_metadata['disk_gb'])
self.assertEqual(20, samples[0].resource_metadata['root_gb'])
self.assertEqual(0, samples[0].resource_metadata['ephemeral_gb'])
@mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
def test_get_samples_instance_flavor(self):
mgr = manager.AgentManager()
pollster = pollsters_instance.InstanceFlavorPollster()
samples = list(pollster.get_samples(mgr, {}, [self.instance]))
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].name, 'instance:m1.small')
self.assertEqual(1, len(samples))
self.assertEqual('instance:m1.small', samples[0].name)

6
ceilometer/tests/compute/pollsters/test_location_metadata.py
View File

@ -90,12 +90,12 @@ class TestLocationMetadata(test.BaseTestCase):
prop = "host"
elif prop == 'OS-EXT-SRV-ATTR:instance_name':
prop = 'name'
self.assertEqual(md[prop], value)
self.assertEqual(value, md[prop])
user_metadata = md['user_metadata']
expected = self.INSTANCE_PROPERTIES[
'metadata']['metering.autoscale.group'][:256]
self.assertEqual(user_metadata['autoscale_group'], expected)
self.assertEqual(len(user_metadata), 1)
self.assertEqual(expected, user_metadata['autoscale_group'])
self.assertEqual(1, len(user_metadata))
def test_metadata_empty_image(self):
self.INSTANCE_PROPERTIES['image'] = ''

14
ceilometer/tests/compute/pollsters/test_net.py
View File

@ -73,18 +73,18 @@ class TestNetPollster(base.TestPollsterBase):
mgr = manager.AgentManager()
pollster = factory()
samples = list(pollster.get_samples(mgr, {}, [self.instance]))
self.assertEqual(len(samples), 3) # one for each nic
self.assertEqual(set([s.name for s in samples]),
set([samples[0].name]))
self.assertEqual(3, len(samples)) # one for each nic
self.assertEqual(set([samples[0].name]),
set([s.name for s in samples]))
def _verify_vnic_metering(ip, expected_volume, expected_rid):
match = [s for s in samples
if s.resource_metadata['parameters']['ip'] == ip
]
self.assertEqual(len(match), 1, 'missing ip %s' % ip)
self.assertEqual(match[0].volume, expected_volume)
self.assertEqual(match[0].type, 'cumulative')
self.assertEqual(match[0].resource_id, expected_rid)
self.assertEqual(expected_volume, match[0].volume)
self.assertEqual('cumulative', match[0].type)
self.assertEqual(expected_rid, match[0].resource_id)
for ip, volume, rid in expected:
_verify_vnic_metering(ip, volume, rid)
@ -158,7 +158,7 @@ class TestNetPollsterCache(base.TestPollsterBase):
},
}
samples = list(pollster.get_samples(mgr, cache, [self.instance]))
self.assertEqual(len(samples), 1)
self.assertEqual(1, len(samples))
def test_incoming_bytes(self):
self._check_get_samples_cache(net.IncomingBytesPollster)

4
ceilometer/tests/compute/test_manager.py
View File

@ -73,8 +73,8 @@ class TestRunTasks(agentbase.BaseAgentManagerTestCase):
def test_interval_exception_isolation(self):
super(TestRunTasks, self).test_interval_exception_isolation()
self.assertEqual(len(self.PollsterException.samples), 1)
self.assertEqual(len(self.PollsterExceptionAnother.samples), 1)
self.assertEqual(1, len(self.PollsterException.samples))
self.assertEqual(1, len(self.PollsterExceptionAnother.samples))
def test_manager_exception_persistency(self):
super(TestRunTasks, self).test_manager_exception_persistency()

84
ceilometer/tests/compute/virt/libvirt/test_inspector.py
View File

@ -58,10 +58,10 @@ class TestLibvirtInspection(test.BaseTestCase):
mock.patch.object(connection, 'lookupByID',
return_value=fake_domain)):
inspected_instances = list(self.inspector.inspect_instances())
self.assertEqual(len(inspected_instances), 1)
self.assertEqual(1, len(inspected_instances))
inspected_instance = inspected_instances[0]
self.assertEqual(inspected_instance.name, 'fake_name')
self.assertEqual(inspected_instance.UUID, 'uuid')
self.assertEqual('fake_name', inspected_instance.name)
self.assertEqual('uuid', inspected_instance.UUID)
def test_inspect_cpus(self):
with contextlib.nested(mock.patch.object(self.inspector.connection,
@ -71,8 +71,8 @@ class TestLibvirtInspection(test.BaseTestCase):
return_value=(0L, 0L, 0L,
2L, 999999L))):
cpu_info = self.inspector.inspect_cpus(self.instance_name)
self.assertEqual(cpu_info.number, 2L)
self.assertEqual(cpu_info.time, 999999L)
self.assertEqual(2L, cpu_info.number)
self.assertEqual(999999L, cpu_info.time)
def test_inspect_vnics(self):
dom_xml = """
@ -152,44 +152,42 @@ class TestLibvirtInspection(test.BaseTestCase):
2L, 999999L))):
interfaces = list(self.inspector.inspect_vnics(self.instance_name))
self.assertEqual(len(interfaces), 3)
self.assertEqual(3, len(interfaces))
vnic0, info0 = interfaces[0]
self.assertEqual(vnic0.name, 'vnet0')
self.assertEqual(vnic0.mac, 'fa:16:3e:71:ec:6d')
self.assertEqual(vnic0.fref,
'nova-instance-00000001-fa163e71ec6d')
self.assertEqual(vnic0.parameters.get('projmask'), '255.255.255.0')
self.assertEqual(vnic0.parameters.get('ip'), '10.0.0.2')
self.assertEqual(vnic0.parameters.get('projnet'), '10.0.0.0')
self.assertEqual(vnic0.parameters.get('dhcpserver'), '10.0.0.1')
self.assertEqual(info0.rx_bytes, 1L)
self.assertEqual(info0.rx_packets, 2L)
self.assertEqual(info0.tx_bytes, 3L)
self.assertEqual(info0.tx_packets, 4L)
self.assertEqual('vnet0', vnic0.name)
self.assertEqual('fa:16:3e:71:ec:6d', vnic0.mac)
self.assertEqual('nova-instance-00000001-fa163e71ec6d', vnic0.fref)
self.assertEqual('255.255.255.0', vnic0.parameters.get('projmask'))
self.assertEqual('10.0.0.2', vnic0.parameters.get('ip'))
self.assertEqual('10.0.0.0', vnic0.parameters.get('projnet'))
self.assertEqual('10.0.0.1', vnic0.parameters.get('dhcpserver'))
self.assertEqual(1L, info0.rx_bytes)
self.assertEqual(2L, info0.rx_packets)
self.assertEqual(3L, info0.tx_bytes)
self.assertEqual(4L, info0.tx_packets)
vnic1, info1 = interfaces[1]
self.assertEqual(vnic1.name, 'vnet1')
self.assertEqual(vnic1.mac, 'fa:16:3e:71:ec:6e')
self.assertEqual(vnic1.fref,
'nova-instance-00000001-fa163e71ec6e')
self.assertEqual(vnic1.parameters.get('projmask'), '255.255.255.0')
self.assertEqual(vnic1.parameters.get('ip'), '192.168.0.2')
self.assertEqual(vnic1.parameters.get('projnet'), '192.168.0.0')
self.assertEqual(vnic1.parameters.get('dhcpserver'), '192.168.0.1')
self.assertEqual(info1.rx_bytes, 5L)
self.assertEqual(info1.rx_packets, 6L)
self.assertEqual(info1.tx_bytes, 7L)
self.assertEqual(info1.tx_packets, 8L)
self.assertEqual('vnet1', vnic1.name)
self.assertEqual('fa:16:3e:71:ec:6e', vnic1.mac)
self.assertEqual('nova-instance-00000001-fa163e71ec6e', vnic1.fref)
self.assertEqual('255.255.255.0', vnic1.parameters.get('projmask'))
self.assertEqual('192.168.0.2', vnic1.parameters.get('ip'))
self.assertEqual('192.168.0.0', vnic1.parameters.get('projnet'))
self.assertEqual('192.168.0.1', vnic1.parameters.get('dhcpserver'))
self.assertEqual(5L, info1.rx_bytes)
self.assertEqual(6L, info1.rx_packets)
self.assertEqual(7L, info1.tx_bytes)
self.assertEqual(8L, info1.tx_packets)
vnic2, info2 = interfaces[2]
self.assertEqual(vnic2.name, 'vnet2')
self.assertEqual(vnic2.mac, 'fa:16:3e:96:33:f0')
self.assertEqual('vnet2', vnic2.name)
self.assertEqual('fa:16:3e:96:33:f0', vnic2.mac)
self.assertIsNone(vnic2.fref)
self.assertEqual(vnic2.parameters, dict())
self.assertEqual(info2.rx_bytes, 9L)
self.assertEqual(info2.rx_packets, 10L)
self.assertEqual(info2.tx_bytes, 11L)
self.assertEqual(info2.tx_packets, 12L)
self.assertEqual(dict(), vnic2.parameters)
self.assertEqual(9L, info2.rx_bytes)
self.assertEqual(10L, info2.rx_packets)
self.assertEqual(11L, info2.tx_bytes)
self.assertEqual(12L, info2.tx_packets)
def test_inspect_vnics_with_domain_shutoff(self):
connection = self.inspector.connection
@ -230,13 +228,13 @@ class TestLibvirtInspection(test.BaseTestCase):
2L, 999999L))):
disks = list(self.inspector.inspect_disks(self.instance_name))
self.assertEqual(len(disks), 1)
self.assertEqual(1, len(disks))
disk0, info0 = disks[0]
self.assertEqual(disk0.device, 'vda')
self.assertEqual(info0.read_requests, 1L)
self.assertEqual(info0.read_bytes, 2L)
self.assertEqual(info0.write_requests, 3L)
self.assertEqual(info0.write_bytes, 4L)
self.assertEqual('vda', disk0.device)
self.assertEqual(1L, info0.read_requests)
self.assertEqual(2L, info0.read_bytes)
self.assertEqual(3L, info0.write_requests)
self.assertEqual(4L, info0.write_bytes)
def test_inspect_disks_with_domain_shutoff(self):
connection = self.inspector.connection

18
ceilometer/tests/energy/test_kwapi.py
View File

@ -72,7 +72,7 @@ class TestKwapi(test.BaseTestCase):
pollster = kwapi.EnergyPollster()
samples = list(pollster.get_samples(self.manager, {}))
self.assertEqual(len(samples), 0)
self.assertEqual(0, len(samples))
class TestEnergyPollster(test.BaseTestCase):
@ -99,15 +99,15 @@ class TestEnergyPollster(test.BaseTestCase):
self.manager,
cache,
))
self.assertEqual(len(samples), 3)
self.assertEqual(3, len(samples))
samples_by_name = dict((s.resource_id, s) for s in samples)
for name, probe in PROBE_DICT['probes'].items():
sample = samples_by_name[name]
expected = datetime.datetime.fromtimestamp(
probe['timestamp']
).isoformat()
self.assertEqual(sample.timestamp, expected)
self.assertEqual(sample.volume, probe['kwh'])
self.assertEqual(expected, sample.timestamp)
self.assertEqual(probe['kwh'], sample.volume)
# self.assert_(
# any(map(lambda sample: sample.volume == probe['w'],
# power_samples)))
@ -132,7 +132,7 @@ class TestEnergyPollsterCache(test.BaseTestCase):
with mock.patch.object(pollster, '_get_probes') as do_not_call:
do_not_call.side_effect = AssertionError('should not be called')
samples = list(pollster.get_samples(self.manager, cache))
self.assertEqual(len(samples), 1)
self.assertEqual(1, len(samples))
class TestPowerPollster(test.BaseTestCase):
@ -159,15 +159,15 @@ class TestPowerPollster(test.BaseTestCase):
self.manager,
cache,
))
self.assertEqual(len(samples), 3)
self.assertEqual(3, len(samples))
samples_by_name = dict((s.resource_id, s) for s in samples)
for name, probe in PROBE_DICT['probes'].items():
sample = samples_by_name[name]
expected = datetime.datetime.fromtimestamp(
probe['timestamp']
).isoformat()
self.assertEqual(sample.timestamp, expected)
self.assertEqual(sample.volume, probe['w'])
self.assertEqual(expected, sample.timestamp)
self.assertEqual(probe['w'], sample.volume)
class TestPowerPollsterCache(test.BaseTestCase):
@ -189,4 +189,4 @@ class TestPowerPollsterCache(test.BaseTestCase):
with mock.patch.object(pollster, '_get_probes') as do_not_call:
do_not_call.side_effect = AssertionError('should not be called')
samples = list(pollster.get_samples(self.manager, cache))
self.assertEqual(len(samples), 1)
self.assertEqual(1, len(samples))

120
ceilometer/tests/event/test_converter.py
View File

@ -58,9 +58,9 @@ class ConverterBase(base.BaseTestCase):
"Trait %s not found in event %s" % (name, event))
trait = traits[0]
if value is not None:
self.assertEqual(trait.value, value)
self.assertEqual(value, trait.value)
if dtype is not None:
self.assertEqual(trait.dtype, dtype)
self.assertEqual(dtype, trait.dtype)
if dtype == models.Trait.INT_TYPE:
self.assertIsInstance(trait.value, int)
elif dtype == models.Trait.FLOAT_TYPE:
@ -134,9 +134,9 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.name, 'test_trait')
self.assertEqual(t.dtype, models.Trait.TEXT_TYPE)
self.assertEqual(t.value, 'foobar')
self.assertEqual('test_trait', t.name)
self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
self.assertEqual('foobar', t.value)
self.test_plugin_class.assert_called_once_with()
self.test_plugin.trait_value.assert_called_once_with([
('payload.instance_id', 'id-for-instance-0001'),
@ -151,9 +151,9 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.name, 'test_trait')
self.assertEqual(t.dtype, models.Trait.TEXT_TYPE)
self.assertEqual(t.value, 'foobar')
self.assertEqual('test_trait', t.name)
self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
self.assertEqual('foobar', t.value)
self.test_plugin_class.assert_called_once_with()
self.test_plugin.trait_value.assert_called_once_with([])
@ -180,9 +180,9 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.name, 'test_trait')
self.assertEqual(t.dtype, models.Trait.TEXT_TYPE)
self.assertEqual(t.value, 'foobar')
self.assertEqual('test_trait', t.name)
self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
self.assertEqual('foobar', t.value)
self.test_plugin_class.assert_called_once_with(a=1, b='foo')
self.test_plugin.trait_value.assert_called_once_with([
('payload.instance_id', 'id-for-instance-0001'),
@ -194,18 +194,18 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.name, 'test_trait')
self.assertEqual(t.dtype, models.Trait.TEXT_TYPE)
self.assertEqual(t.value, 'id-for-instance-0001')
self.assertEqual('test_trait', t.name)
self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
self.assertEqual('id-for-instance-0001', t.value)
cfg = dict(type='int', fields='payload.image_meta.disk_gb')
tdef = converter.TraitDefinition('test_trait', cfg,
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.name, 'test_trait')
self.assertEqual(t.dtype, models.Trait.INT_TYPE)
self.assertEqual(t.value, 20)
self.assertEqual('test_trait', t.name)
self.assertEqual(models.Trait.INT_TYPE, t.dtype)
self.assertEqual(20, t.value)
def test_to_trait_multiple(self):
cfg = dict(type='text', fields=['payload.instance_id',
@ -214,7 +214,7 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.value, 'id-for-instance-0001')
self.assertEqual('id-for-instance-0001', t.value)
cfg = dict(type='text', fields=['payload.instance_uuid',
'payload.instance_id'])
@ -222,7 +222,7 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.value, 'uuid-for-instance-0001')
self.assertEqual('uuid-for-instance-0001', t.value)
def test_to_trait_multiple_different_nesting(self):
cfg = dict(type='int', fields=['payload.foobar',
@ -231,7 +231,7 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.value, 50)
self.assertEqual(50, t.value)
cfg = dict(type='int', fields=['payload.image_meta.disk_gb',
'payload.foobar'])
@ -239,7 +239,7 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.value, 20)
self.assertEqual(20, t.value)
def test_to_trait_some_null_multiple(self):
cfg = dict(type='text', fields=['payload.instance_id2',
@ -248,7 +248,7 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.value, 'uuid-for-instance-0001')
self.assertEqual('uuid-for-instance-0001', t.value)
def test_to_trait_some_missing_multiple(self):
cfg = dict(type='text', fields=['payload.not_here_boss',
@ -257,7 +257,7 @@ class TestTraitDefinition(ConverterBase):
self.fake_plugin_mgr)
t = tdef.to_trait(self.n1)
self.assertIsInstance(t, models.Trait)
self.assertEqual(t.value, 'uuid-for-instance-0001')
self.assertEqual('uuid-for-instance-0001', t.value)
def test_to_trait_missing(self):
cfg = dict(type='text', fields='payload.not_here_boss')
@ -337,19 +337,19 @@ class TestTraitDefinition(ConverterBase):
def test_type_config(self):
cfg = dict(type='text', fields='payload.test')
t = converter.TraitDefinition('test_trait', cfg, self.fake_plugin_mgr)
self.assertEqual(t.trait_type, models.Trait.TEXT_TYPE)
self.assertEqual(models.Trait.TEXT_TYPE, t.trait_type)
cfg = dict(type='int', fields='payload.test')
t = converter.TraitDefinition('test_trait', cfg, self.fake_plugin_mgr)
self.assertEqual(t.trait_type, models.Trait.INT_TYPE)
self.assertEqual(models.Trait.INT_TYPE, t.trait_type)
cfg = dict(type='float', fields='payload.test')
t = converter.TraitDefinition('test_trait', cfg, self.fake_plugin_mgr)
self.assertEqual(t.trait_type, models.Trait.FLOAT_TYPE)
self.assertEqual(models.Trait.FLOAT_TYPE, t.trait_type)
cfg = dict(type='datetime', fields='payload.test')
t = converter.TraitDefinition('test_trait', cfg, self.fake_plugin_mgr)
self.assertEqual(t.trait_type, models.Trait.DATETIME_TYPE)
self.assertEqual(models.Trait.DATETIME_TYPE, t.trait_type)
def test_invalid_type_config(self):
#test invalid jsonpath...
@ -402,9 +402,9 @@ class TestEventDefinition(ConverterBase):
edef = converter.EventDefinition(cfg, self.fake_plugin_mgr)
e = edef.to_event(self.test_notification1)
self.assertEqual(e.event_type, 'test.thing')
self.assertEqual(e.generated,
datetime.datetime(2013, 8, 8, 21, 6, 37, 803826))
self.assertEqual('test.thing', e.event_type)
self.assertEqual(datetime.datetime(2013, 8, 8, 21, 6, 37, 803826),
e.generated)
self.assertHasDefaultTraits(e)
self.assertHasTrait(e, 'host', value='host-1-2-3', dtype=dtype)
@ -455,9 +455,9 @@ class TestEventDefinition(ConverterBase):
def test_included_type_string(self):
cfg = dict(event_type='test.thing', traits=self.traits_cfg)
edef = converter.EventDefinition(cfg, self.fake_plugin_mgr)
self.assertEqual(len(edef._included_types), 1)
self.assertEqual(edef._included_types[0], 'test.thing')
self.assertEqual(len(edef._excluded_types), 0)
self.assertEqual(1, len(edef._included_types))
self.assertEqual('test.thing', edef._included_types[0])
self.assertEqual(0, len(edef._excluded_types))
self.assertTrue(edef.included_type('test.thing'))
self.assertFalse(edef.excluded_type('test.thing'))
self.assertTrue(edef.match_type('test.thing'))
@ -467,8 +467,8 @@ class TestEventDefinition(ConverterBase):
cfg = dict(event_type=['test.thing', 'other.thing'],
traits=self.traits_cfg)
edef = converter.EventDefinition(cfg, self.fake_plugin_mgr)
self.assertEqual(len(edef._included_types), 2)
self.assertEqual(len(edef._excluded_types), 0)
self.assertEqual(2, len(edef._included_types))
self.assertEqual(0, len(edef._excluded_types))
self.assertTrue(edef.included_type('test.thing'))
self.assertTrue(edef.included_type('other.thing'))
self.assertFalse(edef.excluded_type('test.thing'))
@ -479,11 +479,11 @@ class TestEventDefinition(ConverterBase):
def test_excluded_type_string(self):
cfg = dict(event_type='!test.thing', traits=self.traits_cfg)
edef = converter.EventDefinition(cfg, self.fake_plugin_mgr)
self.assertEqual(len(edef._included_types), 1)
self.assertEqual(edef._included_types[0], '*')
self.assertEqual(edef._excluded_types[0], 'test.thing')
self.assertEqual(len(edef._excluded_types), 1)
self.assertEqual(edef._excluded_types[0], 'test.thing')
self.assertEqual(1, len(edef._included_types))
self.assertEqual('*', edef._included_types[0])
self.assertEqual('test.thing', edef._excluded_types[0])
self.assertEqual(1, len(edef._excluded_types))
self.assertEqual('test.thing', edef._excluded_types[0])
self.assertTrue(edef.excluded_type('test.thing'))
self.assertTrue(edef.included_type('random.thing'))
self.assertFalse(edef.match_type('test.thing'))
@ -493,8 +493,8 @@ class TestEventDefinition(ConverterBase):
cfg = dict(event_type=['!test.thing', '!other.thing'],
traits=self.traits_cfg)
edef = converter.EventDefinition(cfg, self.fake_plugin_mgr)
self.assertEqual(len(edef._included_types), 1)
self.assertEqual(len(edef._excluded_types), 2)
self.assertEqual(1, len(edef._included_types))
self.assertEqual(2, len(edef._excluded_types))
self.assertTrue(edef.excluded_type('test.thing'))
self.assertTrue(edef.excluded_type('other.thing'))
self.assertFalse(edef.excluded_type('random.thing'))
@ -506,8 +506,8 @@ class TestEventDefinition(ConverterBase):
cfg = dict(event_type=['*.thing', '!test.thing', '!other.thing'],
traits=self.traits_cfg)
edef = converter.EventDefinition(cfg, self.fake_plugin_mgr)
self.assertEqual(len(edef._included_types), 1)
self.assertEqual(len(edef._excluded_types), 2)
self.assertEqual(1, len(edef._included_types))
self.assertEqual(2, len(edef._excluded_types))
self.assertTrue(edef.excluded_type('test.thing'))
self.assertTrue(edef.excluded_type('other.thing'))
self.assertFalse(edef.excluded_type('random.thing'))
@ -576,8 +576,8 @@ class TestEventDefinition(ConverterBase):
traits = set(edef.traits.keys())
for dt in default_traits:
self.assertIn(dt, traits)
self.assertEqual(len(edef.traits),
len(converter.EventDefinition.DEFAULT_TRAITS))
self.assertEqual(len(converter.EventDefinition.DEFAULT_TRAITS),
len(edef.traits))
def test_traits(self):
cfg = dict(event_type='test.thing', traits=self.traits_cfg)
@ -588,8 +588,8 @@ class TestEventDefinition(ConverterBase):
self.assertIn(dt, traits)
self.assertIn('host', traits)
self.assertIn('instance_id', traits)
self.assertEqual(len(edef.traits),
len(converter.EventDefinition.DEFAULT_TRAITS) + 2)
self.assertEqual(len(converter.EventDefinition.DEFAULT_TRAITS) + 2,
len(edef.traits))
class TestNotificationConverter(ConverterBase):
@ -638,7 +638,7 @@ class TestNotificationConverter(ConverterBase):
'publisher_id': "1"}
e = c.to_event(message)
self.assertIsValidEvent(e, message)
self.assertEqual(len(e.traits), 1)
self.assertEqual(1, len(e.traits))
self.assertEqual("foo", e.event_type)
self.assertEqual(now, e.generated)
@ -647,17 +647,17 @@ class TestNotificationConverter(ConverterBase):
self.valid_event_def1,
self.fake_plugin_mgr,
add_catchall=True)
self.assertEqual(len(c.definitions), 2)
self.assertEqual(2, len(c.definitions))
e = c.to_event(self.test_notification1)
self.assertIsValidEvent(e, self.test_notification1)
self.assertEqual(len(e.traits), 3)
self.assertEqual(3, len(e.traits))
self.assertHasDefaultTraits(e)
self.assertHasTrait(e, 'instance_id')
self.assertHasTrait(e, 'host')
e = c.to_event(self.test_notification2)
self.assertIsValidEvent(e, self.test_notification2)
self.assertEqual(len(e.traits), 1)
self.assertEqual(1, len(e.traits))
self.assertHasDefaultTraits(e)
self.assertDoesNotHaveTrait(e, 'instance_id')
self.assertDoesNotHaveTrait(e, 'host')
@ -667,10 +667,10 @@ class TestNotificationConverter(ConverterBase):
self.valid_event_def1,
self.fake_plugin_mgr,
add_catchall=False)
self.assertEqual(len(c.definitions), 1)
self.assertEqual(1, len(c.definitions))
e = c.to_event(self.test_notification1)
self.assertIsValidEvent(e, self.test_notification1)
self.assertEqual(len(e.traits), 3)
self.assertEqual(3, len(e.traits))
self.assertHasDefaultTraits(e)
self.assertHasTrait(e, 'instance_id')
self.assertHasTrait(e, 'host')
@ -683,15 +683,15 @@ class TestNotificationConverter(ConverterBase):
[],
self.fake_plugin_mgr,
add_catchall=True)
self.assertEqual(len(c.definitions), 1)
self.assertEqual(1, len(c.definitions))
e = c.to_event(self.test_notification1)
self.assertIsValidEvent(e, self.test_notification1)
self.assertEqual(len(e.traits), 1)
self.assertEqual(1, len(e.traits))
self.assertHasDefaultTraits(e)
e = c.to_event(self.test_notification2)
self.assertIsValidEvent(e, self.test_notification2)
self.assertEqual(len(e.traits), 1)
self.assertEqual(1, len(e.traits))
self.assertHasDefaultTraits(e)
def test_converter_empty_cfg_without_catchall(self):
@ -699,7 +699,7 @@ class TestNotificationConverter(ConverterBase):
[],
self.fake_plugin_mgr,
add_catchall=False)
self.assertEqual(len(c.definitions), 0)
self.assertEqual(0, len(c.definitions))
e = c.to_event(self.test_notification1)
self.assertIsNotValidEvent(e, self.test_notification1)
@ -723,7 +723,7 @@ class TestNotificationConverter(ConverterBase):
with mock.patch('os.path.exists', mock_exists):
c = converter.setup_events(self.fake_plugin_mgr)
self.assertIsInstance(c, converter.NotificationEventsConverter)
self.assertEqual(len(c.definitions), 1)
self.assertEqual(1, len(c.definitions))
self.assertTrue(c.definitions[0].is_catchall)
oslo_cfg.CONF.set_override('drop_unmatched_notifications',
@ -732,4 +732,4 @@ class TestNotificationConverter(ConverterBase):
with mock.patch('os.path.exists', mock_exists):
c = converter.setup_events(self.fake_plugin_mgr)
self.assertIsInstance(c, converter.NotificationEventsConverter)
self.assertEqual(len(c.definitions), 0)
self.assertEqual(0, len(c.definitions))

16
ceilometer/tests/event/test_trait_plugins.py
View File

@ -31,26 +31,26 @@ class TestSplitterPlugin(test.BaseTestCase):
plugin = self.pclass(**param)
match_list = [('test.thing', 'test-foobar-baz')]
value = plugin.trait_value(match_list)
self.assertEqual(value, 'test')
self.assertEqual('test', value)
param = dict(separator='-', segment=1)
plugin = self.pclass(**param)
match_list = [('test.thing', 'test-foobar-baz')]
value = plugin.trait_value(match_list)
self.assertEqual(value, 'foobar')
self.assertEqual('foobar', value)
param = dict(separator='-', segment=1, max_split=1)
plugin = self.pclass(**param)
match_list = [('test.thing', 'test-foobar-baz')]
value = plugin.trait_value(match_list)
self.assertEqual(value, 'foobar-baz')
self.assertEqual('foobar-baz', value)
def test_no_sep(self):
param = dict(separator='-', segment=0)
plugin = self.pclass(**param)
match_list = [('test.thing', 'test.foobar.baz')]
value = plugin.trait_value(match_list)
self.assertEqual(value, 'test.foobar.baz')
self.assertEqual('test.foobar.baz', value)
def test_no_segment(self):
param = dict(separator='-', segment=5)
@ -90,7 +90,7 @@ class TestBitfieldPlugin(test.BaseTestCase):
plugin = self.pclass(**self.params)
value = plugin.trait_value(match_list)
self.assertEqual(value, 0x412)
self.assertEqual(0x412, value)
def test_initial(self):
match_list = [('payload.foo', 12),
@ -99,13 +99,13 @@ class TestBitfieldPlugin(test.BaseTestCase):
self.params['initial_bitfield'] = 0x2000
plugin = self.pclass(**self.params)
value = plugin.trait_value(match_list)
self.assertEqual(value, 0x2412)
self.assertEqual(0x2412, value)
def test_no_match(self):
match_list = []
plugin = self.pclass(**self.params)
value = plugin.trait_value(match_list)
self.assertEqual(value, self.init)
self.assertEqual(self.init, value)
def test_multi(self):
match_list = [('payload.foo', 12),
@ -115,4 +115,4 @@ class TestBitfieldPlugin(test.BaseTestCase):
plugin = self.pclass(**self.params)
value = plugin.trait_value(match_list)
self.assertEqual(value, 0x412)
self.assertEqual(0x412, value)

16
ceilometer/tests/image/test_glance.py
View File

@ -139,7 +139,7 @@ class TestImagePollster(test.BaseTestCase):
# list when there is nothing in the cache
images = list(glance.ImagePollster().
_iter_images(self.manager.keystone, {}))
self.assertEqual(len(images), len(set(image.id for image in images)))
self.assertEqual(len(set(image.id for image in images)), len(images))
def test_iter_images_cached(self):
# Tests whether the iter_images method returns the values from
@ -147,18 +147,18 @@ class TestImagePollster(test.BaseTestCase):
cache = {'images': []}
images = list(glance.ImagePollster().
_iter_images(self.manager.keystone, cache))
self.assertEqual(images, [])
self.assertEqual([], images)
def test_image(self):
samples = list(glance.ImagePollster().get_samples(self.manager, {}))
self.assertEqual(len(samples), 3)
self.assertEqual(3, len(samples))
for sample in samples:
self.assertEqual(sample.volume, 1)
self.assertEqual(1, sample.volume)
def test_image_size(self):
samples = list(glance.ImageSizePollster().get_samples(self.manager,
{}))
self.assertEqual(len(samples), 3)
self.assertEqual(3, len(samples))
for image in IMAGE_LIST:
self.assertTrue(
any(map(lambda sample: sample.volume == image.size,
@ -166,11 +166,9 @@ class TestImagePollster(test.BaseTestCase):
def test_image_get_sample_names(self):
samples = list(glance.ImagePollster().get_samples(self.manager, {}))
self.assertEqual(set([s.name for s in samples]),
set(['image']))
self.assertEqual(set(['image']), set([s.name for s in samples]))
def test_image_size_get_sample_names(self):
samples = list(glance.ImageSizePollster().get_samples(self.manager,
{}))
self.assertEqual(set([s.name for s in samples]),
set(['image.size']))
self.assertEqual(set(['image.size']), set([s.name for s in samples]))

66
ceilometer/tests/image/test_notifications.py
View File

@ -94,106 +94,106 @@ class TestNotification(test.BaseTestCase):
def _verify_common_counter(self, c, name, volume):
self.assertIsNotNone(c)
self.assertEqual(c.name, name)
self.assertEqual(c.resource_id, fake_uuid('c'))
self.assertEqual(c.timestamp, NOW)
self.assertEqual(c.volume, volume)
self.assertEqual(fake_uuid('c'), c.resource_id)
self.assertEqual(NOW, c.timestamp)
self.assertEqual(volume, c.volume)
metadata = c.resource_metadata
self.assertEqual(metadata.get('host'), u'images.example.com')
self.assertEqual(u'images.example.com', metadata.get('host'))
def test_image_download(self):
handler = notifications.ImageDownload()
counters = list(handler.process_notification(NOTIFICATION_SEND))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
download = counters[0]
self._verify_common_counter(download, 'image.download', 42)
self.assertEqual(download.user_id, fake_uuid('d'))
self.assertEqual(download.project_id, fake_uuid('b'))
self.assertEqual(download.type, sample.TYPE_DELTA)
self.assertEqual(fake_uuid('d'), download.user_id)
self.assertEqual(fake_uuid('b'), download.project_id)
self.assertEqual(sample.TYPE_DELTA, download.type)
def test_image_serve(self):
handler = notifications.ImageServe()
counters = list(handler.process_notification(NOTIFICATION_SEND))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
serve = counters[0]
self._verify_common_counter(serve, 'image.serve', 42)
self.assertEqual(serve.project_id, fake_uuid('e'))
self.assertEqual(serve.resource_metadata.get('receiver_user_id'),
fake_uuid('d'))
self.assertEqual(serve.resource_metadata.get('receiver_tenant_id'),
fake_uuid('b'))
self.assertEqual(serve.type, sample.TYPE_DELTA)
self.assertEqual(fake_uuid('e'), serve.project_id)
self.assertEqual(fake_uuid('d'),
serve.resource_metadata.get('receiver_user_id'))
self.assertEqual(fake_uuid('b'),
serve.resource_metadata.get('receiver_tenant_id'))
self.assertEqual(sample.TYPE_DELTA, serve.type)
def test_image_crud_on_update(self):
handler = notifications.ImageCRUD()
counters = list(handler.process_notification(NOTIFICATION_UPDATE))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
update = counters[0]
self._verify_common_counter(update, 'image.update', 1)
self.assertEqual(update.type, sample.TYPE_DELTA)
self.assertEqual(sample.TYPE_DELTA, update.type)
def test_image_on_update(self):
handler = notifications.Image()
counters = list(handler.process_notification(NOTIFICATION_UPDATE))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
update = counters[0]
self._verify_common_counter(update, 'image', 1)
self.assertEqual(update.type, sample.TYPE_GAUGE)
self.assertEqual(sample.TYPE_GAUGE, update.type)
def test_image_size_on_update(self):
handler = notifications.ImageSize()
counters = list(handler.process_notification(NOTIFICATION_UPDATE))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
update = counters[0]
self._verify_common_counter(update, 'image.size',
IMAGE_META['size'])
self.assertEqual(update.type, sample.TYPE_GAUGE)
self.assertEqual(sample.TYPE_GAUGE, update.type)
def test_image_crud_on_upload(self):
handler = notifications.ImageCRUD()
counters = list(handler.process_notification(NOTIFICATION_UPLOAD))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
upload = counters[0]
self._verify_common_counter(upload, 'image.upload', 1)
self.assertEqual(upload.type, sample.TYPE_DELTA)
self.assertEqual(sample.TYPE_DELTA, upload.type)
def test_image_on_upload(self):
handler = notifications.Image()
counters = list(handler.process_notification(NOTIFICATION_UPLOAD))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
upload = counters[0]
self._verify_common_counter(upload, 'image', 1)
self.assertEqual(upload.type, sample.TYPE_GAUGE)
self.assertEqual(sample.TYPE_GAUGE, upload.type)
def test_image_size_on_upload(self):
handler = notifications.ImageSize()
counters = list(handler.process_notification(NOTIFICATION_UPLOAD))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
upload = counters[0]
self._verify_common_counter(upload, 'image.size',
IMAGE_META['size'])
self.assertEqual(upload.type, sample.TYPE_GAUGE)
self.assertEqual(sample.TYPE_GAUGE, upload.type)
def test_image_crud_on_delete(self):
handler = notifications.ImageCRUD()
counters = list(handler.process_notification(NOTIFICATION_DELETE))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
delete = counters[0]
self._verify_common_counter(delete, 'image.delete', 1)
self.assertEqual(delete.type, sample.TYPE_DELTA)
self.assertEqual(sample.TYPE_DELTA, delete.type)
def test_image_on_delete(self):
handler = notifications.Image()
counters = list(handler.process_notification(NOTIFICATION_DELETE))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
delete = counters[0]
self._verify_common_counter(delete, 'image', 1)
self.assertEqual(delete.type, sample.TYPE_GAUGE)
self.assertEqual(sample.TYPE_GAUGE, delete.type)
def test_image_size_on_delete(self):
handler = notifications.ImageSize()
counters = list(handler.process_notification(NOTIFICATION_DELETE))
self.assertEqual(len(counters), 1)
self.assertEqual(1, len(counters))
delete = counters[0]
self._verify_common_counter(delete, 'image.size',
IMAGE_META['size'])
self.assertEqual(delete.type, sample.TYPE_GAUGE)
self.assertEqual(sample.TYPE_GAUGE, delete.type)

25
ceilometer/tests/network/test_floatingip.py
View File

@ -69,28 +69,27 @@ class TestFloatingIPPollster(test.BaseTestCase):
def test_get_samples_not_empty(self):
samples = list(self.pollster.get_samples(self.manager, {}))
self.assertEqual(len(samples), 3)
self.assertEqual(3, len(samples))
# It's necessary to verify all the attributes extracted by Nova
# API /os-floating-ips to make sure they're available and correct.
self.assertEqual(samples[0].resource_id, 1)
self.assertEqual(samples[0].resource_metadata["address"], "1.1.1.1")
self.assertEqual(samples[0].resource_metadata["pool"], "public")
self.assertEqual(1, samples[0].resource_id)
self.assertEqual("1.1.1.1", samples[0].resource_metadata["address"])
self.assertEqual("public", samples[0].resource_metadata["pool"])
self.assertEqual(samples[1].resource_id, 2)
self.assertEqual(samples[1].resource_metadata["address"], "1.1.1.2")
self.assertEqual(samples[1].resource_metadata["pool"], "public")
self.assertEqual(2, samples[1].resource_id)
self.assertEqual("1.1.1.2", samples[1].resource_metadata["address"])
self.assertEqual("public", samples[1].resource_metadata["pool"])
self.assertEqual(samples[2].resource_id, 3)
self.assertEqual(samples[2].resource_metadata["address"], "1.1.1.3")
self.assertEqual(samples[2].resource_metadata["pool"], "public")
self.assertEqual(3, samples[2].resource_id)
self.assertEqual("1.1.1.3", samples[2].resource_metadata["address"])
self.assertEqual("public", samples[2].resource_metadata["pool"])
def test_get_meter_names(self):
samples = list(self.pollster.get_samples(self.manager, {}))
self.assertEqual(set([s.name for s in samples]),
set(['ip.floating']))
self.assertEqual(set(['ip.floating']), set([s.name for s in samples]))
def test_get_samples_cached(self):
cache = {}
cache['floating_ips'] = self.fake_get_ips()[:2]
samples = list(self.pollster.get_samples(self.manager, cache))
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))

28
ceilometer/tests/network/test_notifications.py
View File

@ -252,42 +252,42 @@ class TestNotifications(test.BaseTestCase):
def test_network_create(self):
v = notifications.Network()
samples = list(v.process_notification(NOTIFICATION_NETWORK_CREATE))
self.assertEqual(len(samples), 2)
self.assertEqual(samples[1].name, "network.create")
self.assertEqual(2, len(samples))
self.assertEqual("network.create", samples[1].name)
def test_subnet_create(self):
v = notifications.Subnet()
samples = list(v.process_notification(NOTIFICATION_SUBNET_CREATE))
self.assertEqual(len(samples), 2)
self.assertEqual(samples[1].name, "subnet.create")
self.assertEqual(2, len(samples))
self.assertEqual("subnet.create", samples[1].name)
def test_port_create(self):
v = notifications.Port()
samples = list(v.process_notification(NOTIFICATION_PORT_CREATE))
self.assertEqual(len(samples), 2)
self.assertEqual(samples[1].name, "port.create")
self.assertEqual(2, len(samples))
self.assertEqual("port.create", samples[1].name)
def test_port_update(self):
v = notifications.Port()
samples = list(v.process_notification(NOTIFICATION_PORT_UPDATE))
self.assertEqual(len(samples), 2)
self.assertEqual(samples[1].name, "port.update")
self.assertEqual(2, len(samples))
self.assertEqual("port.update", samples[1].name)
def test_network_exists(self):
v = notifications.Network()
samples = v.process_notification(NOTIFICATION_NETWORK_EXISTS)
self.assertEqual(len(list(samples)), 1)
self.assertEqual(1, len(list(samples)))
def test_router_exists(self):
v = notifications.Router()
samples = v.process_notification(NOTIFICATION_ROUTER_EXISTS)
self.assertEqual(len(list(samples)), 1)
self.assertEqual(1, len(list(samples)))
def test_floatingip_exists(self):
v = notifications.FloatingIP()
samples = list(v.process_notification(NOTIFICATION_FLOATINGIP_EXISTS))
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].name, "ip.floating")
self.assertEqual(1, len(samples))
self.assertEqual("ip.floating", samples[0].name)
def test_floatingip_update(self):
v = notifications.FloatingIP()
@ -298,8 +298,8 @@ class TestNotifications(test.BaseTestCase):
def test_metering_report(self):
v = notifications.Bandwidth()
samples = list(v.process_notification(NOTIFICATION_L3_METER))
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].name, "bandwidth")
self.assertEqual(1, len(samples))
self.assertEqual("bandwidth", samples[0].name)
class TestEventTypes(test.BaseTestCase):

12
ceilometer/tests/objectstore/test_swift.py
View File

@ -108,7 +108,7 @@ class TestSwiftPollster(test.BaseTestCase):
self.assertTrue(self.pollster.CACHE_KEY_TENANT in cache)
self.assertTrue(self.pollster.CACHE_KEY_METHOD in cache)
self.assertEqual(data, [])
self.assertEqual([], data)
def test_iter_accounts_tenants_cached(self):
# Verify that if there are tenants pre-cached then the account
@ -130,7 +130,7 @@ class TestSwiftPollster(test.BaseTestCase):
}
data = list(self.pollster._iter_accounts(mock.Mock(), cache))
self.assertTrue(self.pollster.CACHE_KEY_METHOD in cache)
self.assertEqual(data[0][0], self.ACCOUNTS[0][0])
self.assertEqual(self.ACCOUNTS[0][0], data[0][0])
def test_neaten_url(self):
test_endpoint = 'http://127.0.0.1:8080'
@ -155,19 +155,19 @@ class TestSwiftPollster(test.BaseTestCase):
side_effect=self.fake_iter_accounts):
samples = list(self.pollster.get_samples(self.manager, {}))
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
def test_get_meter_names(self):
with PatchObject(self.factory, '_iter_accounts',
side_effect=self.fake_iter_accounts):
samples = list(self.pollster.get_samples(self.manager, {}))
self.assertEqual(set([s.name for s in samples]),
set([samples[0].name]))
self.assertEqual(set([samples[0].name]),
set([s.name for s in samples]))
def test_endpoint_notfound(self):
with PatchObject(self.manager.keystone.service_catalog, 'url_for',
side_effect=self.fake_ks_service_catalog_url_for):
samples = list(self.pollster.get_samples(self.manager, {}))
self.assertEqual(len(samples), 0)
self.assertEqual(0, len(samples))

134
ceilometer/tests/objectstore/test_swift_middleware.py
View File

@ -76,27 +76,27 @@ class TestSwiftMiddleware(test.BaseTestCase):
def test_rpc_setup(self):
swift_middleware.CeilometerMiddleware(FakeApp(), {})
self.assertEqual(self.CONF.control_exchange, 'ceilometer')
self.assertEqual('ceilometer', self.CONF.control_exchange)
def test_get(self):
app = swift_middleware.CeilometerMiddleware(FakeApp(), {})
req = webob.Request.blank('/1.0/account/container/obj',
environ={'REQUEST_METHOD': 'GET'})
resp = app(req.environ, self.start_response)
self.assertEqual(list(resp), ["This string is 28 bytes long"])
self.assertEqual(["This string is 28 bytes long"], list(resp))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
self.assertEqual(data.volume, 28)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(data.resource_metadata['object'], 'obj')
self.assertEqual(28, data.volume)
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertEqual('obj', data.resource_metadata['object'])
# test the # of request and the request method
data = samples[1]
self.assertEqual(data.name, 'storage.api.request')
self.assertEqual(data.volume, 1)
self.assertEqual(data.resource_metadata['method'], 'get')
self.assertEqual('storage.api.request', data.name)
self.assertEqual(1, data.volume)
self.assertEqual('get', data.resource_metadata['method'])
def test_put(self):
app = swift_middleware.CeilometerMiddleware(FakeApp(body=['']), {})
@ -106,18 +106,18 @@ class TestSwiftMiddleware(test.BaseTestCase):
six.moves.cStringIO('some stuff')})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
self.assertEqual(data.volume, 10)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(data.resource_metadata['object'], 'obj')
self.assertEqual(10, data.volume)
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertEqual('obj', data.resource_metadata['object'])
# test the # of request and the request method
data = samples[1]
self.assertEqual(data.name, 'storage.api.request')
self.assertEqual(data.volume, 1)
self.assertEqual(data.resource_metadata['method'], 'put')
self.assertEqual('storage.api.request', data.name)
self.assertEqual(1, data.volume)
self.assertEqual('put', data.resource_metadata['method'])
def test_post(self):
app = swift_middleware.CeilometerMiddleware(FakeApp(body=['']), {})
@ -127,18 +127,18 @@ class TestSwiftMiddleware(test.BaseTestCase):
'wsgi.input': six.moves.cStringIO('some other stuff')})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
self.assertEqual(data.volume, 16)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(data.resource_metadata['object'], 'obj')
self.assertEqual(16, data.volume)
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertEqual('obj', data.resource_metadata['object'])
# test the # of request and the request method
data = samples[1]
self.assertEqual(data.name, 'storage.api.request')
self.assertEqual(data.volume, 1)
self.assertEqual(data.resource_metadata['method'], 'post')
self.assertEqual('storage.api.request', data.name)
self.assertEqual(1, data.volume)
self.assertEqual('post', data.resource_metadata['method'])
def test_head(self):
app = swift_middleware.CeilometerMiddleware(FakeApp(body=['']), {})
@ -146,15 +146,15 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'HEAD'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 1)
self.assertEqual(1, len(samples))
data = samples[0]
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(data.resource_metadata['object'], 'obj')
self.assertEqual(data.resource_metadata['method'], 'head')
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertEqual('obj', data.resource_metadata['object'])
self.assertEqual('head', data.resource_metadata['method'])
self.assertEqual(data.name, 'storage.api.request')
self.assertEqual(data.volume, 1)
self.assertEqual('storage.api.request', data.name)
self.assertEqual(1, data.volume)
def test_bogus_request(self):
"""Test even for arbitrary request method, this will still work."""
@ -164,15 +164,15 @@ class TestSwiftMiddleware(test.BaseTestCase):
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 1)
self.assertEqual(1, len(samples))
data = samples[0]
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(data.resource_metadata['object'], 'obj')
self.assertEqual(data.resource_metadata['method'], 'bogus')
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertEqual('obj', data.resource_metadata['object'])
self.assertEqual('bogus', data.resource_metadata['method'])
self.assertEqual(data.name, 'storage.api.request')
self.assertEqual(data.volume, 1)
self.assertEqual('storage.api.request', data.name)
self.assertEqual(1, data.volume)
def test_get_container(self):
app = swift_middleware.CeilometerMiddleware(FakeApp(), {})
@ -180,11 +180,11 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
self.assertEqual(data.volume, 28)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(28, data.volume)
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertIsNone(data.resource_metadata['object'])
def test_no_metadata_headers(self):
@ -193,13 +193,13 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
http_headers = [k for k in data.resource_metadata.keys()
if k.startswith('http_header_')]
self.assertEqual(len(http_headers), 0)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(0, len(http_headers))
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertIsNone(data.resource_metadata['object'])
def test_metadata_headers(self):
@ -212,18 +212,18 @@ class TestSwiftMiddleware(test.BaseTestCase):
'X_VAR2': 'value2'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
http_headers = [k for k in data.resource_metadata.keys()
if k.startswith('http_header_')]
self.assertEqual(len(http_headers), 2)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(2, len(http_headers))
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertIsNone(data.resource_metadata['object'])
self.assertEqual(data.resource_metadata['http_header_x_var1'],
'value1')
self.assertEqual(data.resource_metadata['http_header_x_var2'],
'value2')
self.assertEqual('value1',
data.resource_metadata['http_header_x_var1'])
self.assertEqual('value2',
data.resource_metadata['http_header_x_var2'])
self.assertFalse('http_header_x_var3' in data.resource_metadata)
def test_metadata_headers_on_not_existing_header(self):
@ -234,13 +234,13 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 2)
self.assertEqual(2, len(samples))
data = samples[0]
http_headers = [k for k in data.resource_metadata.keys()
if k.startswith('http_header_')]
self.assertEqual(len(http_headers), 0)
self.assertEqual(data.resource_metadata['version'], '1.0')
self.assertEqual(data.resource_metadata['container'], 'container')
self.assertEqual(0, len(http_headers))
self.assertEqual('1.0', data.resource_metadata['version'])
self.assertEqual('container', data.resource_metadata['container'])
self.assertIsNone(data.resource_metadata['object'])
def test_bogus_path(self):
@ -249,14 +249,14 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 0)
self.assertEqual(0, len(samples))
def test_missing_resource_id(self):
app = swift_middleware.CeilometerMiddleware(FakeApp(), {})
req = webob.Request.blank('/5.0/', environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 0)
self.assertEqual(0, len(samples))
@mock.patch.object(swift_middleware.CeilometerMiddleware,
'publish_sample')
@ -267,8 +267,8 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
resp = list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples
self.assertEqual(len(samples), 0)
self.assertEqual(resp, ["test"])
self.assertEqual(0, len(samples))
self.assertEqual(["test"], resp)
mocked_publish_sample.assert_called_once_with(mock.ANY, 0, 4)
def test_reseller_prefix(self):
@ -278,7 +278,7 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples[0]
self.assertEqual(samples.resource_id, "account")
self.assertEqual("account", samples.resource_id)
# Custom reseller prefix set
app = swift_middleware.CeilometerMiddleware(
@ -287,7 +287,7 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples[0]
self.assertEqual(samples.resource_id, "account")
self.assertEqual("account", samples.resource_id)
def test_invalid_reseller_prefix(self):
# Custom reseller prefix set, but without trailing underscore
@ -297,4 +297,4 @@ class TestSwiftMiddleware(test.BaseTestCase):
environ={'REQUEST_METHOD': 'GET'})
list(app(req.environ, self.start_response))
samples = self.pipeline_manager.pipelines[0].samples[0]
self.assertEqual(samples.resource_id, "account")
self.assertEqual("account", samples.resource_id)

12
ceilometer/tests/publisher/test_file.py
View File

@ -81,9 +81,9 @@ class TestFilePublisher(test.BaseTestCase):
handler = publisher.publisher_logger.handlers[0]
self.assertIsInstance(handler,
logging.handlers.RotatingFileHandler)
self.assertEqual([handler.maxBytes, handler.baseFilename,
handler.backupCount],
[50, name, 3])
self.assertEqual([50, name, 3], [handler.maxBytes,
handler.baseFilename,
handler.backupCount])
# The rotating file gets created since only allow 50 bytes.
self.assertTrue(os.path.exists('%s.1' % name))
@ -99,9 +99,9 @@ class TestFilePublisher(test.BaseTestCase):
handler = publisher.publisher_logger.handlers[0]
self.assertIsInstance(handler,
logging.handlers.RotatingFileHandler)
self.assertEqual([handler.maxBytes, handler.baseFilename,
handler.backupCount],
[0, name, 0])
self.assertEqual([0, name, 0], [handler.maxBytes,
handler.baseFilename,
handler.backupCount])
# Test the content is corrected saved in the file
self.assertTrue(os.path.exists(name))
with open(name, 'r') as f:

108
ceilometer/tests/publisher/test_rpc_publisher.py
View File

@ -114,38 +114,37 @@ class TestPublish(test.BaseTestCase):
network_utils.urlsplit('rpc://'))
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 1)
self.assertEqual(self.published[0][0],
self.CONF.publisher_rpc.metering_topic)
self.assertEqual(1, len(self.published))
self.assertEqual(self.CONF.publisher_rpc.metering_topic,
self.published[0][0])
self.assertIsInstance(self.published[0][1]['args']['data'], list)
self.assertEqual(self.published[0][1]['method'],
'record_metering_data')
self.assertEqual('record_metering_data',
self.published[0][1]['method'])
def test_publish_target(self):
publisher = rpc.RPCPublisher(
network_utils.urlsplit('rpc://?target=custom_procedure_call'))
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 1)
self.assertEqual(self.published[0][0],
self.CONF.publisher_rpc.metering_topic)
self.assertEqual(1, len(self.published))
self.assertEqual(self.CONF.publisher_rpc.metering_topic,
self.published[0][0])
self.assertIsInstance(self.published[0][1]['args']['data'], list)
self.assertEqual(self.published[0][1]['method'],
'custom_procedure_call')
self.assertEqual('custom_procedure_call',
self.published[0][1]['method'])
def test_published_with_per_meter_topic(self):
publisher = rpc.RPCPublisher(
network_utils.urlsplit('rpc://?per_meter_topic=1'))
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 4)
self.assertEqual(4, len(self.published))
for topic, rpc_call in self.published:
meters = rpc_call['args']['data']
self.assertIsInstance(meters, list)
if topic != self.CONF.publisher_rpc.metering_topic:
self.assertEqual(len(set(meter['counter_name']
for meter in meters)),
1,
self.assertEqual(1, len(set(meter['counter_name']
for meter in meters)),
"Meter are published grouped by name")
topics = [topic for topic, meter in self.published]
@ -184,9 +183,9 @@ class TestPublish(test.BaseTestCase):
job1.wait()
job2.wait()
self.assertEqual(publisher.policy, 'default')
self.assertEqual(len(self.published), 2)
self.assertEqual(len(publisher.local_queue), 0)
self.assertEqual('default', publisher.policy)
self.assertEqual(2, len(self.published))
self.assertEqual(0, len(publisher.local_queue))
@mock.patch('ceilometer.publisher.rpc.LOG')
def test_published_with_no_policy(self, mylog):
@ -199,9 +198,9 @@ class TestPublish(test.BaseTestCase):
SystemExit,
publisher.publish_samples,
None, self.test_data)
self.assertEqual(publisher.policy, 'default')
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 0)
self.assertEqual('default', publisher.policy)
self.assertEqual(0, len(self.published))
self.assertEqual(0, len(publisher.local_queue))
@mock.patch('ceilometer.publisher.rpc.LOG')
def test_published_with_policy_block(self, mylog):
@ -213,8 +212,8 @@ class TestPublish(test.BaseTestCase):
SystemExit,
publisher.publish_samples,
None, self.test_data)
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 0)
self.assertEqual(0, len(self.published))
self.assertEqual(0, len(publisher.local_queue))
@mock.patch('ceilometer.publisher.rpc.LOG')
def test_published_with_policy_incorrect(self, mylog):
@ -226,9 +225,9 @@ class TestPublish(test.BaseTestCase):
publisher.publish_samples,
None, self.test_data)
self.assertTrue(mylog.warn.called)
self.assertEqual(publisher.policy, 'default')
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 0)
self.assertEqual('default', publisher.policy)
self.assertEqual(0, len(self.published))
self.assertEqual(0, len(publisher.local_queue))
def test_published_with_policy_drop_and_rpc_down(self):
self.rpc_unreachable = True
@ -236,8 +235,8 @@ class TestPublish(test.BaseTestCase):
network_utils.urlsplit('rpc://?policy=drop'))
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 0)
self.assertEqual(0, len(self.published))
self.assertEqual(0, len(publisher.local_queue))
def test_published_with_policy_queue_and_rpc_down(self):
self.rpc_unreachable = True
@ -245,8 +244,8 @@ class TestPublish(test.BaseTestCase):
network_utils.urlsplit('rpc://?policy=queue'))
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 1)
self.assertEqual(0, len(self.published))
self.assertEqual(1, len(publisher.local_queue))
def test_published_with_policy_queue_and_rpc_down_up(self):
self.rpc_unreachable = True
@ -254,15 +253,15 @@ class TestPublish(test.BaseTestCase):
network_utils.urlsplit('rpc://?policy=queue'))
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 1)
self.assertEqual(0, len(self.published))
self.assertEqual(1, len(publisher.local_queue))
self.rpc_unreachable = False
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 2)
self.assertEqual(len(publisher.local_queue), 0)
self.assertEqual(2, len(self.published))
self.assertEqual(0, len(publisher.local_queue))
def test_published_with_policy_sized_queue_and_rpc_down(self):
self.rpc_unreachable = True
@ -273,20 +272,17 @@ class TestPublish(test.BaseTestCase):
s.source = 'test-%d' % i
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 3)
self.assertEqual(
publisher.local_queue[0][2]['args']['data'][0]['source'],
'test-2'
)
self.assertEqual(
publisher.local_queue[1][2]['args']['data'][0]['source'],
'test-3'
)
self.assertEqual(
publisher.local_queue[2][2]['args']['data'][0]['source'],
'test-4'
)
self.assertEqual(0, len(self.published))
self.assertEqual(3, len(publisher.local_queue))
self.assertEqual('test-2',
publisher.local_queue[0][2]['args']['data'][0]
['source'])
self.assertEqual('test-3',
publisher.local_queue[1][2]['args']['data'][0]
['source'])
self.assertEqual('test-4',
publisher.local_queue[2][2]['args']['data'][0]
['source'])
def test_published_with_policy_default_sized_queue_and_rpc_down(self):
self.rpc_unreachable = True
@ -297,13 +293,11 @@ class TestPublish(test.BaseTestCase):
s.source = 'test-%d' % i
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.published), 0)
self.assertEqual(len(publisher.local_queue), 1024)
self.assertEqual(
publisher.local_queue[0][2]['args']['data'][0]['source'],
'test-976'
)
self.assertEqual(
publisher.local_queue[1023][2]['args']['data'][0]['source'],
'test-1999'
)
self.assertEqual(0, len(self.published))
self.assertEqual(1024, len(publisher.local_queue))
self.assertEqual('test-976',
publisher.local_queue[0][2]['args']['data'][0]
['source'])
self.assertEqual('test-1999',
publisher.local_queue[1023][2]['args']['data'][0]
['source'])

15
ceilometer/tests/publisher/test_udp.py
View File

@ -123,7 +123,7 @@ class TestUDPPublisher(test.BaseTestCase):
publisher.publish_samples(None,
self.test_data)
self.assertEqual(len(self.data_sent), 5)
self.assertEqual(5, len(self.data_sent))
sent_counters = []
@ -132,16 +132,13 @@ class TestUDPPublisher(test.BaseTestCase):
sent_counters.append(counter)
# Check destination
self.assertEqual(dest, ('somehost',
self.CONF.collector.udp_port))
self.assertEqual(('somehost',
self.CONF.collector.udp_port), dest)
# Check that counters are equal
self.assertEqual(sorted(sent_counters),
sorted(
[utils.meter_message_from_counter(
d,
"not-so-secret")
for d in self.test_data]))
self.assertEqual(sorted(
[utils.meter_message_from_counter(d, "not-so-secret")
for d in self.test_data]), sorted(sent_counters))
@staticmethod
def _raise_ioerror(*args):