particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

PR#3846: cli: streamline python/json options in call command

Browse source 
Merges #3846
https://pagure.io/koji/pull-request/3846

Fixes #3852
https://pagure.io/koji/issue/3852
This commit is contained in:
Tomas Kopecek 2023-07-04 08:55:20 +02:00
commit c1fae34cb4
2 changed files with 111 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

76
cli/koji_cli/commands.py
View file

@ -963,39 +963,77 @@ def handle_call(goptions, session, args):
usage = """\
usage: %prog call [options] <name> [<arg> ...]
<arg> values of the form NAME=VALUE are treated as keyword arguments
Note, that you can use global option --noauth for anonymous calls here"""
usage = textwrap.dedent(usage)
parser = OptionParser(usage=get_usage_str(usage))
parser.add_option("--python", action="store_true",
parser.add_option("-p", "--python", action="store_true",
help="Use python syntax for RPC parameter values")
parser.add_option("--kwargs",
help="Specify keyword arguments as a dictionary (implies --python)")
help="Specify keyword arguments as a dictionary (implies --python or "
"--json-input)")
parser.add_option("-j", "--json", action="store_true",
help="Use JSON syntax for input and output")
parser.add_option("--json-input", action="store_true", help="Use JSON syntax for input")
parser.add_option("--json-output", action="store_true", help="Use JSON syntax for output")
parser.add_option("-b", "--bare-strings", action="store_true",
help="Treat invalid json/python as bare strings")
(options, args) = parser.parse_args(args)
if len(args) < 1:
parser.error("Please specify the name of the XML-RPC method")
if options.kwargs:
if options.json:
options.json_input = True
options.json_output = True
if options.python and options.json_input:
parser.error('The --python option conflicts with using --json-input')
if options.kwargs and not options.json_input:
# for backwards compatibility, --python is implied
options.python = True
if options.python and ast is None:
parser.error("The ast module is required to read python syntax")
if options.json_output and json is None:
parser.error("The json module is required to output JSON syntax")
activate_session(session, goptions)
name = args[0]
non_kw = []
kw = {}
if options.python:
non_kw = [ast.literal_eval(a) for a in args[1:]]
if options.kwargs:
kw = ast.literal_eval(options.kwargs)
else:
for arg in args[1:]:
if arg.find('=') != -1:
key, value = arg.split('=', 1)
kw[key] = arg_filter(value)
if (options.json_output or options.json_input) and json is None:
parser.error("The json module is required to use JSON syntax")
def parse_arg(arg):
try:
if options.python:
return ast.literal_eval(arg)
elif options.json_input:
return json.loads(arg)
else:
non_kw.append(arg_filter(arg))
return arg_filter(arg)
except ValueError as e:
if options.bare_strings:
return arg
else:
parser.error("Invalid value: %r" % arg)
# the method to call
name = args[0]
# base kw args
# we update with name=value args later
kw = {}
if options.kwargs:
kw = parse_arg(options.kwargs)
kw_pat = re.compile(r'^([^\W0-9]\w*)=(.*)$')
# read the args
non_kw = []
for arg in args[1:]:
m = kw_pat.match(arg)
if m:
key, value = m.groups()
kw[key] = parse_arg(value)
else:
non_kw.append(parse_arg(arg))
# make the call
activate_session(session, goptions)
response = getattr(session, name).__call__(*non_kw, **kw)
# print the result
if options.json_output:
print(json.dumps(response, indent=2, separators=(',', ': '), cls=DatetimeJSONEncoder))
else:

60
tests/test_cli/test_call.py
View file

@ -22,6 +22,7 @@ class TestCall(utils.CliTestCase):
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
self.error_format = """Usage: %s call [options] <name> [<arg> ...]
<arg> values of the form NAME=VALUE are treated as keyword arguments
Note, that you can use global option --noauth for anonymous calls here
(Specify the --help global option for a list of other help options)
@ -57,6 +58,49 @@ Note, that you can use global option --noauth for anonymous calls here
self.session.ssl_login.assert_called_with(cert='/etc/pki/cert')
self.assert_console_message(stdout, "'%s'\n" % response[1])
@mock.patch('sys.stdout', new_callable=six.StringIO)
def test_handle_call_json_syntax(self, stdout):
"""Test handle_call with json input syntax"""
response = ["SUCCESS", "FAKE-RESPONSE"]
self.session.ssl_login.return_value = response[1]
# Invalid json syntax
arguments = ['--json-input', 'ssl_login', 'cert=/etc/pki/cert']
self.assert_system_exit(
handle_call,
self.options, self.session, arguments,
stderr=self.format_error_message("Invalid value: '/etc/pki/cert'"),
activate_session=None)
self.activate_session_mock.assert_not_called()
# Incompatible opts
arguments = ['--json', '--python', 'ssl_login', 'cert=/etc/pki/cert']
self.assert_system_exit(
handle_call,
self.options, self.session, arguments,
stderr=self.format_error_message("The --python option conflicts with using --json-input"),
activate_session=None)
self.activate_session_mock.assert_not_called()
arguments = ['--json-input', 'ssl_login', '--kwargs', '{"cert":"/etc/pki/cert"}']
handle_call(self.options, self.session, arguments)
self.activate_session_mock.assert_called_with(self.session, self.options)
self.session.ssl_login.assert_called_with(cert='/etc/pki/cert')
self.assert_console_message(stdout, "'%s'\n" % response[1])
@mock.patch('sys.stdout', new_callable=six.StringIO)
def test_handle_call_bare_strings(self, stdout):
"""Test handle_call with bare string fallback"""
response = ["SUCCESS", "FAKE-RESPONSE"]
self.session.ssl_login.return_value = response[1]
# Invalid json syntax, but with bare-string fallback
arguments = ['--json-input', '--bare-strings', 'ssl_login', 'cert=/etc/pki/cert']
handle_call(self.options, self.session, arguments)
self.activate_session_mock.assert_called_with(self.session, self.options)
self.session.ssl_login.assert_called_with(cert='/etc/pki/cert')
self.assert_console_message(stdout, "'%s'\n" % response[1])
@mock.patch('sys.stdout', new_callable=six.StringIO)
def test_handle_call_json_output(self, stdout):
"""Test handle_call with json output"""
@ -97,7 +141,7 @@ Note, that you can use global option --noauth for anonymous calls here
module = {
'ast': "The ast module is required to read python syntax",
'json': "The json module is required to output JSON syntax",
'json': "The json module is required to use JSON syntax",
}
for mod, msg in module.items():
@ -115,15 +159,19 @@ Note, that you can use global option --noauth for anonymous calls here
handle_call,
"""Usage: %s call [options] <name> [<arg> ...]
<arg> values of the form NAME=VALUE are treated as keyword arguments
Note, that you can use global option --noauth for anonymous calls here
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--python Use python syntax for RPC parameter values
--kwargs=KWARGS Specify keyword arguments as a dictionary (implies
--python)
--json-output Use JSON syntax for output
-h, --help show this help message and exit
-p, --python Use python syntax for RPC parameter values
--kwargs=KWARGS Specify keyword arguments as a dictionary (implies
--python or --json-input)
-j, --json Use JSON syntax for input and output
--json-input Use JSON syntax for input
--json-output Use JSON syntax for output
-b, --bare-strings Treat invalid json/python as bare strings
""" % self.progname)