openstack/aodh
Code Issues Proposed changes

Merge "Fix order of arguments in assertEquals"

Browse Source
This commit is contained in:
Jenkins 2014-03-27 02:08:57 +00:00 committed by Gerrit Code Review
commit 43b74b232b
23 changed files with 334 additions and 334 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
ceilometer/tests/agentbase.py
View File

@ -265,16 +265,16 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
def test_setup_polling_tasks(self):
polling_tasks = self.mgr.setup_polling_tasks()
self.assertEqual(len(polling_tasks), 1)
self.assertEqual(1, len(polling_tasks))
self.assertTrue(60 in polling_tasks.keys())
per_task_resources = polling_tasks[60].resources
self.assertEqual(len(per_task_resources), 1)
self.assertEqual(set(per_task_resources['test'].resources),
set(self.pipeline_cfg[0]['resources']))
self.assertEqual(1, len(per_task_resources))
self.assertEqual(set(self.pipeline_cfg[0]['resources']),
set(per_task_resources['test'].resources))
self.mgr.interval_task(polling_tasks.values()[0])
pub = self.mgr.pipeline_manager.pipelines[0].publishers[0]
del pub.samples[0].resource_metadata['resources']
self.assertEqual(pub.samples[0], self.Pollster.test_data)
self.assertEqual(self.Pollster.test_data, pub.samples[0])
def test_setup_polling_tasks_multiple_interval(self):
self.pipeline_cfg.append({
@ -287,7 +287,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
})
self.setup_pipeline()
polling_tasks = self.mgr.setup_polling_tasks()
self.assertEqual(len(polling_tasks), 2)
self.assertEqual(2, len(polling_tasks))
self.assertTrue(60 in polling_tasks.keys())
self.assertTrue(10 in polling_tasks.keys())
@ -302,7 +302,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
'publishers': ["test"],
})
polling_tasks = self.mgr.setup_polling_tasks()
self.assertEqual(len(polling_tasks), 1)
self.assertEqual(1, len(polling_tasks))
self.assertTrue(60 in polling_tasks.keys())
def test_setup_polling_task_same_interval(self):
@ -316,15 +316,15 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
})
self.setup_pipeline()
polling_tasks = self.mgr.setup_polling_tasks()
self.assertEqual(len(polling_tasks), 1)
self.assertEqual(1, len(polling_tasks))
pollsters = polling_tasks.get(60).pollsters
self.assertEqual(len(pollsters), 2)
self.assertEqual(2, len(pollsters))
per_task_resources = polling_tasks[60].resources
self.assertEqual(len(per_task_resources), 2)
self.assertEqual(set(per_task_resources['test'].resources),
set(self.pipeline_cfg[0]['resources']))
self.assertEqual(set(per_task_resources['testanother'].resources),
set(self.pipeline_cfg[1]['resources']))
self.assertEqual(2, len(per_task_resources))
self.assertEqual(set(self.pipeline_cfg[0]['resources']),
set(per_task_resources['test'].resources))
self.assertEqual(set(self.pipeline_cfg[1]['resources']),
set(per_task_resources['testanother'].resources))
def test_interval_exception_isolation(self):
self.pipeline_cfg = [
@ -350,11 +350,11 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
self.transformer_manager)
polling_tasks = self.mgr.setup_polling_tasks()
self.assertEqual(len(polling_tasks.keys()), 1)
self.assertEqual(1, len(polling_tasks.keys()))
polling_tasks.get(10)
self.mgr.interval_task(polling_tasks.get(10))
pub = self.mgr.pipeline_manager.pipelines[0].publishers[0]
self.assertEqual(len(pub.samples), 0)
self.assertEqual(0, len(pub.samples))
def test_agent_manager_start(self):
mgr = self.create_manager()
@ -375,9 +375,9 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
self.setup_pipeline()
def _verify_discovery_params(self, expected):
self.assertEqual(self.Discovery.params, expected)
self.assertEqual(self.DiscoveryAnother.params, expected)
self.assertEqual(self.DiscoveryException.params, expected)
self.assertEqual(expected, self.Discovery.params)
self.assertEqual(expected, self.DiscoveryAnother.params)
self.assertEqual(expected, self.DiscoveryException.params)
def _do_test_per_agent_discovery(self,
discovered_resources,
@ -396,8 +396,8 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
self._verify_discovery_params([None] if discovered_resources else [])
discovery = self.Discovery.resources + self.DiscoveryAnother.resources
# compare resource lists modulo ordering
self.assertEqual(set(self.Pollster.resources),
set(static_resources or discovery))
self.assertEqual(set(static_resources or discovery),
set(self.Pollster.resources))
def test_per_agent_discovery_discovered_only(self):
self._do_test_per_agent_discovery(['discovered_1', 'discovered_2'],
@ -424,8 +424,8 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
self.setup_pipeline()
polling_tasks = self.mgr.setup_polling_tasks()
self.mgr.interval_task(polling_tasks.get(60))
self.assertEqual(set(self.Pollster.resources),
set(self.DiscoveryAnother.resources))
self.assertEqual(set(self.DiscoveryAnother.resources),
set(self.Pollster.resources))
def _do_test_per_pipeline_discovery(self,
discovered_resources,
@ -444,8 +444,8 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
self.mgr.interval_task(polling_tasks.get(60))
discovery = self.Discovery.resources + self.DiscoveryAnother.resources
# compare resource lists modulo ordering
self.assertEqual(set(self.Pollster.resources),
set(static_resources + discovery))
self.assertEqual(set(static_resources + discovery),
set(self.Pollster.resources))
def test_per_pipeline_discovery_discovered_only(self):
self._do_test_per_pipeline_discovery(['discovered_1', 'discovered_2'],
@ -476,16 +476,16 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
self.Discovery.resources = []
self.setup_pipeline()
polling_tasks = self.mgr.setup_polling_tasks()
self.assertEqual(len(polling_tasks), 1)
self.assertEqual(1, len(polling_tasks))
self.assertTrue(60 in polling_tasks.keys())
self.mgr.interval_task(polling_tasks.get(60))
self._verify_discovery_params([])
self.assertEqual(len(self.Pollster.samples), 1)
self.assertEqual(1, len(self.Pollster.samples))
amalgamated_resources = set(['test://', 'another://'])
self.assertEqual(set(self.Pollster.samples[0][1]),
amalgamated_resources)
self.assertEqual(amalgamated_resources,
set(self.Pollster.samples[0][1]))
for pipeline in self.mgr.pipeline_manager.pipelines:
self.assertEqual(len(pipeline.publishers[0].samples), 1)
self.assertEqual(1, len(pipeline.publishers[0].samples))
published = pipeline.publishers[0].samples[0]
self.assertEqual(set(published.resource_metadata['resources']),
amalgamated_resources)
self.assertEqual(amalgamated_resources,
set(published.resource_metadata['resources']))

2
ceilometer/tests/alarm/evaluator/base.py
View File

@ -50,4 +50,4 @@ class TestEvaluatorBase(test.BaseTestCase):
def _assert_all_alarms(self, state):
for alarm in self.alarms:
self.assertEqual(alarm.state, state)
self.assertEqual(state, alarm.state)

32
ceilometer/tests/alarm/evaluator/test_combination.py
View File

@ -135,7 +135,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='insufficient data')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
expected = [mock.call(
alarm,
'ok',
@ -143,7 +143,7 @@ class TestEvaluate(base.TestEvaluatorBase):
(",".join(alarm.rule['alarm_ids']))),
self._reason_data(alarm.rule['alarm_ids']))
for alarm in self.alarms]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_to_ok_with_all_ok(self):
self._set_all_alarms('insufficient data')
@ -159,7 +159,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='ok')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons, reason_datas = self._combination_transition_reason(
'ok',
self.alarms[0].rule['alarm_ids'],
@ -168,7 +168,7 @@ class TestEvaluate(base.TestEvaluatorBase):
reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_to_ok_with_one_alarm(self):
self._set_all_alarms('alarm')
@ -184,7 +184,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='ok')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons, reason_datas = self._combination_transition_reason(
'ok',
self.alarms[0].rule['alarm_ids'],
@ -192,7 +192,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'alarm', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_to_alarm_with_all_alarm(self):
self._set_all_alarms('ok')
@ -208,7 +208,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='alarm')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons, reason_datas = self._combination_transition_reason(
'alarm',
self.alarms[0].rule['alarm_ids'],
@ -216,7 +216,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'ok', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_to_alarm_with_one_ok(self):
self._set_all_alarms('ok')
@ -232,7 +232,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='alarm')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons, reason_datas = self._combination_transition_reason(
'alarm',
[self.alarms[0].rule['alarm_ids'][1]],
@ -240,7 +240,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'ok', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_to_unknown(self):
self._set_all_alarms('ok')
@ -257,7 +257,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='insufficient data')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Alarms %s are in unknown state'
% self.alarms[0].rule['alarm_ids'][0],
'Alarms %s are in unknown state'
@ -268,7 +268,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'ok', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_no_state_change(self):
self._set_all_alarms('ok')
@ -282,8 +282,8 @@ class TestEvaluate(base.TestEvaluatorBase):
]
self._evaluate_all_alarms()
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, [])
self.assertEqual(self.notifier.notify.call_args_list, [])
self.assertEqual([], update_calls)
self.assertEqual([], self.notifier.notify.call_args_list)
def test_no_state_change_and_repeat_actions(self):
self.alarms[0].repeat_actions = True
@ -299,7 +299,7 @@ class TestEvaluate(base.TestEvaluatorBase):
]
self._evaluate_all_alarms()
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, [])
self.assertEqual([], update_calls)
reasons, reason_datas = self._combination_remaining_reason(
'ok',
self.alarms[0].rule['alarm_ids'],
@ -308,7 +308,7 @@ class TestEvaluate(base.TestEvaluatorBase):
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_state_change_inside_time_constraint(self):
self._set_all_alarms('insufficient data')

63
ceilometer/tests/alarm/evaluator/test_threshold.py
View File

@ -148,7 +148,7 @@ class TestEvaluate(base.TestEvaluatorBase):
alarm.rule['evaluation_periods'],
None))
for alarm in self.alarms]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_simple_alarm_trip(self):
self._set_all_alarms('ok')
@ -164,7 +164,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='alarm')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Transition to alarm due to 5 samples outside'
' threshold, most recent: %s' % avgs[-1].avg,
'Transition to alarm due to 4 samples outside'
@ -174,7 +174,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'ok', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_simple_alarm_clear(self):
self._set_all_alarms('alarm')
@ -190,7 +190,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='ok')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Transition to ok due to 5 samples inside'
' threshold, most recent: %s' % avgs[-1].avg,
'Transition to ok due to 4 samples inside'
@ -200,7 +200,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'alarm', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_equivocal_from_known_state(self):
self._set_all_alarms('ok')
@ -213,50 +213,55 @@ class TestEvaluate(base.TestEvaluatorBase):
self.api_client.statistics.list.side_effect = [avgs, maxs]
self._evaluate_all_alarms()
self._assert_all_alarms('ok')
self.assertEqual(self.api_client.alarms.set_state.call_args_list,
[])
self.assertEqual(self.notifier.notify.call_args_list, [])
self.assertEqual(
[],
self.api_client.alarms.set_state.call_args_list)
self.assertEqual([], self.notifier.notify.call_args_list)
def test_equivocal_from_known_state_and_repeat_actions(self):
self._set_all_alarms('ok')
self.alarms[1].repeat_actions = True
with mock.patch('ceilometerclient.client.get_client',
return_value=self.api_client):
avgs = [self._get_stat('avg', self.alarms[0].rule['threshold'] + v)
avgs = [self._get_stat('avg',
self.alarms[0].rule['threshold'] + v)
for v in moves.xrange(5)]
maxs = [self._get_stat('max', self.alarms[1].rule['threshold'] - v)
maxs = [self._get_stat('max',
self.alarms[1].rule['threshold'] - v)
for v in moves.xrange(-1, 3)]
self.api_client.statistics.list.side_effect = [avgs, maxs]
self._evaluate_all_alarms()
self._assert_all_alarms('ok')
self.assertEqual(self.api_client.alarms.set_state.call_args_list,
[])
self.assertEqual([],
self.api_client.alarms.set_state.call_args_list)
reason = 'Remaining as ok due to 4 samples inside' \
' threshold, most recent: 8.0'
reason_datas = self._reason_data('inside', 4, 8.0)
expected = [mock.call(self.alarms[1], 'ok', reason, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_unequivocal_from_known_state_and_repeat_actions(self):
self._set_all_alarms('alarm')
self.alarms[1].repeat_actions = True
with mock.patch('ceilometerclient.client.get_client',
return_value=self.api_client):
avgs = [self._get_stat('avg', self.alarms[0].rule['threshold'] + v)
avgs = [self._get_stat('avg',
self.alarms[0].rule['threshold'] + v)
for v in moves.xrange(1, 6)]
maxs = [self._get_stat('max', self.alarms[1].rule['threshold'] - v)
maxs = [self._get_stat('max',
self.alarms[1].rule['threshold'] - v)
for v in moves.xrange(4)]
self.api_client.statistics.list.side_effect = [avgs, maxs]
self._evaluate_all_alarms()
self._assert_all_alarms('alarm')
self.assertEqual(self.api_client.alarms.set_state.call_args_list,
[])
self.assertEqual([],
self.api_client.alarms.set_state.call_args_list)
reason = 'Remaining as alarm due to 4 samples outside' \
' threshold, most recent: 7.0'
reason_datas = self._reason_data('outside', 4, 7.0)
expected = [mock.call(self.alarms[1], 'alarm',
reason, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_state_change_and_repeat_actions(self):
self._set_all_alarms('ok')
@ -274,7 +279,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='alarm')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Transition to alarm due to 5 samples outside'
' threshold, most recent: %s' % avgs[-1].avg,
'Transition to alarm due to 4 samples outside'
@ -284,7 +289,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'ok', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_equivocal_from_unknown(self):
self._set_all_alarms('insufficient data')
@ -300,7 +305,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='alarm')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Transition to alarm due to 5 samples outside'
' threshold, most recent: %s' % avgs[-1].avg,
'Transition to alarm due to 4 samples outside'
@ -311,7 +316,7 @@ class TestEvaluate(base.TestEvaluatorBase):
reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def _do_test_bound_duration(self, start, exclude_outliers=None):
alarm = self.alarms[0]
@ -319,14 +324,14 @@ class TestEvaluate(base.TestEvaluatorBase):
alarm.rule['exclude_outliers'] = exclude_outliers
timeutils.utcnow.override_time = datetime.datetime(2012, 7, 2, 10, 45)
constraint = self.evaluator._bound_duration(alarm, [])
self.assertEqual(constraint, [
self.assertEqual([
{'field': 'timestamp',
'op': 'le',
'value': timeutils.utcnow().isoformat()},
{'field': 'timestamp',
'op': 'ge',
'value': start},
])
], constraint)
def test_bound_duration_outlier_exclusion_defaulted(self):
self._do_test_bound_duration('2012-07-02T10:39:00')
@ -357,7 +362,7 @@ class TestEvaluate(base.TestEvaluatorBase):
os_endpoint_type=conf.os_endpoint_type,
insecure=conf.insecure)]
actual = client.call_args_list
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def _do_test_simple_alarm_trip_outlier_exclusion(self, exclude_outliers):
self._set_all_rules('exclude_outliers', exclude_outliers)
@ -383,7 +388,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='alarm')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Transition to alarm due to 5 samples outside'
' threshold, most recent: %s' % avgs[-2].avg,
'Transition to alarm due to 4 samples outside'
@ -393,7 +398,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'ok', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_simple_alarm_trip_with_outlier_exclusion(self):
self. _do_test_simple_alarm_trip_outlier_exclusion(True)
@ -425,7 +430,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm.alarm_id, state='ok')
for alarm in self.alarms]
update_calls = self.api_client.alarms.set_state.call_args_list
self.assertEqual(update_calls, expected)
self.assertEqual(expected, update_calls)
reasons = ['Transition to ok due to 5 samples inside'
' threshold, most recent: %s' % avgs[-2].avg,
'Transition to ok due to 4 samples inside'
@ -435,7 +440,7 @@ class TestEvaluate(base.TestEvaluatorBase):
expected = [mock.call(alarm, 'alarm', reason, reason_data)
for alarm, reason, reason_data
in zip(self.alarms, reasons, reason_datas)]
self.assertEqual(self.notifier.notify.call_args_list, expected)
self.assertEqual(expected, self.notifier.notify.call_args_list)
def test_simple_alarm_clear_with_outlier_exclusion(self):
self. _do_test_simple_alarm_clear_outlier_exclusion(True)

39
ceilometer/tests/alarm/partition/test_coordination.py
View File

@ -153,17 +153,16 @@ class TestCoordinate(test.BaseTestCase):
target, alarms = args
self.assertTrue(target in uneffected)
uneffected.remove(target)
self.assertEqual(len(alarms), per_worker)
self.assertEqual(per_worker, len(alarms))
for aid in alarms:
self.assertTrue(aid in remainder)
remainder.remove(aid)
self.assertEqual(set(uneffected), set(expect_uneffected))
self.assertEqual(set(expect_uneffected), set(uneffected))
return remainder
def _forget_assignments(self, expected_assignments):
rpc = self.partition_coordinator.coordination_rpc
self.assertEqual(len(rpc.assign.call_args_list),
expected_assignments)
self.assertEqual(expected_assignments, len(rpc.assign.call_args_list))
rpc.reset_mock()
def test_mastership_not_assumed_during_warmup(self):
@ -237,8 +236,8 @@ class TestCoordinate(test.BaseTestCase):
self._check_mastership(True)
remainder = self._check_assignments(others, alarm_ids, 10)
self.assertEqual(set(remainder),
set(self.partition_coordinator.assignment))
self.assertEqual(set(self.partition_coordinator.assignment),
set(remainder))
def test_rebalance_on_partition_startup(self):
alarm_ids = self._some_alarms(49)
@ -255,8 +254,8 @@ class TestCoordinate(test.BaseTestCase):
self._check_mastership(True)
remainder = self._check_assignments(others, alarm_ids, 9)
self.assertEqual(set(remainder),
set(self.partition_coordinator.assignment))
self.assertEqual(set(self.partition_coordinator.assignment),
set(remainder))
def test_rebalance_on_partition_staleness(self):
alarm_ids = self._some_alarms(49)
@ -278,8 +277,8 @@ class TestCoordinate(test.BaseTestCase):
self._check_mastership(True)
remainder = self._check_assignments(others, alarm_ids, 13, [stale])
self.assertEqual(set(remainder),
set(self.partition_coordinator.assignment))
self.assertEqual(set(self.partition_coordinator.assignment),
set(remainder))
def test_rebalance_on_sufficient_deletion(self):
alarm_ids = self._some_alarms(49)
@ -297,8 +296,8 @@ class TestCoordinate(test.BaseTestCase):
self._check_mastership(True)
remainder = self._check_assignments(others, alarm_ids, 5)
self.assertEqual(set(remainder),
set(self.partition_coordinator.assignment))
self.assertEqual(set(self.partition_coordinator.assignment),
set(remainder))
def test_no_rebalance_on_insufficient_deletion(self):
alarm_ids = self._some_alarms(49)
@ -335,9 +334,9 @@ class TestCoordinate(test.BaseTestCase):
self._check_mastership(True)
remainder = self._check_allocation(others, new_alarm_ids, 2)
self.assertEqual(len(remainder), 0)
self.assertEqual(master_assignment,
set(self.partition_coordinator.assignment))
self.assertEqual(0, len(remainder))
self.assertEqual(set(self.partition_coordinator.assignment),
master_assignment)
def test_bail_when_overtaken_in_distribution(self):
self._some_alarms(49)
@ -355,11 +354,11 @@ class TestCoordinate(test.BaseTestCase):
self._check_mastership(False)
self.assertEqual(len(rpc.assign.call_args_list), 1)
self.assertEqual(1, len(rpc.assign.call_args_list))
def test_assigned_alarms_no_assignment(self):
alarms = self.partition_coordinator.assigned_alarms(self.api_client)
self.assertEqual(len(alarms), 0)
self.assertEqual(0, len(alarms))
def test_assigned_alarms_assignment(self):
alarm_ids = self._some_alarms(6)
@ -368,7 +367,7 @@ class TestCoordinate(test.BaseTestCase):
self.partition_coordinator.assign(uuid, alarm_ids)
alarms = self.partition_coordinator.assigned_alarms(self.api_client)
self.assertEqual(alarms, self._current_alarms())
self.assertEqual(self._current_alarms(), alarms)
def test_assigned_alarms_allocation(self):
alarm_ids = self._some_alarms(6)
@ -380,7 +379,7 @@ class TestCoordinate(test.BaseTestCase):
self.partition_coordinator.allocate(uuid, new_alarm_ids)
alarms = self.partition_coordinator.assigned_alarms(self.api_client)
self.assertEqual(alarms, self._current_alarms())
self.assertEqual(self._current_alarms(), alarms)
def test_assigned_alarms_deleted_assignment(self):
alarm_ids = self._some_alarms(6)
@ -391,7 +390,7 @@ class TestCoordinate(test.BaseTestCase):
self._dump_alarms(len(alarm_ids) / 2)
alarms = self.partition_coordinator.assigned_alarms(self.api_client)
self.assertEqual(alarms, self._current_alarms())
self.assertEqual(self._current_alarms(), alarms)
def test__record_oldest(self):
# Test when the partition to be recorded is the same as the oldest.

16
ceilometer/tests/alarm/test_notifier.py
View File

@ -63,14 +63,14 @@ class TestAlarmNotifier(test.BaseTestCase):
}
self.service.notify_alarm(context.get_admin_context(), data)
notifications = self.service.notifiers['test'].obj.notifications
self.assertEqual(len(notifications), 1)
self.assertEqual(notifications[0], (
urlparse.urlsplit(data['actions'][0]),
data['alarm_id'],
data['previous'],
data['current'],
data['reason'],
data['reason_data']))
self.assertEqual(1, len(notifications))
self.assertEqual((urlparse.urlsplit(data['actions'][0]),
data['alarm_id'],
data['previous'],
data['current'],
data['reason'],
data['reason_data']),
notifications[0])
def test_notify_alarm_no_action(self):
self.service.notify_alarm(context.get_admin_context(), {})

2
ceilometer/tests/alarm/test_partitioned_alarm_svc.py
View File

@ -81,7 +81,7 @@ class TestPartitionedAlarmService(test.BaseTestCase):
mock.call(604800, mock.ANY),
]
actual = self.partitioned.tg.add_timer.call_args_list
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_presence_reporting(self):
priority = 42

56
ceilometer/tests/alarm/test_rpc.py
View File

@ -82,24 +82,24 @@ class TestRPCAlarmNotifier(test.BaseTestCase):
for i, a in enumerate(self.alarms):
self.notifier.notify(a, previous[i], "what? %d" % i,
{'fire': '%d' % i})
self.assertEqual(len(self.notified), 2)
self.assertEqual(2, len(self.notified))
for i, a in enumerate(self.alarms):
actions = getattr(a, models.Alarm.ALARM_ACTIONS_MAP[a.state])
self.assertEqual(self.notified[i][0],
self.CONF.alarm.notifier_rpc_topic)
self.assertEqual(self.notified[i][1]["args"]["data"]["alarm_id"],
self.alarms[i].alarm_id)
self.assertEqual(self.notified[i][1]["args"]["data"]["actions"],
actions)
self.assertEqual(self.notified[i][1]["args"]["data"]["previous"],
previous[i])
self.assertEqual(self.notified[i][1]["args"]["data"]["current"],
self.alarms[i].state)
self.assertEqual(self.notified[i][1]["args"]["data"]["reason"],
"what? %d" % i)
self.assertEqual(self.CONF.alarm.notifier_rpc_topic,
self.notified[i][0])
self.assertEqual(self.alarms[i].alarm_id,
self.notified[i][1]["args"]["data"]["alarm_id"])
self.assertEqual(actions,
self.notified[i][1]["args"]["data"]["actions"])
self.assertEqual(previous[i],
self.notified[i][1]["args"]["data"]["previous"])
self.assertEqual(self.alarms[i].state,
self.notified[i][1]["args"]["data"]["current"])
self.assertEqual("what? %d" % i,
self.notified[i][1]["args"]["data"]["reason"])
self.assertEqual(
self.notified[i][1]["args"]["data"]["reason_data"],
{'fire': '%d' % i})
{'fire': '%d' % i},
self.notified[i][1]["args"]["data"]["reason_data"])
def test_notify_non_string_reason(self):
self.notifier.notify(self.alarms[0], 'ok', 42, {})
@ -124,7 +124,7 @@ class TestRPCAlarmNotifier(test.BaseTestCase):
'my_instance'}
})
self.notifier.notify(alarm, 'alarm', "what?", {})
self.assertEqual(len(self.notified), 0)
self.assertEqual(0, len(self.notified))
class TestRPCAlarmPartitionCoordination(test.BaseTestCase):
@ -145,25 +145,25 @@ class TestRPCAlarmPartitionCoordination(test.BaseTestCase):
priority = float(timeutils.utcnow().strftime('%s.%f'))
self.ordination.presence(id, priority)
topic, msg = self.notified[0]
self.assertEqual(topic, 'alarm_partition_coordination')
self.assertEqual(msg['args']['data']['uuid'], id)
self.assertEqual(msg['args']['data']['priority'], priority)
self.assertEqual(msg['method'], 'presence')
self.assertEqual('alarm_partition_coordination', topic)
self.assertEqual(id, msg['args']['data']['uuid'])
self.assertEqual(priority, msg['args']['data']['priority'])
self.assertEqual('presence', msg['method'])
def test_ordination_assign(self):
id = uuid.uuid4()
self.ordination.assign(id, self.alarms)
topic, msg = self.notified[0]
self.assertEqual(topic, 'alarm_partition_coordination')
self.assertEqual(msg['args']['data']['uuid'], id)
self.assertEqual(len(msg['args']['data']['alarms']), 2)
self.assertEqual(msg['method'], 'assign')
self.assertEqual('alarm_partition_coordination', topic)
self.assertEqual(id, msg['args']['data']['uuid'])
self.assertEqual(2, len(msg['args']['data']['alarms']))
self.assertEqual('assign', msg['method'])
def test_ordination_allocate(self):
id = uuid.uuid4()
self.ordination.allocate(id, self.alarms)
topic, msg = self.notified[0]
self.assertEqual(topic, 'alarm_partition_coordination')
self.assertEqual(msg['args']['data']['uuid'], id)
self.assertEqual(len(msg['args']['data']['alarms']), 2)
self.assertEqual(msg['method'], 'allocate')
self.assertEqual('alarm_partition_coordination', topic)
self.assertEqual(id, msg['args']['data']['uuid'])
self.assertEqual(2, len(msg['args']['data']['alarms']))
self.assertEqual('allocate', msg['method'])

4
ceilometer/tests/alarm/test_singleton_alarm_svc.py
View File

@ -61,7 +61,7 @@ class TestSingletonAlarmService(test.BaseTestCase):
mock.call(604800, mock.ANY),
]
actual = self.singleton.tg.add_timer.call_args_list
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_evaluation_cycle(self):
alarm = mock.Mock(type='threshold')
@ -104,4 +104,4 @@ class TestSingletonAlarmService(test.BaseTestCase):
os_endpoint_type=conf.os_endpoint_type,
insecure=conf.insecure)]
actual = client.call_args_list
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)

10
ceilometer/tests/api/v1/test_compute_duration_by_resource_scenarios.py
View File

@ -83,35 +83,35 @@ class TestComputeDurationByResource(tests_api.TestBase,
def _assert_times_match(self, actual, expected):
actual = timeutils.parse_isotime(actual).replace(tzinfo=None)
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_overlap_range_start(self):
with self._patch_get_stats(self.early1, self.middle1):
data = self._invoke_api()
self._assert_times_match(data['start_timestamp'], self.start)
self._assert_times_match(data['end_timestamp'], self.middle1)
self.assertEqual(data['duration'], 8 * 60 * 60)
self.assertEqual(8 * 60 * 60, data['duration'])
def test_within_range(self):
with self._patch_get_stats(self.middle1, self.middle2):
data = self._invoke_api()
self._assert_times_match(data['start_timestamp'], self.middle1)
self._assert_times_match(data['end_timestamp'], self.middle2)
self.assertEqual(data['duration'], 10 * 60 * 60)
self.assertEqual(10 * 60 * 60, data['duration'])
def test_within_range_zero_duration(self):
with self._patch_get_stats(self.middle1, self.middle1):
data = self._invoke_api()
self._assert_times_match(data['start_timestamp'], self.middle1)
self._assert_times_match(data['end_timestamp'], self.middle1)
self.assertEqual(data['duration'], 0)
self.assertEqual(0, data['duration'])
def test_overlap_range_end(self):
with self._patch_get_stats(self.middle2, self.late1):
data = self._invoke_api()
self._assert_times_match(data['start_timestamp'], self.middle2)
self._assert_times_match(data['end_timestamp'], self.end)
self.assertEqual(data['duration'], ((6 * 60) - 1) * 60)
self.assertEqual(((6 * 60) - 1) * 60, data['duration'])
def test_after_range(self):
with self._patch_get_stats(self.late1, self.late2):

8
ceilometer/tests/api/v1/test_get_query_ts.py
View File

@ -34,7 +34,7 @@ class TestQueryTimestamps(test.BaseTestCase):
'query_end': None,
'search_offset': 0,
}
self.assertEqual(result, expected)
self.assertEqual(expected, result)
def test_get_query_timestamps_start(self):
args = {'start_timestamp': '2012-09-20T12:13:14'}
@ -47,7 +47,7 @@ class TestQueryTimestamps(test.BaseTestCase):
'query_end': None,
'search_offset': 0,
}
self.assertEqual(result, expected)
self.assertEqual(expected, result)
def test_get_query_timestamps_end(self):
args = {'end_timestamp': '2012-09-20T12:13:14'}
@ -60,7 +60,7 @@ class TestQueryTimestamps(test.BaseTestCase):
'query_start': None,
'search_offset': 0,
}
self.assertEqual(result, expected)
self.assertEqual(expected, result)
def test_get_query_timestamps_with_offset(self):
args = {'start_timestamp': '2012-09-20T12:13:14',
@ -78,4 +78,4 @@ class TestQueryTimestamps(test.BaseTestCase):
12, 13, 14),
'search_offset': 20,
}
self.assertEqual(result, expected)
self.assertEqual(expected, result)

28
ceilometer/tests/api/v1/test_list_meters_scenarios.py
View File

@ -116,23 +116,23 @@ class TestListMeters(tests_api.TestBase,
def test_list_meters(self):
data = self.get('/meters')
self.assertEqual(4, len(data['meters']))
self.assertEqual(set(r['resource_id'] for r in data['meters']),
set(['resource-id',
self.assertEqual(set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']))
self.assertEqual(set(r['name'] for r in data['meters']),
set(['meter.test', 'meter.mine']))
'resource-id4']),
set(r['resource_id'] for r in data['meters']))
self.assertEqual(set(['meter.test', 'meter.mine']),
set(r['name'] for r in data['meters']))
def test_list_meters_non_admin(self):
data = self.get('/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEqual(2, len(data['meters']))
self.assertEqual(set(r['resource_id'] for r in data['meters']),
set(['resource-id', 'resource-id2']))
self.assertEqual(set(r['name'] for r in data['meters']),
set(['meter.test', 'meter.mine']))
self.assertEqual(set(['resource-id', 'resource-id2']),
set(r['resource_id'] for r in data['meters']))
self.assertEqual(set(['meter.test', 'meter.mine']),
set(r['name'] for r in data['meters']))
def test_with_resource(self):
data = self.get('/resources/resource-id/meters')
@ -156,7 +156,7 @@ class TestListMeters(tests_api.TestBase,
def test_with_source_non_existent(self):
data = self.get('/sources/test_list_resources_dont_exist/meters')
self.assertEqual(data['meters'], [])
self.assertEqual([], data['meters'])
def test_with_user(self):
data = self.get('/users/user-id/meters')
@ -182,11 +182,11 @@ class TestListMeters(tests_api.TestBase,
headers={"X-Roles": "Member",
"X-Project-Id": "project666"})
self.assertEqual(data['meters'], [])
self.assertEqual([], data['meters'])
def test_with_user_non_existent(self):
data = self.get('/users/user-id-foobar123/meters')
self.assertEqual(data['meters'], [])
self.assertEqual([], data['meters'])
def test_with_project(self):
data = self.get('/projects/project-id2/meters')
@ -204,11 +204,11 @@ class TestListMeters(tests_api.TestBase,
data = self.get('/projects/project-id2/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEqual(data.status_code, 404)
self.assertEqual(404, data.status_code)
def test_with_project_non_existent(self):
data = self.get('/projects/jd-was-here/meters')
self.assertEqual(data['meters'], [])
self.assertEqual([], data['meters'])
class TestListMetersMetaquery(TestListMeters,

24
ceilometer/tests/api/v1/test_list_resources_scenarios.py
View File

@ -112,18 +112,18 @@ class TestListResources(TestListResourcesBase):
def test_list_resources(self):
data = self.get('/resources')
self.assertEqual(3, len(data['resources']))
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id',
self.assertEqual(set(['resource-id',
'resource-id-alternate',
'resource-id2']))
'resource-id2']),
set(r['resource_id'] for r in data['resources']))
def test_list_resources_non_admin(self):
data = self.get('/resources',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEqual(2, len(data['resources']))
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id', 'resource-id-alternate']))
self.assertEqual(set(['resource-id', 'resource-id-alternate']),
set(r['resource_id'] for r in data['resources']))
def test_list_resources_with_timestamps(self):
data = self.get('/resources',
@ -131,8 +131,8 @@ class TestListResources(TestListResourcesBase):
2012, 7, 2, 10, 41).isoformat(),
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 43).isoformat())
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id-alternate', 'resource-id2']))
self.assertEqual(set(['resource-id-alternate', 'resource-id2']),
set(r['resource_id'] for r in data['resources']))
def test_list_resources_with_timestamps_non_admin(self):
data = self.get('/resources',
@ -142,8 +142,8 @@ class TestListResources(TestListResourcesBase):
2012, 7, 2, 10, 43).isoformat(),
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id-alternate']))
self.assertEqual(set(['resource-id-alternate']),
set(r['resource_id'] for r in data['resources']))
def test_with_source(self):
data = self.get('/sources/test_list_resources/resources')
@ -181,7 +181,7 @@ class TestListResources(TestListResourcesBase):
def test_with_source_non_existent(self):
data = self.get('/sources/test_list_resources_dont_exist/resources')
self.assertEqual(data['resources'], [])
self.assertEqual([], data['resources'])
def test_with_user(self):
data = self.get('/users/user-id/resources')
@ -224,7 +224,7 @@ class TestListResources(TestListResourcesBase):
def test_with_user_non_existent(self):
data = self.get('/users/user-id-foobar123/resources')
self.assertEqual(data['resources'], [])
self.assertEqual([], data['resources'])
def test_with_project(self):
data = self.get('/projects/project-id/resources')
@ -260,7 +260,7 @@ class TestListResources(TestListResourcesBase):
def test_with_project_non_existent(self):
data = self.get('/projects/jd-was-here/resources')
self.assertEqual(data['resources'], [])
self.assertEqual([], data['resources'])
class TestListResourcesMetaquery(TestListResourcesBase,

4
ceilometer/tests/api/v1/test_list_sources_scenarios.py
View File

@ -31,8 +31,8 @@ class TestListSource(tests_api.TestBase,
def test_source(self):
ydata = self.get('/sources/test_source')
self.assertIn("somekey", ydata)
self.assertEqual(ydata["somekey"], 666)
self.assertEqual(666, ydata["somekey"])
def test_unknownsource(self):
ydata = self.get('/sources/test_source_that_does_not_exist')
self.assertEqual(ydata, {})
self.assertEqual({}, ydata)

2
ceilometer/tests/api/v1/test_list_users_scenarios.py
View File

@ -101,7 +101,7 @@ class TestListUsers(tests_api.TestBase,
def test_with_source(self):
data = self.get('/sources/test_list_users/users')
self.assertEqual(['user-id'], data['users'])
self.assertEqual(data['users'], ['user-id'])
def test_with_source_non_admin(self):
data = self.get('/sources/test_list_users/users',

16
ceilometer/tests/api/v1/test_max_project_volume_scenarios.py
View File

@ -60,47 +60,47 @@ class TestMaxProjectVolume(tests_api.TestBase,
def test_no_time_bounds(self):
data = self.get('/projects/project1/meters/volume.size/volume/max')
expected = {'volume': 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_no_time_bounds_non_admin(self):
data = self.get('/projects/project1/meters/volume.size/volume/max',
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEqual(data, {'volume': 7})
self.assertEqual({'volume': 7}, data)
def test_no_time_bounds_wrong_tenant(self):
resp = self.get('/projects/project1/meters/volume.size/volume/max',
headers={"X-Roles": "Member",
"X-Project-Id": "?"})
self.assertEqual(resp.status_code, 404)
self.assertEqual(404, resp.status_code)
def test_start_timestamp(self):
data = self.get('/projects/project1/meters/volume.size/volume/max',
start_timestamp='2012-09-25T11:30:00')
expected = {'volume': 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_timestamp_after(self):
data = self.get('/projects/project1/meters/volume.size/volume/max',
start_timestamp='2012-09-25T12:34:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp(self):
data = self.get('/projects/project1/meters/volume.size/volume/max',
end_timestamp='2012-09-25T11:30:00')
expected = {'volume': 5}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp_before(self):
data = self.get('/projects/project1/meters/volume.size/volume/max',
end_timestamp='2012-09-25T09:54:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_end_timestamp(self):
data = self.get('/projects/project1/meters/volume.size/volume/max',
start_timestamp='2012-09-25T11:30:00',
end_timestamp='2012-09-25T11:32:00')
expected = {'volume': 6}
self.assertEqual(data, expected)
self.assertEqual(expected, data)

16
ceilometer/tests/api/v1/test_max_resource_volume_scenarios.py
View File

@ -59,47 +59,47 @@ class TestMaxResourceVolume(tests_api.TestBase,
def test_no_time_bounds(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max')
expected = {'volume': 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_no_time_bounds_non_admin(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEqual(data, {'volume': 7})
self.assertEqual({'volume': 7}, data)
def test_no_time_bounds_wrong_tenant(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
headers={"X-Roles": "Member",
"X-Project-Id": "??"})
self.assertEqual(data, {'volume': None})
self.assertEqual({'volume': None}, data)
def test_start_timestamp(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
start_timestamp='2012-09-25T11:30:00')
expected = {'volume': 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_timestamp_after(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
start_timestamp='2012-09-25T12:34:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
end_timestamp='2012-09-25T11:30:00')
expected = {'volume': 5}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp_before(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
end_timestamp='2012-09-25T09:54:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_end_timestamp(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/max',
start_timestamp='2012-09-25T11:30:00',
end_timestamp='2012-09-25T11:32:00')
expected = {'volume': 6}
self.assertEqual(data, expected)
self.assertEqual(expected, data)

16
ceilometer/tests/api/v1/test_sum_project_volume_scenarios.py
View File

@ -60,47 +60,47 @@ class TestSumProjectVolume(tests_api.TestBase,
def test_no_time_bounds(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum')
expected = {'volume': 5 + 6 + 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_no_time_bounds_non_admin(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum',
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEqual(data, {'volume': 5 + 6 + 7})
self.assertEqual({'volume': 5 + 6 + 7}, data)
def test_no_time_bounds_wrong_tenant(self):
resp = self.get('/projects/project1/meters/volume.size/volume/sum',
headers={"X-Roles": "Member",
"X-Project-Id": "???"})
self.assertEqual(resp.status_code, 404)
self.assertEqual(404, resp.status_code)
def test_start_timestamp(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum',
start_timestamp='2012-09-25T11:30:00')
expected = {'volume': 6 + 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_timestamp_after(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum',
start_timestamp='2012-09-25T12:34:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum',
end_timestamp='2012-09-25T11:30:00')
expected = {'volume': 5}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp_before(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum',
end_timestamp='2012-09-25T09:54:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_end_timestamp(self):
data = self.get('/projects/project1/meters/volume.size/volume/sum',
start_timestamp='2012-09-25T11:30:00',
end_timestamp='2012-09-25T11:32:00')
expected = {'volume': 6}
self.assertEqual(data, expected)
self.assertEqual(expected, data)

16
ceilometer/tests/api/v1/test_sum_resource_volume_scenarios.py
View File

@ -60,47 +60,47 @@ class TestSumResourceVolume(tests_api.TestBase,
def test_no_time_bounds(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum')
expected = {'volume': 5 + 6 + 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_no_time_bounds_non_admin(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEqual(data, {'volume': 5 + 6 + 7})
self.assertEqual({'volume': 5 + 6 + 7}, data)
def test_no_time_bounds_wrong_tenant(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
headers={"X-Roles": "Member",
"X-Project-Id": "?"})
self.assertEqual(data, {'volume': None})
self.assertEqual({'volume': None}, data)
def test_start_timestamp(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
start_timestamp='2012-09-25T11:30:00')
expected = {'volume': 6 + 7}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_timestamp_after(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
start_timestamp='2012-09-25T12:34:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
end_timestamp='2012-09-25T11:30:00')
expected = {'volume': 5}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_end_timestamp_before(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
end_timestamp='2012-09-25T09:54:00')
expected = {'volume': None}
self.assertEqual(data, expected)
self.assertEqual(expected, data)
def test_start_end_timestamp(self):
data = self.get('/resources/resource-id/meters/volume.size/volume/sum',
start_timestamp='2012-09-25T11:30:00',
end_timestamp='2012-09-25T11:32:00')
expected = {'volume': 6}
self.assertEqual(data, expected)
self.assertEqual(expected, data)

10
ceilometer/tests/api/v2/test_acl_scenarios.py
View File

@ -124,7 +124,7 @@ class TestAPIACL(FunctionalTest,
def test_non_authenticated(self):
response = self.get_json('/meters', expect_errors=True)
self.assertEqual(response.status_int, 401)
self.assertEqual(401, response.status_int)
def test_authenticated_wrong_role(self):
response = self.get_json('/meters',
@ -135,7 +135,7 @@ class TestAPIACL(FunctionalTest,
"X-Project-Id":
"bc23a9d531064583ace8f67dad60f6bb",
})
self.assertEqual(response.status_int, 401)
self.assertEqual(401, response.status_int)
# FIXME(dhellmann): This test is not properly looking at the tenant
# info. We do not correctly detect the improper tenant. That's
@ -150,7 +150,7 @@ class TestAPIACL(FunctionalTest,
# "X-Tenant-Name": "achoo",
# "X-Project-Id": "bc23a9d531064583ace8f67dad60f6bb",
# })
# self.assertEqual(response.status_int, 401)
# self.assertEqual(401, response.status_int)
def test_authenticated(self):
data = self.get_json('/meters',
@ -191,7 +191,7 @@ class TestAPIACL(FunctionalTest,
q=[{'field': 'project_id',
'value': 'project-wrong',
}])
self.assertEqual(data.status_int, 401)
self.assertEqual(401, data.status_int)
def test_non_admin_two_projects(self):
data = self.get_json('/meters',
@ -205,7 +205,7 @@ class TestAPIACL(FunctionalTest,
{'field': 'project_id',
'value': 'project-naughty',
}])
self.assertEqual(data.status_int, 401)
self.assertEqual(401, data.status_int)
def test_non_admin_get_events(self):

154
ceilometer/tests/api/v2/test_alarm_scenarios.py
View File

@ -170,20 +170,19 @@ class TestAlarms(FunctionalTest,
storage_key = 'rule'
else:
storage_key = key
self.assertEqual(getattr(alarm, storage_key),
json[key])
self.assertEqual(json[key], getattr(alarm, storage_key))
def test_list_alarms(self):
data = self.get_json('/alarms')
self.assertEqual(4, len(data))
self.assertEqual(set(r['name'] for r in data),
set(['name1', 'name2', 'name3', 'name4']))
self.assertEqual(set(r['threshold_rule']['meter_name']
for r in data if 'threshold_rule' in r),
set(['meter.test', 'meter.mine']))
self.assertEqual(set(r['combination_rule']['operator']
for r in data if 'combination_rule' in r),
set(['or']))
self.assertEqual(set(['name1', 'name2', 'name3', 'name4']),
set(r['name'] for r in data))
self.assertEqual(set(['meter.test', 'meter.mine']),
set(r['threshold_rule']['meter_name']
for r in data if 'threshold_rule' in r))
self.assertEqual(set(['or']),
set(r['combination_rule']['operator']
for r in data if 'combination_rule' in r))
def test_alarms_query_with_timestamp(self):
date_time = datetime.datetime(2012, 7, 2, 10, 41)
@ -201,28 +200,27 @@ class TestAlarms(FunctionalTest,
def test_get_not_existing_alarm(self):
resp = self.get_json('/alarms/alarm-id-3', expect_errors=True)
self.assertEqual(resp.status_code, 404)
self.assertEqual(jsonutils.loads(resp.body)['error_message']
['faultstring'],
"Alarm alarm-id-3 Not Found")
self.assertEqual(404, resp.status_code)
self.assertEqual("Alarm alarm-id-3 Not Found",
jsonutils.loads(resp.body)['error_message']
['faultstring'])
def test_get_alarm(self):
alarms = self.get_json('/alarms',
q=[{'field': 'name',
'value': 'name1',
}])
self.assertEqual(alarms[0]['name'], 'name1')
self.assertEqual(alarms[0]['threshold_rule']['meter_name'],
'meter.test')
self.assertEqual('name1', alarms[0]['name'])
self.assertEqual('meter.test',
alarms[0]['threshold_rule']['meter_name'])
one = self.get_json('/alarms/%s' % alarms[0]['alarm_id'])
self.assertEqual(one['name'], 'name1')
self.assertEqual(one['threshold_rule']['meter_name'],
'meter.test')
self.assertEqual(one['alarm_id'], alarms[0]['alarm_id'])
self.assertEqual(one['repeat_actions'], alarms[0]['repeat_actions'])
self.assertEqual(one['time_constraints'],
alarms[0]['time_constraints'])
self.assertEqual('name1', one['name'])
self.assertEqual('meter.test', one['threshold_rule']['meter_name'])
self.assertEqual(alarms[0]['alarm_id'], one['alarm_id'])
self.assertEqual(alarms[0]['repeat_actions'], one['repeat_actions'])
self.assertEqual(alarms[0]['time_constraints'],
one['time_constraints'])
def test_get_alarm_disabled(self):
alarm = models.Alarm(name='disabled',
@ -246,29 +244,29 @@ class TestAlarms(FunctionalTest,
alarms = self.get_json('/alarms',
q=[{'field': 'enabled',
'value': 'False'}])
self.assertEqual(len(alarms), 1)
self.assertEqual(alarms[0]['name'], 'disabled')
self.assertEqual(1, len(alarms))
self.assertEqual('disabled', alarms[0]['name'])
one = self.get_json('/alarms/%s' % alarms[0]['alarm_id'])
self.assertEqual(one['name'], 'disabled')
self.assertEqual('disabled', one['name'])
def test_get_alarm_combination(self):
alarms = self.get_json('/alarms',
q=[{'field': 'name',
'value': 'name4',
}])
self.assertEqual(alarms[0]['name'], 'name4')
self.assertEqual(alarms[0]['combination_rule']['alarm_ids'],
['a', 'b'])
self.assertEqual(alarms[0]['combination_rule']['operator'], 'or')
self.assertEqual('name4', alarms[0]['name'])
self.assertEqual(['a', 'b'],
alarms[0]['combination_rule']['alarm_ids'])
self.assertEqual('or', alarms[0]['combination_rule']['operator'])
one = self.get_json('/alarms/%s' % alarms[0]['alarm_id'])
self.assertEqual(one['name'], 'name4')
self.assertEqual(alarms[0]['combination_rule']['alarm_ids'],
['a', 'b'])
self.assertEqual(alarms[0]['combination_rule']['operator'], 'or')
self.assertEqual(one['alarm_id'], alarms[0]['alarm_id'])
self.assertEqual(one['repeat_actions'], alarms[0]['repeat_actions'])
self.assertEqual('name4', one['name'])
self.assertEqual(['a', 'b'],
alarms[0]['combination_rule']['alarm_ids'])
self.assertEqual('or', alarms[0]['combination_rule']['operator'])
self.assertEqual(alarms[0]['alarm_id'], one['alarm_id'])
self.assertEqual(alarms[0]['repeat_actions'], one['repeat_actions'])
def test_post_alarm_wsme_workaround(self):
jsons = {
@ -309,11 +307,10 @@ class TestAlarms(FunctionalTest,
for field, json in jsons.iteritems():
resp = self.post_json('/alarms', params=json, expect_errors=True,
status=400, headers=self.auth_headers)
self.assertEqual(
resp.json['error_message']['faultstring'],
"Invalid input for field/attribute %s."
" Value: \'None\'. Mandatory field missing."
% field.split('/', 1)[-1])
self.assertEqual("Invalid input for field/attribute %s."
" Value: \'None\'. Mandatory field missing."
% field.split('/', 1)[-1],
resp.json['error_message']['faultstring'])
alarms = list(self.conn.get_alarms())
self.assertEqual(4, len(alarms))
@ -520,10 +517,9 @@ class TestAlarms(FunctionalTest,
status=400, headers=self.auth_headers)
alarms = list(self.conn.get_alarms())
self.assertEqual(4, len(alarms))
self.assertEqual(
resp.json['error_message']['faultstring'],
'threshold_rule and combination_rule cannot '
'be set at the same time')
self.assertEqual('threshold_rule and combination_rule cannot '
'be set at the same time',
resp.json['error_message']['faultstring'])
def test_post_invalid_alarm_timestamp_in_threshold_rule(self):
date_time = datetime.datetime(2012, 7, 2, 10, 41)
@ -596,8 +592,8 @@ class TestAlarms(FunctionalTest,
storage_key = 'rule'
else:
storage_key = key
self.assertEqual(getattr(alarm, storage_key),
to_check[key])
self.assertEqual(to_check[key],
getattr(alarm, storage_key))
break
else:
self.fail("Alarm not found")
@ -752,27 +748,27 @@ class TestAlarms(FunctionalTest,
headers=headers)
alarms = list(self.conn.get_alarms(enabled=False))
self.assertEqual(1, len(alarms))
self.assertEqual(alarms[0].user_id, 'auseridthatisnotmine')
self.assertEqual(alarms[0].project_id, 'aprojectidthatisnotmine')
self.assertEqual('auseridthatisnotmine', alarms[0].user_id)
self.assertEqual('aprojectidthatisnotmine', alarms[0].project_id)
self._add_default_threshold_rule(json)
if alarms[0].name == 'added_alarm':
for key in json:
if key.endswith('_rule'):
storage_key = 'rule'
if explicit_project_constraint:
self.assertEqual(getattr(alarms[0], storage_key),
json[key])
self.assertEqual(json[key],
getattr(alarms[0], storage_key))
else:
query = getattr(alarms[0], storage_key).get('query')
self.assertEqual(len(query), 2)
self.assertEqual(2, len(query))
implicit_constraint = {
u'field': u'project_id',
u'value': u'aprojectidthatisnotmine',
u'op': u'eq'
}
self.assertEqual(query[1], implicit_constraint)
self.assertEqual(implicit_constraint, query[1])
else:
self.assertEqual(getattr(alarms[0], key), json[key])
self.assertEqual(json[key], getattr(alarms[0], key))
else:
self.fail("Alarm not found")
@ -820,8 +816,8 @@ class TestAlarms(FunctionalTest,
headers=headers)
alarms = list(self.conn.get_alarms(enabled=False))
self.assertEqual(1, len(alarms))
self.assertEqual(alarms[0].user_id, self.auth_headers['X-User-Id'])
self.assertEqual(alarms[0].project_id, 'aprojectidthatisnotmine')
self.assertEqual(self.auth_headers['X-User-Id'], alarms[0].user_id)
self.assertEqual('aprojectidthatisnotmine', alarms[0].project_id)
self._verify_alarm(json, alarms[0], 'added_alarm')
def test_post_alarm_as_admin_no_project(self):
@ -856,9 +852,9 @@ class TestAlarms(FunctionalTest,
headers=headers)
alarms = list(self.conn.get_alarms(enabled=False))
self.assertEqual(1, len(alarms))
self.assertEqual(alarms[0].user_id, 'auseridthatisnotmine')
self.assertEqual(alarms[0].project_id,
self.auth_headers['X-Project-Id'])
self.assertEqual('auseridthatisnotmine', alarms[0].user_id)
self.assertEqual(self.auth_headers['X-Project-Id'],
alarms[0].project_id)
self._verify_alarm(json, alarms[0], 'added_alarm')
def test_post_alarm_combination(self):
@ -887,8 +883,7 @@ class TestAlarms(FunctionalTest,
storage_key = 'rule'
else:
storage_key = key
self.assertEqual(getattr(alarms[0], storage_key),
json[key])
self.assertEqual(json[key], getattr(alarms[0], storage_key))
else:
self.fail("Alarm not found")
@ -1074,8 +1069,7 @@ class TestAlarms(FunctionalTest,
storage_key = 'rule'
else:
storage_key = key
self.assertEqual(getattr(alarms[0], storage_key),
json[key])
self.assertEqual(json[key], getattr(alarms[0], storage_key))
else:
self.fail("Alarm not found")
@ -1184,8 +1178,8 @@ class TestAlarms(FunctionalTest,
params=json,
headers=headers)
alarm = list(self.conn.get_alarms(alarm_id=alarm_id, enabled=False))[0]
self.assertEqual(alarm.user_id, 'myuserid')
self.assertEqual(alarm.project_id, 'myprojectid')
self.assertEqual('myuserid', alarm.user_id)
self.assertEqual('myprojectid', alarm.project_id)
self._verify_alarm(json, alarm)
def test_put_alarm_wrong_field(self):
@ -1224,7 +1218,7 @@ class TestAlarms(FunctionalTest,
resp = self.put_json('/alarms/%s' % alarm_id,
params=json,
headers=self.auth_headers)
self.assertEqual(resp.status_code, 200)
self.assertEqual(200, resp.status_code)
def test_put_alarm_with_existing_name(self):
"""Test that update a threshold alarm with an existing name.
@ -1273,7 +1267,7 @@ class TestAlarms(FunctionalTest,
resp = self.delete('/alarms/%s' % data[0]['alarm_id'],
headers=self.auth_headers,
status=204)
self.assertEqual(resp.body, '')
self.assertEqual('', resp.body)
alarms = list(self.conn.get_alarms())
self.assertEqual(3, len(alarms))
@ -1283,7 +1277,7 @@ class TestAlarms(FunctionalTest,
resp = self.get_json('/alarms/%s/state' % data[0]['alarm_id'],
headers=self.auth_headers)
self.assertEqual(data[0]['state'], resp)
self.assertEqual(resp, data[0]['state'])
def test_set_state_alarm(self):
data = self.get_json('/alarms')
@ -1294,8 +1288,8 @@ class TestAlarms(FunctionalTest,
params='alarm')
alarms = list(self.conn.get_alarms(alarm_id=data[0]['alarm_id']))
self.assertEqual(1, len(alarms))
self.assertEqual(alarms[0].state, 'alarm')
self.assertEqual(resp.json, 'alarm')
self.assertEqual('alarm', alarms[0].state)
self.assertEqual('alarm', resp.json)
def test_set_invalid_state_alarm(self):
data = self.get_json('/alarms')
@ -1321,7 +1315,7 @@ class TestAlarms(FunctionalTest,
headers=auth_headers or self.auth_headers,
expect_errors=expect_errors)
if expect_errors:
self.assertEqual(resp.status_code, status)
self.assertEqual(status, resp.status_code)
return resp
def _update_alarm(self, alarm, updated_data, auth_headers=None):
@ -1620,13 +1614,13 @@ class TestAlarms(FunctionalTest,
self.post_json('/alarms', params=json, headers=self.auth_headers)
calls = notifier.call_args_list
self.assertEqual(len(calls), 1)
self.assertEqual(1, len(calls))
args, _ = calls[0]
context, publisher, event_type, priority, payload = args
self.assertTrue(publisher.startswith('ceilometer.api'))
self.assertEqual(event_type, 'alarm.creation')
self.assertEqual(priority, 'INFO')
self.assertEqual(payload['detail']['name'], 'sent_notification')
self.assertEqual('alarm.creation', event_type)
self.assertEqual('INFO', priority)
self.assertEqual('sent_notification', payload['detail']['name'])
self.assertTrue(set(['alarm_id', 'detail', 'event_id', 'on_behalf_of',
'project_id', 'timestamp', 'type',
'user_id']).issubset(payload.keys()))
@ -1640,13 +1634,13 @@ class TestAlarms(FunctionalTest,
headers=self.auth_headers, status=204)
calls = notifier.call_args_list
self.assertEqual(len(calls), 1)
self.assertEqual(1, len(calls))
args, _ = calls[0]
context, publisher, event_type, priority, payload = args
self.assertTrue(publisher.startswith('ceilometer.api'))
self.assertEqual(event_type, 'alarm.deletion')
self.assertEqual(priority, 'INFO')
self.assertEqual(payload['detail']['name'], 'name1')
self.assertEqual('alarm.deletion', event_type)
self.assertEqual('INFO', priority)
self.assertEqual('name1', payload['detail']['name'])
self.assertTrue(set(['alarm_id', 'detail', 'event_id', 'on_behalf_of',
'project_id', 'timestamp', 'type',
'user_id']).issubset(payload.keys()))

72
ceilometer/tests/api/v2/test_app.py
View File

@ -80,7 +80,7 @@ class TestPecanApp(FunctionalTest):
def test_pecan_extension_guessing_unset(self):
# check Pecan does not assume .jpg is an extension
response = self.app.get(self.PATH_PREFIX + '/meters/meter.jpg')
self.assertEqual(response.content_type, 'application/json')
self.assertEqual('application/json', response.content_type)
class TestApiMiddleware(FunctionalTest):
@ -103,16 +103,16 @@ class TestApiMiddleware(FunctionalTest):
headers={"Accept":
"application/json"}
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(404, response.status_int)
self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message'])
response = self.get_json('/invalid_path',
expect_errors=True,
headers={"Accept":
"application/json,application/xml"}
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(404, response.status_int)
self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message'])
response = self.get_json('/invalid_path',
expect_errors=True,
@ -120,22 +120,22 @@ class TestApiMiddleware(FunctionalTest):
"application/xml;q=0.8, \
application/json"}
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(404, response.status_int)
self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message'])
response = self.get_json('/invalid_path',
expect_errors=True
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(404, response.status_int)
self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message'])
response = self.get_json('/invalid_path',
expect_errors=True,
headers={"Accept":
"text/html,*/*"}
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(404, response.status_int)
self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message'])
def test_json_parsable_error_middleware_translation_400(self):
@ -148,11 +148,11 @@ class TestApiMiddleware(FunctionalTest):
headers={"Accept":
"application/json"}
)
self.assertEqual(response.status_int, 400)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(400, response.status_int)
self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message'])
self.assertEqual(response.json['error_message']['faultstring'],
self.no_lang_translated_error)
self.assertEqual(self.no_lang_translated_error,
response.json['error_message']['faultstring'])
def test_xml_parsable_error_middleware_404(self):
response = self.get_json('/invalid_path',
@ -160,18 +160,18 @@ class TestApiMiddleware(FunctionalTest):
headers={"Accept":
"application/xml,*/*"}
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/xml")
self.assertEqual(response.xml.tag, 'error_message')
self.assertEqual(404, response.status_int)
self.assertEqual("application/xml", response.content_type)
self.assertEqual('error_message', response.xml.tag)
response = self.get_json('/invalid_path',
expect_errors=True,
headers={"Accept":
"application/json;q=0.8 \
,application/xml"}
)
self.assertEqual(response.status_int, 404)
self.assertEqual(response.content_type, "application/xml")
self.assertEqual(response.xml.tag, 'error_message')
self.assertEqual(404, response.status_int)
self.assertEqual("application/xml", response.content_type)
self.assertEqual('error_message', response.xml.tag)
def test_xml_parsable_error_middleware_translation_400(self):
# Ensure translated messages get placed properly into xml faults
@ -183,12 +183,12 @@ class TestApiMiddleware(FunctionalTest):
headers={"Accept":
"application/xml,*/*"}
)
self.assertEqual(response.status_int, 400)
self.assertEqual(response.content_type, "application/xml")
self.assertEqual(response.xml.tag, 'error_message')
self.assertEqual(400, response.status_int)
self.assertEqual("application/xml", response.content_type)
self.assertEqual('error_message', response.xml.tag)
fault = response.xml.findall('./error/faultstring')
for fault_string in fault:
self.assertEqual(fault_string.text, self.no_lang_translated_error)
self.assertEqual(self.no_lang_translated_error, fault_string.text)
def test_best_match_language(self):
# Ensure that we are actually invoking language negotiation
@ -203,18 +203,19 @@ class TestApiMiddleware(FunctionalTest):
"en-US"}
)
self.assertEqual(response.status_int, 400)
self.assertEqual(response.content_type, "application/xml")
self.assertEqual(response.xml.tag, 'error_message')
self.assertEqual(400, response.status_int)
self.assertEqual("application/xml", response.content_type)
self.assertEqual('error_message', response.xml.tag)
fault = response.xml.findall('./error/faultstring')
for fault_string in fault:
self.assertEqual(fault_string.text, self.en_US_translated_error)
self.assertEqual(self.en_US_translated_error, fault_string.text)
def test_translated_then_untranslated_error(self):
resp = self.get_json('/alarms/alarm-id-3', expect_errors=True)
self.assertEqual(resp.status_code, 404)
self.assertEqual(json.loads(resp.body)['error_message']
['faultstring'], "Alarm alarm-id-3 Not Found")
self.assertEqual(404, resp.status_code)
self.assertEqual("Alarm alarm-id-3 Not Found",
json.loads(resp.body)['error_message']
['faultstring'])
with mock.patch('ceilometer.api.controllers.v2.EntityNotFound') \
as CustomErrorClass:
@ -222,6 +223,7 @@ class TestApiMiddleware(FunctionalTest):
"untranslated_error", status_code=404)
resp = self.get_json('/alarms/alarm-id-5', expect_errors=True)
self.assertEqual(resp.status_code, 404)
self.assertEqual(json.loads(resp.body)['error_message']
['faultstring'], "untranslated_error")
self.assertEqual(404, resp.status_code)
self.assertEqual("untranslated_error",
json.loads(resp.body)['error_message']
['faultstring'])

14
ceilometer/tests/api/v2/test_compute_duration_by_resource_scenarios.py
View File

@ -99,46 +99,46 @@ class TestComputeDurationByResource(FunctionalTest,
def test_before_range(self):
with self._patch_get_interval(self.early1, self.early2):
data = self._invoke_api()
self.assertEqual(data, [])
self.assertEqual([], data)
def _assert_times_match(self, actual, expected):
if actual:
actual = timeutils.parse_isotime(actual)
actual = actual.replace(tzinfo=None)
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_overlap_range_start(self):
with self._patch_get_interval(self.early1, self.middle1):
data = self._invoke_api()
self._assert_times_match(data[0]['duration_start'], self.start)
self._assert_times_match(data[0]['duration_end'], self.middle1)
self.assertEqual(data[0]['duration'], 8 * 60 * 60)
self.assertEqual(8 * 60 * 60, data[0]['duration'])
def test_within_range(self):
with self._patch_get_interval(self.middle1, self.middle2):
data = self._invoke_api()
self._assert_times_match(data[0]['duration_start'], self.middle1)
self._assert_times_match(data[0]['duration_end'], self.middle2)
self.assertEqual(data[0]['duration'], 10 * 60 * 60)
self.assertEqual(10 * 60 * 60, data[0]['duration'])
def test_within_range_zero_duration(self):
with self._patch_get_interval(self.middle1, self.middle1):
data = self._invoke_api()
self._assert_times_match(data[0]['duration_start'], self.middle1)
self._assert_times_match(data[0]['duration_end'], self.middle1)
self.assertEqual(data[0]['duration'], 0)
self.assertEqual(0, data[0]['duration'])
def test_overlap_range_end(self):
with self._patch_get_interval(self.middle2, self.late1):
data = self._invoke_api()
self._assert_times_match(data[0]['duration_start'], self.middle2)
self._assert_times_match(data[0]['duration_end'], self.end)
self.assertEqual(data[0]['duration'], ((6 * 60) - 1) * 60)
self.assertEqual(((6 * 60) - 1) * 60, data[0]['duration'])
def test_after_range(self):
with self._patch_get_interval(self.late1, self.late2):
data = self._invoke_api()
self.assertEqual(data, [])
self.assertEqual([], data)
def test_without_end_timestamp(self):
statistics = [