diff options
| -rw-r--r-- | cloudinit/cmd/query.py | 43 | ||||
| -rw-r--r-- | cloudinit/cmd/tests/test_query.py | 397 |
2 files changed, 272 insertions, 168 deletions
diff --git a/cloudinit/cmd/query.py b/cloudinit/cmd/query.py index 0fb48ebd..03b77c17 100644 --- a/cloudinit/cmd/query.py +++ b/cloudinit/cmd/query.py @@ -1,6 +1,17 @@ # This file is part of cloud-init. See LICENSE file for license information. -"""Query standardized instance metadata from the command line.""" +"""Query standardized instance metadata provided to machine, returning a JSON +structure. + +Some instance-data values may be binary on some platforms, such as userdata and +vendordata. Attempt to decompress and decode UTF-8 any binary values. + +Any binary values in the instance metadata will be base64-encoded and prefixed +with "ci-b64:" in the output. userdata and, where applicable, vendordata may +be provided to the machine gzip-compressed (and therefore as binary data). +query will attempt to decompress these to a string before emitting the JSON +output; if this fails, they are treated as binary. +""" import argparse from errno import EACCES @@ -30,7 +41,7 @@ def get_parser(parser=None): """ if not parser: parser = argparse.ArgumentParser( - prog=NAME, description='Query cloud-init instance data') + prog=NAME, description=__doc__) parser.add_argument( '-d', '--debug', action='store_true', default=False, help='Add verbose messages during template render') @@ -52,8 +63,10 @@ def get_parser(parser=None): ' /var/lib/cloud/instance/vendor-data.txt')) parser.add_argument( 'varname', type=str, nargs='?', - help=('A dot-delimited instance data variable to query from' - ' instance-data query. For example: v2.local_hostname')) + help=('A dot-delimited specific variable to query from' + ' instance-data. For example: v1.local_hostname. If the' + ' value is not JSON serializable, it will be base64-encoded and' + ' will contain the prefix "ci-b64:". ')) parser.add_argument( '-a', '--all', action='store_true', default=False, dest='dump_all', help='Dump all available instance-data') @@ -65,6 +78,24 @@ def get_parser(parser=None): return parser +def load_userdata(ud_file_path): + """Attempt to return a string of user-data from ud_file_path + + Attempt to decode or decompress if needed. + If unable to decode the content, raw bytes will be returned. + + @returns: String of uncompressed userdata if possible, otherwise bytes. + """ + try: + return util.load_file(ud_file_path) + except UnicodeDecodeError: + encoded_data = util.load_file(ud_file_path, decode=False) + try: + return util.decomp_gzip(encoded_data, quiet=False) + except util.DecompressionError: + return encoded_data + + def handle_args(name, args): """Handle calls to 'cloud-init query' as a subcommand.""" paths = None @@ -121,8 +152,8 @@ def handle_args(name, args): instance_data['vendordata'] = ( '<%s> file:%s' % (REDACT_SENSITIVE_VALUE, vendor_data_fn)) else: - instance_data['userdata'] = util.load_file(user_data_fn) - instance_data['vendordata'] = util.load_file(vendor_data_fn) + instance_data['userdata'] = load_userdata(user_data_fn) + instance_data['vendordata'] = load_userdata(vendor_data_fn) if args.format: payload = '## template: jinja\n{fmt}'.format(fmt=args.format) rendered_payload = render_jinja_payload( diff --git a/cloudinit/cmd/tests/test_query.py b/cloudinit/cmd/tests/test_query.py index cb15b2d2..421031aa 100644 --- a/cloudinit/cmd/tests/test_query.py +++ b/cloudinit/cmd/tests/test_query.py @@ -1,195 +1,265 @@ # This file is part of cloud-init. See LICENSE file for license information. import errno -from io import StringIO +import gzip +from io import BytesIO +import json from textwrap import dedent -import os + +import pytest from collections import namedtuple from cloudinit.cmd import query from cloudinit.helpers import Paths from cloudinit.sources import ( REDACT_SENSITIVE_VALUE, INSTANCE_JSON_FILE, INSTANCE_JSON_SENSITIVE_FILE) -from cloudinit.tests.helpers import CiTestCase, mock -from cloudinit.util import ensure_dir, write_file +from cloudinit.tests.helpers import mock + +from cloudinit.util import b64e, write_file + +def _gzip_data(data): + with BytesIO() as iobuf: + with gzip.GzipFile(mode="wb", fileobj=iobuf) as gzfp: + gzfp.write(data) + return iobuf.getvalue() -class TestQuery(CiTestCase): - with_logs = True +@mock.patch("cloudinit.cmd.query.addLogHandlerCLI", lambda *args: "") +class TestQuery: args = namedtuple( 'queryargs', ('debug dump_all format instance_data list_keys user_data vendor_data' ' varname')) - def setUp(self): - super(TestQuery, self).setUp() - self.tmp = self.tmp_dir() - self.instance_data = self.tmp_path('instance-data', dir=self.tmp) + def _setup_paths(self, tmpdir, ud_val=None, vd_val=None): + """Write userdata and vendordata into a tmpdir. - def test_handle_args_error_on_missing_param(self): + Return: + 4-tuple : (paths, run_dir_path, userdata_path, vendordata_path) + """ + if ud_val: + user_data = tmpdir.join('user-data') + write_file(user_data.strpath, ud_val) + else: + user_data = None + if vd_val: + vendor_data = tmpdir.join('vendor-data') + write_file(vendor_data.strpath, vd_val) + else: + vendor_data = None + run_dir = tmpdir.join('run_dir') + run_dir.ensure_dir() + return ( + Paths({'run_dir': run_dir.strpath}), + run_dir, + user_data, + vendor_data + ) + + def test_handle_args_error_on_missing_param(self, caplog, capsys): """Error when missing required parameters and print usage.""" args = self.args( debug=False, dump_all=False, format=None, instance_data=None, list_keys=False, user_data=None, vendor_data=None, varname=None) - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - self.assertEqual(1, query.handle_args('anyname', args)) + with mock.patch( + "cloudinit.cmd.query.addLogHandlerCLI", return_value="" + ) as m_cli_log: + assert 1 == query.handle_args('anyname', args) expected_error = ( - 'ERROR: Expected one of the options: --all, --format, --list-keys' + 'Expected one of the options: --all, --format, --list-keys' ' or varname\n') - self.assertIn(expected_error, self.logs.getvalue()) - self.assertIn('usage: query', m_stdout.getvalue()) - self.assertIn(expected_error, m_stderr.getvalue()) + assert expected_error in caplog.text + out, _err = capsys.readouterr() + assert 'usage: query' in out + assert 1 == m_cli_log.call_count - def test_handle_args_error_on_missing_instance_data(self): + def test_handle_args_error_on_missing_instance_data(self, caplog, tmpdir): """When instance_data file path does not exist, log an error.""" - absent_fn = self.tmp_path('absent', dir=self.tmp) + absent_fn = tmpdir.join('absent') args = self.args( - debug=False, dump_all=True, format=None, instance_data=absent_fn, + debug=False, dump_all=True, format=None, + instance_data=absent_fn.strpath, list_keys=False, user_data='ud', vendor_data='vd', varname=None) - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - self.assertEqual(1, query.handle_args('anyname', args)) - self.assertIn( - 'ERROR: Missing instance-data file: %s' % absent_fn, - self.logs.getvalue()) - self.assertIn( - 'ERROR: Missing instance-data file: %s' % absent_fn, - m_stderr.getvalue()) + assert 1 == query.handle_args('anyname', args) - def test_handle_args_error_when_no_read_permission_instance_data(self): + msg = 'Missing instance-data file: %s' % absent_fn + assert msg in caplog.text + + def test_handle_args_error_when_no_read_permission_instance_data( + self, caplog, tmpdir + ): """When instance_data file is unreadable, log an error.""" - noread_fn = self.tmp_path('unreadable', dir=self.tmp) - write_file(noread_fn, 'thou shall not pass') + noread_fn = tmpdir.join('unreadable') + noread_fn.write('thou shall not pass') args = self.args( - debug=False, dump_all=True, format=None, instance_data=noread_fn, + debug=False, dump_all=True, format=None, + instance_data=noread_fn.strpath, list_keys=False, user_data='ud', vendor_data='vd', varname=None) - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - with mock.patch('cloudinit.cmd.query.util.load_file') as m_load: - m_load.side_effect = OSError(errno.EACCES, 'Not allowed') - self.assertEqual(1, query.handle_args('anyname', args)) - self.assertIn( - "ERROR: No read permission on '%s'. Try sudo" % noread_fn, - self.logs.getvalue()) - self.assertIn( - "ERROR: No read permission on '%s'. Try sudo" % noread_fn, - m_stderr.getvalue()) + with mock.patch('cloudinit.cmd.query.util.load_file') as m_load: + m_load.side_effect = OSError(errno.EACCES, 'Not allowed') + assert 1 == query.handle_args('anyname', args) + msg = "No read permission on '%s'. Try sudo" % noread_fn + assert msg in caplog.text - def test_handle_args_defaults_instance_data(self): + def test_handle_args_defaults_instance_data(self, caplog, tmpdir): """When no instance_data argument, default to configured run_dir.""" args = self.args( debug=False, dump_all=True, format=None, instance_data=None, list_keys=False, user_data=None, vendor_data=None, varname=None) - run_dir = self.tmp_path('run_dir', dir=self.tmp) - ensure_dir(run_dir) - paths = Paths({'run_dir': run_dir}) - self.add_patch('cloudinit.cmd.query.read_cfg_paths', 'm_paths') - self.m_paths.return_value = paths - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - self.assertEqual(1, query.handle_args('anyname', args)) - json_file = os.path.join(run_dir, INSTANCE_JSON_FILE) - self.assertIn( - 'ERROR: Missing instance-data file: %s' % json_file, - self.logs.getvalue()) - self.assertIn( - 'ERROR: Missing instance-data file: %s' % json_file, - m_stderr.getvalue()) + paths, run_dir, _, _ = self._setup_paths(tmpdir) + with mock.patch('cloudinit.cmd.query.read_cfg_paths') as m_paths: + m_paths.return_value = paths + assert 1 == query.handle_args('anyname', args) + json_file = run_dir.join(INSTANCE_JSON_FILE) + msg = 'Missing instance-data file: %s' % json_file.strpath + assert msg in caplog.text - def test_handle_args_root_fallsback_to_instance_data(self): + def test_handle_args_root_fallsback_to_instance_data(self, caplog, tmpdir): """When no instance_data argument, root falls back to redacted json.""" args = self.args( debug=False, dump_all=True, format=None, instance_data=None, list_keys=False, user_data=None, vendor_data=None, varname=None) - run_dir = self.tmp_path('run_dir', dir=self.tmp) - ensure_dir(run_dir) - paths = Paths({'run_dir': run_dir}) - self.add_patch('cloudinit.cmd.query.read_cfg_paths', 'm_paths') - self.m_paths.return_value = paths - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: + paths, run_dir, _, _ = self._setup_paths(tmpdir) + with mock.patch('cloudinit.cmd.query.read_cfg_paths') as m_paths: + m_paths.return_value = paths with mock.patch('os.getuid') as m_getuid: m_getuid.return_value = 0 - self.assertEqual(1, query.handle_args('anyname', args)) - json_file = os.path.join(run_dir, INSTANCE_JSON_FILE) - sensitive_file = os.path.join(run_dir, INSTANCE_JSON_SENSITIVE_FILE) - self.assertIn( - 'WARNING: Missing root-readable %s. Using redacted %s instead.' % ( - sensitive_file, json_file), - m_stderr.getvalue()) + assert 1 == query.handle_args('anyname', args) + json_file = run_dir.join(INSTANCE_JSON_FILE) + sensitive_file = run_dir.join(INSTANCE_JSON_SENSITIVE_FILE) + msg = ( + 'Missing root-readable %s. Using redacted %s instead.' % + ( + sensitive_file.strpath, json_file.strpath + ) + ) + assert msg in caplog.text - def test_handle_args_root_uses_instance_sensitive_data(self): - """When no instance_data argument, root uses semsitive json.""" - user_data = self.tmp_path('user-data', dir=self.tmp) - vendor_data = self.tmp_path('vendor-data', dir=self.tmp) - write_file(user_data, 'ud') - write_file(vendor_data, 'vd') - run_dir = self.tmp_path('run_dir', dir=self.tmp) - sensitive_file = os.path.join(run_dir, INSTANCE_JSON_SENSITIVE_FILE) - write_file(sensitive_file, '{"my-var": "it worked"}') - ensure_dir(run_dir) - paths = Paths({'run_dir': run_dir}) - self.add_patch('cloudinit.cmd.query.read_cfg_paths', 'm_paths') - self.m_paths.return_value = paths + @pytest.mark.parametrize( + 'ud_src,ud_expected,vd_src,vd_expected', + ( + ('hi mom', 'hi mom', 'hi pops', 'hi pops'), + ('ud'.encode('utf-8'), 'ud', 'vd'.encode('utf-8'), 'vd'), + (_gzip_data(b'ud'), 'ud', _gzip_data(b'vd'), 'vd'), + (_gzip_data('ud'.encode('utf-8')), 'ud', _gzip_data(b'vd'), 'vd'), + (_gzip_data(b'ud') + b'invalid', 'ci-b64:', + _gzip_data(b'vd') + b'invalid', 'ci-b64:'), + # non-utf-8 encodable content + ('hi mom'.encode('utf-16'), 'ci-b64://5oAGkAIABtAG8AbQA=', + 'hi pops'.encode('utf-16'), 'ci-b64://5oAGkAIABwAG8AcABzAA=='), + ) + ) + def test_handle_args_root_processes_user_data( + self, ud_src, ud_expected, vd_src, vd_expected, capsys, tmpdir + ): + """Support reading multiple user-data file content types""" + paths, run_dir, user_data, vendor_data = self._setup_paths( + tmpdir, ud_val=ud_src, vd_val=vd_src + ) + sensitive_file = run_dir.join(INSTANCE_JSON_SENSITIVE_FILE) + sensitive_file.write('{"my-var": "it worked"}') args = self.args( debug=False, dump_all=True, format=None, instance_data=None, - list_keys=False, user_data=vendor_data, vendor_data=vendor_data, - varname=None) - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: + list_keys=False, user_data=user_data.strpath, + vendor_data=vendor_data.strpath, varname=None) + with mock.patch('cloudinit.cmd.query.read_cfg_paths') as m_paths: + m_paths.return_value = paths with mock.patch('os.getuid') as m_getuid: m_getuid.return_value = 0 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual( - '{\n "my_var": "it worked",\n "userdata": "vd",\n ' - '"vendordata": "vd"\n}\n', m_stdout.getvalue()) + assert 0 == query.handle_args('anyname', args) + out, _err = capsys.readouterr() + cmd_output = json.loads(out) + assert "it worked" == cmd_output['my_var'] + if ud_expected == "ci-b64:": + ud_expected = "ci-b64:{}".format(b64e(ud_src)) + if vd_expected == "ci-b64:": + vd_expected = "ci-b64:{}".format(b64e(vd_src)) + assert ud_expected == cmd_output['userdata'] + assert vd_expected == cmd_output['vendordata'] - def test_handle_args_dumps_all_instance_data(self): + def test_handle_args_root_uses_instance_sensitive_data( + self, capsys, tmpdir + ): + """When no instance_data argument, root uses sensitive json.""" + paths, run_dir, user_data, vendor_data = self._setup_paths( + tmpdir, ud_val='ud', vd_val='vd' + ) + sensitive_file = run_dir.join(INSTANCE_JSON_SENSITIVE_FILE) + sensitive_file.write('{"my-var": "it worked"}') + args = self.args( + debug=False, dump_all=True, format=None, instance_data=None, + list_keys=False, user_data=user_data.strpath, + vendor_data=vendor_data.strpath, varname=None) + with mock.patch('cloudinit.cmd.query.read_cfg_paths') as m_paths: + m_paths.return_value = paths + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 0 + assert 0 == query.handle_args('anyname', args) + expected = ( + '{\n "my_var": "it worked",\n "userdata": "ud",\n ' + '"vendordata": "vd"\n}\n' + ) + out, _err = capsys.readouterr() + assert expected == out + + def test_handle_args_dumps_all_instance_data(self, capsys, tmpdir): """When --all is specified query will dump all instance data vars.""" - write_file(self.instance_data, '{"my-var": "it worked"}') + instance_data = tmpdir.join('instance-data') + instance_data.write('{"my-var": "it worked"}') args = self.args( debug=False, dump_all=True, format=None, - instance_data=self.instance_data, list_keys=False, + instance_data=instance_data.strpath, list_keys=False, user_data='ud', vendor_data='vd', varname=None) - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual( + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 0 == query.handle_args('anyname', args) + expected = ( '{\n "my_var": "it worked",\n "userdata": "<%s> file:ud",\n' ' "vendordata": "<%s> file:vd"\n}\n' % ( - REDACT_SENSITIVE_VALUE, REDACT_SENSITIVE_VALUE), - m_stdout.getvalue()) + REDACT_SENSITIVE_VALUE, REDACT_SENSITIVE_VALUE + ) + ) + out, _err = capsys.readouterr() + assert expected == out - def test_handle_args_returns_top_level_varname(self): + def test_handle_args_returns_top_level_varname(self, capsys, tmpdir): """When the argument varname is passed, report its value.""" - write_file(self.instance_data, '{"my-var": "it worked"}') + instance_data = tmpdir.join('instance-data') + instance_data.write('{"my-var": "it worked"}') args = self.args( debug=False, dump_all=True, format=None, - instance_data=self.instance_data, list_keys=False, + instance_data=instance_data.strpath, list_keys=False, user_data='ud', vendor_data='vd', varname='my_var') - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual('it worked\n', m_stdout.getvalue()) + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 0 == query.handle_args('anyname', args) + out, _err = capsys.readouterr() + assert 'it worked\n' == out - def test_handle_args_returns_nested_varname(self): + def test_handle_args_returns_nested_varname(self, capsys, tmpdir): """If user_data file is a jinja template render instance-data vars.""" - write_file(self.instance_data, - '{"v1": {"key-2": "value-2"}, "my-var": "it worked"}') + instance_data = tmpdir.join('instance-data') + instance_data.write( + '{"v1": {"key-2": "value-2"}, "my-var": "it worked"}' + ) args = self.args( debug=False, dump_all=False, format=None, - instance_data=self.instance_data, user_data='ud', vendor_data='vd', - list_keys=False, varname='v1.key_2') - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual('value-2\n', m_stdout.getvalue()) + instance_data=instance_data.strpath, user_data='ud', + vendor_data='vd', list_keys=False, varname='v1.key_2') + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 0 == query.handle_args('anyname', args) + out, _err = capsys.readouterr() + assert 'value-2\n' == out - def test_handle_args_returns_standardized_vars_to_top_level_aliases(self): + def test_handle_args_returns_standardized_vars_to_top_level_aliases( + self, capsys, tmpdir + ): """Any standardized vars under v# are promoted as top-level aliases.""" - write_file( - self.instance_data, + instance_data = tmpdir.join('instance-data') + instance_data.write( '{"v1": {"v1_1": "val1.1"}, "v2": {"v2_2": "val2.2"},' ' "top": "gun"}') expected = dedent("""\ @@ -209,65 +279,68 @@ class TestQuery(CiTestCase): """) args = self.args( debug=False, dump_all=True, format=None, - instance_data=self.instance_data, user_data='ud', vendor_data='vd', - list_keys=False, varname=None) - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual(expected, m_stdout.getvalue()) + instance_data=instance_data.strpath, user_data='ud', + vendor_data='vd', list_keys=False, varname=None) + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 0 == query.handle_args('anyname', args) + out, _err = capsys.readouterr() + assert expected == out - def test_handle_args_list_keys_sorts_top_level_keys_when_no_varname(self): + def test_handle_args_list_keys_sorts_top_level_keys_when_no_varname( + self, capsys, tmpdir + ): """Sort all top-level keys when only --list-keys provided.""" - write_file( - self.instance_data, + instance_data = tmpdir.join('instance-data') + instance_data.write( '{"v1": {"v1_1": "val1.1"}, "v2": {"v2_2": "val2.2"},' ' "top": "gun"}') expected = 'top\nuserdata\nv1\nv1_1\nv2\nv2_2\nvendordata\n' args = self.args( debug=False, dump_all=False, format=None, - instance_data=self.instance_data, list_keys=True, user_data='ud', - vendor_data='vd', varname=None) - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual(expected, m_stdout.getvalue()) + instance_data=instance_data.strpath, list_keys=True, + user_data='ud', vendor_data='vd', varname=None) + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 0 == query.handle_args('anyname', args) + out, _err = capsys.readouterr() + assert expected == out - def test_handle_args_list_keys_sorts_nested_keys_when_varname(self): + def test_handle_args_list_keys_sorts_nested_keys_when_varname( + self, capsys, tmpdir + ): """Sort all nested keys of varname object when --list-keys provided.""" - write_file( - self.instance_data, + instance_data = tmpdir.join('instance-data') + instance_data.write( '{"v1": {"v1_1": "val1.1", "v1_2": "val1.2"}, "v2":' + ' {"v2_2": "val2.2"}, "top": "gun"}') expected = 'v1_1\nv1_2\n' args = self.args( debug=False, dump_all=False, format=None, - instance_data=self.instance_data, list_keys=True, + instance_data=instance_data.strpath, list_keys=True, user_data='ud', vendor_data='vd', varname='v1') - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(0, query.handle_args('anyname', args)) - self.assertEqual(expected, m_stdout.getvalue()) + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 0 == query.handle_args('anyname', args) + out, _err = capsys.readouterr() + assert expected == out - def test_handle_args_list_keys_errors_when_varname_is_not_a_dict(self): + def test_handle_args_list_keys_errors_when_varname_is_not_a_dict( + self, caplog, tmpdir + ): """Raise an error when --list-keys and varname specify a non-list.""" - write_file( - self.instance_data, + instance_data = tmpdir.join('instance-data') + instance_data.write( '{"v1": {"v1_1": "val1.1", "v1_2": "val1.2"}, "v2": ' + '{"v2_2": "val2.2"}, "top": "gun"}') - expected_error = "ERROR: --list-keys provided but 'top' is not a dict" + expected_error = "--list-keys provided but 'top' is not a dict" args = self.args( debug=False, dump_all=False, format=None, - instance_data=self.instance_data, list_keys=True, user_data='ud', - vendor_data='vd', varname='top') - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with mock.patch('os.getuid') as m_getuid: - m_getuid.return_value = 100 - self.assertEqual(1, query.handle_args('anyname', args)) - self.assertEqual('', m_stdout.getvalue()) - self.assertIn(expected_error, m_stderr.getvalue()) + instance_data=instance_data.strpath, list_keys=True, + user_data='ud', vendor_data='vd', varname='top') + with mock.patch('os.getuid') as m_getuid: + m_getuid.return_value = 100 + assert 1 == query.handle_args('anyname', args) + assert expected_error in caplog.text # vi: ts=4 expandtab |