openstack/deb-python-ceilometerclient
Code Issues Proposed changes

use mock instead of try...finally

Browse Source 
There are some test code use try...finally block to catch or ignore
the stdout message. The job can be done by mock too, and it is cleaner.

Change-Id: I0c29382e7bcd0b7f399565b26f27315f7e17a8fc
This commit is contained in:
ZhiQiang Fan
2014-05-30 17:20:47 +08:00
parent d55038ceaf
commit c343b184b8
3 changed files with 82 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
ceilometerclient/tests/test_shell.py
View File

@@ -42,24 +42,19 @@ class ShellTest(utils.BaseTestCase):
def setUp(self): def setUp(self):
super(ShellTest, self).setUp() super(ShellTest, self).setUp()
@mock.patch('sys.stdout', new=six.StringIO())
@mock.patch.object(ksclient, 'Client') @mock.patch.object(ksclient, 'Client')
@mock.patch.object(v1client.http.HTTPClient, 'json_request') @mock.patch.object(v1client.http.HTTPClient, 'json_request')
@mock.patch.object(v1client.http.HTTPClient, 'raw_request') @mock.patch.object(v1client.http.HTTPClient, 'raw_request')
def shell(self, argstr, mock_ksclient, mock_json, mock_raw): def shell(self, argstr, mock_ksclient, mock_json, mock_raw):
orig = sys.stdout
try: try:
sys.stdout = six.StringIO()
_shell = ceilometer_shell.CeilometerShell() _shell = ceilometer_shell.CeilometerShell()
_shell.main(argstr.split()) _shell.main(argstr.split())
except SystemExit: except SystemExit:
exc_type, exc_value, exc_traceback = sys.exc_info() exc_type, exc_value, exc_traceback = sys.exc_info()
self.assertEqual(exc_value.code, 0) self.assertEqual(exc_value.code, 0)
finally:
out = sys.stdout.getvalue()
sys.stdout.close()
sys.stdout = orig
return out return sys.stdout.getvalue()
def test_help_unknown_command(self): def test_help_unknown_command(self):
self.assertRaises(exc.CommandError, self.shell, 'help foofoo') self.assertRaises(exc.CommandError, self.shell, 'help foofoo')

21
ceilometerclient/tests/test_utils.py
View File

@@ -17,7 +17,6 @@
import itertools import itertools
import mock import mock
import six import six
import sys
from ceilometerclient.common import utils from ceilometerclient.common import utils
from ceilometerclient.tests import utils as test_utils from ceilometerclient.tests import utils as test_utils
@@ -31,22 +30,16 @@ class UtilsTest(test_utils.BaseTestCase):
self.__dict__.update(entries) self.__dict__.update(entries)
# test that the prettytable output is wellformatted (left-aligned) # test that the prettytable output is wellformatted (left-aligned)
saved_stdout = sys.stdout with mock.patch('sys.stdout', new=six.StringIO()) as stdout:
try:
sys.stdout = output_dict = six.StringIO()
utils.print_dict({'K': 'k', 'Key': 'Value'}) utils.print_dict({'K': 'k', 'Key': 'Value'})
self.assertEqual('''\
finally:
sys.stdout = saved_stdout
self.assertEqual(output_dict.getvalue(), '''\
+----------+-------+ +----------+-------+
| Property | Value | | Property | Value |
+----------+-------+ +----------+-------+
| K | k | | K | k |
| Key | Value | | Key | Value |
+----------+-------+ +----------+-------+
''') ''', stdout.getvalue())
def test_print_list(self): def test_print_list(self):
class Foo: class Foo:
@@ -61,17 +54,13 @@ class UtilsTest(test_utils.BaseTestCase):
Foo(12, '0', 'Z')] Foo(12, '0', 'Z')]
def do_print_list(sortby): def do_print_list(sortby):
saved_stdout = sys.stdout with mock.patch('sys.stdout', new=six.StringIO()) as stdout:
try:
sys.stdout = output = six.StringIO()
utils.print_list(foo_list, utils.print_list(foo_list,
['one', 'two', 'three'], ['one', 'two', 'three'],
['1st', '2nd', '3rd'], ['1st', '2nd', '3rd'],
{'one': lambda o: o.one * 10}, {'one': lambda o: o.one * 10},
sortby) sortby)
finally: return stdout.getvalue()
sys.stdout = saved_stdout
return output.getvalue()
printed = do_print_list(None) printed = do_print_list(None)
self.assertEqual(printed, '''\ self.assertEqual(printed, '''\

191
ceilometerclient/tests/v2/test_shell.py
View File

@@ -107,32 +107,27 @@ class ShellAlarmHistoryCommandTest(utils.BaseTestCase):
self.args = mock.Mock() self.args = mock.Mock()
self.args.alarm_id = self.ALARM_ID self.args.alarm_id = self.ALARM_ID
@mock.patch('sys.stdout', new=six.StringIO())
def _do_test_alarm_history(self, raw_query=None, parsed_query=None): def _do_test_alarm_history(self, raw_query=None, parsed_query=None):
self.args.query = raw_query self.args.query = raw_query
orig = sys.stdout
sys.stdout = six.StringIO()
history = [alarms.AlarmChange(mock.Mock(), change) history = [alarms.AlarmChange(mock.Mock(), change)
for change in self.ALARM_HISTORY] for change in self.ALARM_HISTORY]
self.cc.alarms.get_history.return_value = history self.cc.alarms.get_history.return_value = history
try: ceilometer_shell.do_alarm_history(self.cc, self.args)
ceilometer_shell.do_alarm_history(self.cc, self.args) self.cc.alarms.get_history.assert_called_once_with(
self.cc.alarms.get_history.assert_called_once_with( q=parsed_query,
q=parsed_query, alarm_id=self.ALARM_ID
alarm_id=self.ALARM_ID )
) out = sys.stdout.getvalue()
out = sys.stdout.getvalue() required = [
required = [ '.*creation%sname: scombo.*' % self.TIMESTAMP_RE,
'.*creation%sname: scombo.*' % self.TIMESTAMP_RE, '.*rule change%sdescription: combination of one.*' %
'.*rule change%sdescription: combination of one.*' % self.TIMESTAMP_RE,
self.TIMESTAMP_RE, '.*state transition%sstate: alarm.*' % self.TIMESTAMP_RE,
'.*state transition%sstate: alarm.*' % self.TIMESTAMP_RE, ]
] for r in required:
for r in required: self.assertThat(out, matchers.MatchesRegex(r, re.DOTALL))
self.assertThat(out, matchers.MatchesRegex(r, re.DOTALL))
finally:
sys.stdout.close()
sys.stdout = orig
def test_alarm_all_history(self): def test_alarm_all_history(self):
self._do_test_alarm_history() self._do_test_alarm_history()
@@ -190,28 +185,23 @@ class ShellAlarmCommandTest(utils.BaseTestCase):
self.args = mock.Mock() self.args = mock.Mock()
self.args.alarm_id = self.ALARM_ID self.args.alarm_id = self.ALARM_ID
@mock.patch('sys.stdout', new=six.StringIO())
def _do_test_alarm_update_repeat_actions(self, method, repeat_actions): def _do_test_alarm_update_repeat_actions(self, method, repeat_actions):
self.args.threshold = 42.0 self.args.threshold = 42.0
if repeat_actions is not None: if repeat_actions is not None:
self.args.repeat_actions = repeat_actions self.args.repeat_actions = repeat_actions
orig = sys.stdout
sys.stdout = six.StringIO()
alarm = [alarms.Alarm(mock.Mock(), self.ALARM)] alarm = [alarms.Alarm(mock.Mock(), self.ALARM)]
self.cc.alarms.get.return_value = alarm self.cc.alarms.get.return_value = alarm
self.cc.alarms.update.return_value = alarm[0] self.cc.alarms.update.return_value = alarm[0]
try: method(self.cc, self.args)
method(self.cc, self.args) args, kwargs = self.cc.alarms.update.call_args
args, kwargs = self.cc.alarms.update.call_args self.assertEqual(self.ALARM_ID, args[0])
self.assertEqual(self.ALARM_ID, args[0]) self.assertEqual(42.0, kwargs.get('threshold'))
self.assertEqual(42.0, kwargs.get('threshold')) if repeat_actions is not None:
if repeat_actions is not None: self.assertEqual(repeat_actions, kwargs.get('repeat_actions'))
self.assertEqual(repeat_actions, kwargs.get('repeat_actions')) else:
else: self.assertFalse('repeat_actions' in kwargs)
self.assertFalse('repeat_actions' in kwargs)
finally:
sys.stdout.close()
sys.stdout = orig
def test_alarm_update_repeat_actions_untouched(self): def test_alarm_update_repeat_actions_untouched(self):
method = ceilometer_shell.do_alarm_update method = ceilometer_shell.do_alarm_update
@@ -249,6 +239,7 @@ class ShellAlarmCommandTest(utils.BaseTestCase):
method = ceilometer_shell.do_alarm_threshold_update method = ceilometer_shell.do_alarm_threshold_update
self._do_test_alarm_update_repeat_actions(method, False) self._do_test_alarm_update_repeat_actions(method, False)
@mock.patch('sys.stdout', new=six.StringIO())
def test_alarm_threshold_create_args(self): def test_alarm_threshold_create_args(self):
shell = base_shell.CeilometerShell() shell = base_shell.CeilometerShell()
argv = ['alarm-threshold-create', argv = ['alarm-threshold-create',
@@ -265,33 +256,28 @@ class ShellAlarmCommandTest(utils.BaseTestCase):
'--query', 'resource_id=INSTANCE_ID'] '--query', 'resource_id=INSTANCE_ID']
_, args = shell.parse_args(argv) _, args = shell.parse_args(argv)
orig = sys.stdout
sys.stdout = six.StringIO()
alarm = alarms.Alarm(mock.Mock(), self.ALARM) alarm = alarms.Alarm(mock.Mock(), self.ALARM)
self.cc.alarms.create.return_value = alarm self.cc.alarms.create.return_value = alarm
try: ceilometer_shell.do_alarm_threshold_create(self.cc, args)
ceilometer_shell.do_alarm_threshold_create(self.cc, args) _, kwargs = self.cc.alarms.create.call_args
_, kwargs = self.cc.alarms.create.call_args self.assertEqual('cpu_high', kwargs.get('name'))
self.assertEqual('cpu_high', kwargs.get('name')) self.assertEqual('instance running hot', kwargs.get('description'))
self.assertEqual('instance running hot', kwargs.get('description')) actions = ['log://', 'http://example.com/alarm/state']
actions = ['log://', 'http://example.com/alarm/state'] self.assertEqual(actions, kwargs.get('alarm_actions'))
self.assertEqual(actions, kwargs.get('alarm_actions')) self.assertTrue('threshold_rule' in kwargs)
self.assertTrue('threshold_rule' in kwargs) rule = kwargs['threshold_rule']
rule = kwargs['threshold_rule'] self.assertEqual('cpu_util', rule.get('meter_name'))
self.assertEqual('cpu_util', rule.get('meter_name')) self.assertEqual(70.0, rule.get('threshold'))
self.assertEqual(70.0, rule.get('threshold')) self.assertEqual('gt', rule.get('comparison_operator'))
self.assertEqual('gt', rule.get('comparison_operator')) self.assertEqual('avg', rule.get('statistic'))
self.assertEqual('avg', rule.get('statistic')) self.assertEqual(600, rule.get('period'))
self.assertEqual(600, rule.get('period')) self.assertEqual(3, rule.get('evaluation_periods'))
self.assertEqual(3, rule.get('evaluation_periods')) query = dict(field='resource_id', type='',
query = dict(field='resource_id', type='', value='INSTANCE_ID', op='eq')
value='INSTANCE_ID', op='eq') self.assertEqual([query], rule['query'])
self.assertEqual([query], rule['query'])
finally:
sys.stdout.close()
sys.stdout = orig
@mock.patch('sys.stdout', new=six.StringIO())
def test_alarm_create_time_constraints(self): def test_alarm_create_time_constraints(self):
shell = base_shell.CeilometerShell() shell = base_shell.CeilometerShell()
argv = ['alarm-threshold-create', argv = ['alarm-threshold-create',
@@ -305,22 +291,16 @@ class ShellAlarmCommandTest(utils.BaseTestCase):
] ]
_, args = shell.parse_args(argv) _, args = shell.parse_args(argv)
orig = sys.stdout
sys.stdout = six.StringIO()
alarm = alarms.Alarm(mock.Mock(), self.ALARM) alarm = alarms.Alarm(mock.Mock(), self.ALARM)
self.cc.alarms.create.return_value = alarm self.cc.alarms.create.return_value = alarm
try: ceilometer_shell.do_alarm_threshold_create(self.cc, args)
ceilometer_shell.do_alarm_threshold_create(self.cc, args) _, kwargs = self.cc.alarms.create.call_args
_, kwargs = self.cc.alarms.create.call_args time_constraints = [dict(name='cons1', start='0 11 * * *',
time_constraints = [dict(name='cons1', start='0 11 * * *', duration='300'),
duration='300'), dict(name='cons2', start='0 23 * * *',
dict(name='cons2', start='0 23 * * *', duration='600')]
duration='600')] self.assertEqual(time_constraints, kwargs['time_constraints'])
self.assertEqual(time_constraints, kwargs['time_constraints'])
finally:
sys.stdout.close()
sys.stdout = orig
class ShellSampleListCommandTest(utils.BaseTestCase): class ShellSampleListCommandTest(utils.BaseTestCase):
@@ -359,24 +339,19 @@ class ShellSampleListCommandTest(utils.BaseTestCase):
self.args.query = None self.args.query = None
self.args.limit = None self.args.limit = None
@mock.patch('sys.stdout', new=six.StringIO())
def test_sample_list(self): def test_sample_list(self):
sample_list = [samples.Sample(mock.Mock(), sample) sample_list = [samples.Sample(mock.Mock(), sample)
for sample in self.SAMPLES] for sample in self.SAMPLES]
self.cc.samples.list.return_value = sample_list self.cc.samples.list.return_value = sample_list
org_stdout = sys.stdout ceilometer_shell.do_sample_list(self.cc, self.args)
try: self.cc.samples.list.assert_called_once_with(
sys.stdout = output = six.StringIO() meter_name=self.METER,
ceilometer_shell.do_sample_list(self.cc, self.args) q=None,
self.cc.samples.list.assert_called_once_with( limit=None)
meter_name=self.METER,
q=None,
limit=None)
finally:
sys.stdout = org_stdout
self.assertEqual(output.getvalue(), '''\ self.assertEqual('''\
+--------------------------------------+----------+-------+----------------\ +--------------------------------------+----------+-------+----------------\
+------+---------------------+ +------+---------------------+
| Resource ID | Name | Type | Volume \ | Resource ID | Name | Type | Volume \
@@ -393,7 +368,7 @@ class ShellSampleListCommandTest(utils.BaseTestCase):
| % | 2013-10-15T05:40:29 | | % | 2013-10-15T05:40:29 |
+--------------------------------------+----------+-------+----------------\ +--------------------------------------+----------+-------+----------------\
+------+---------------------+ +------+---------------------+
''') ''', sys.stdout.getvalue())
class ShellSampleCreateCommandTest(utils.BaseTestCase): class ShellSampleCreateCommandTest(utils.BaseTestCase):
@@ -427,19 +402,15 @@ class ShellSampleCreateCommandTest(utils.BaseTestCase):
self.args.resource_id = self.RESOURCE_ID self.args.resource_id = self.RESOURCE_ID
self.args.sample_volume = self.SAMPLE_VOLUME self.args.sample_volume = self.SAMPLE_VOLUME
@mock.patch('sys.stdout', new=six.StringIO())
def test_sample_create(self): def test_sample_create(self):
ret_sample = [samples.Sample(mock.Mock(), sample) ret_sample = [samples.Sample(mock.Mock(), sample)
for sample in self.SAMPLE] for sample in self.SAMPLE]
self.cc.samples.create.return_value = ret_sample self.cc.samples.create.return_value = ret_sample
org_stdout = sys.stdout
try:
sys.stdout = output = six.StringIO()
ceilometer_shell.do_sample_create(self.cc, self.args)
finally:
sys.stdout = org_stdout
self.assertEqual(output.getvalue(), '''\ ceilometer_shell.do_sample_create(self.cc, self.args)
self.assertEqual('''\
+-------------------+---------------------------------------------+ +-------------------+---------------------------------------------+
| Property | Value | | Property | Value |
+-------------------+---------------------------------------------+ +-------------------+---------------------------------------------+
@@ -455,7 +426,7 @@ class ShellSampleCreateCommandTest(utils.BaseTestCase):
| user_id | 21b442b8101d407d8242b6610e0ed0eb | | user_id | 21b442b8101d407d8242b6610e0ed0eb |
| volume | 1.0 | | volume | 1.0 |
+-------------------+---------------------------------------------+ +-------------------+---------------------------------------------+
''') ''', sys.stdout.getvalue())
class ShellQuerySamplesCommandTest(utils.BaseTestCase): class ShellQuerySamplesCommandTest(utils.BaseTestCase):
@@ -487,17 +458,13 @@ class ShellQuerySamplesCommandTest(utils.BaseTestCase):
self.args.orderby = self.QUERY["orderby"] self.args.orderby = self.QUERY["orderby"]
self.args.limit = self.QUERY["limit"] self.args.limit = self.QUERY["limit"]
@mock.patch('sys.stdout', new=six.StringIO())
def test_query(self): def test_query(self):
ret_sample = [samples.Sample(mock.Mock(), sample) ret_sample = [samples.Sample(mock.Mock(), sample)
for sample in self.SAMPLE] for sample in self.SAMPLE]
self.cc.query_samples.query.return_value = ret_sample self.cc.query_samples.query.return_value = ret_sample
org_stdout = sys.stdout
try: ceilometer_shell.do_query_samples(self.cc, self.args)
sys.stdout = output = six.StringIO()
ceilometer_shell.do_query_samples(self.cc, self.args)
finally:
sys.stdout = org_stdout
self.assertEqual('''\ self.assertEqual('''\
+--------------------------------------+----------+-------+--------+---------\ +--------------------------------------+----------+-------+--------+---------\
@@ -510,7 +477,7 @@ class ShellQuerySamplesCommandTest(utils.BaseTestCase):
| 2014-02-19T05:50:16.673604 | | 2014-02-19T05:50:16.673604 |
+--------------------------------------+----------+-------+--------+---------\ +--------------------------------------+----------+-------+--------+---------\
-+----------------------------+ -+----------------------------+
''', output.getvalue()) ''', sys.stdout.getvalue())
class ShellQueryAlarmsCommandTest(utils.BaseTestCase): class ShellQueryAlarmsCommandTest(utils.BaseTestCase):
@@ -549,17 +516,13 @@ class ShellQueryAlarmsCommandTest(utils.BaseTestCase):
self.args.orderby = self.QUERY["orderby"] self.args.orderby = self.QUERY["orderby"]
self.args.limit = self.QUERY["limit"] self.args.limit = self.QUERY["limit"]
@mock.patch('sys.stdout', new=six.StringIO())
def test_query(self): def test_query(self):
ret_alarm = [alarms.Alarm(mock.Mock(), alarm) ret_alarm = [alarms.Alarm(mock.Mock(), alarm)
for alarm in self.ALARM] for alarm in self.ALARM]
self.cc.query_alarms.query.return_value = ret_alarm self.cc.query_alarms.query.return_value = ret_alarm
org_stdout = sys.stdout
try: ceilometer_shell.do_query_alarms(self.cc, self.args)
sys.stdout = output = six.StringIO()
ceilometer_shell.do_query_alarms(self.cc, self.args)
finally:
sys.stdout = org_stdout
self.assertEqual('''\ self.assertEqual('''\
+--------------------------------------+------------------+-------+---------\ +--------------------------------------+------------------+-------+---------\
@@ -577,7 +540,7 @@ class ShellQueryAlarmsCommandTest(utils.BaseTestCase):
+--------------------------------------+------------------+-------+---------\ +--------------------------------------+------------------+-------+---------\
+------------+--------------------------------------------------------------\ +------------+--------------------------------------------------------------\
----------------------------------------+ ----------------------------------------+
''', output.getvalue()) ''', sys.stdout.getvalue())
class ShellQueryAlarmHistoryCommandTest(utils.BaseTestCase): class ShellQueryAlarmHistoryCommandTest(utils.BaseTestCase):
@@ -606,17 +569,13 @@ class ShellQueryAlarmHistoryCommandTest(utils.BaseTestCase):
self.args.orderby = self.QUERY["orderby"] self.args.orderby = self.QUERY["orderby"]
self.args.limit = self.QUERY["limit"] self.args.limit = self.QUERY["limit"]
@mock.patch('sys.stdout', new=six.StringIO())
def test_query(self): def test_query(self):
ret_alarm_history = [alarms.AlarmChange(mock.Mock(), alarm_history) ret_alarm_history = [alarms.AlarmChange(mock.Mock(), alarm_history)
for alarm_history in self.ALARM_HISTORY] for alarm_history in self.ALARM_HISTORY]
self.cc.query_alarm_history.query.return_value = ret_alarm_history self.cc.query_alarm_history.query.return_value = ret_alarm_history
org_stdout = sys.stdout
try: ceilometer_shell.do_query_alarm_history(self.cc, self.args)
sys.stdout = output = six.StringIO()
ceilometer_shell.do_query_alarm_history(self.cc, self.args)
finally:
sys.stdout = org_stdout
self.assertEqual('''\ self.assertEqual('''\
+----------------------------------+--------------------------------------+-\ +----------------------------------+--------------------------------------+-\
@@ -634,7 +593,7 @@ rule change | {"threshold": 42.0, "evaluation_periods": 4} | 2014-03-11T16:0\
+----------------------------------+--------------------------------------+-\ +----------------------------------+--------------------------------------+-\
------------+----------------------------------------------+----------------\ ------------+----------------------------------------------+----------------\
------------+ ------------+
''', output.getvalue()) ''', sys.stdout.getvalue())
class ShellStatisticsTest(utils.BaseTestCase): class ShellStatisticsTest(utils.BaseTestCase):