Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 45 additions & 38 deletions pymodbus/repl/client/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,51 @@ def convert(self, value, param, ctx):
return None


def _process_args(args: list, string: bool=True):
"""Internal function to parse arguments provided on command line.

:param args: Array of argument values
:param string: True if arguments values are strings, false if argument values are integers

:return Tuple, where the first member is hash of parsed values, and second is boolean flag
indicating if parsing succeeded.
"""
kwargs = {}
execute = True
skip_index = None

def _parse_val(arg_name, val):
if not string:
if "," in val:
val = val.split(",")
val = [int(v, 0) for v in val]
else:
val = int(val, 0)
kwargs[arg_name] = val

for i, arg in enumerate(args):
if i == skip_index:
continue
arg = arg.strip()
if "=" in arg:
arg_name, val = arg.split("=")
_parse_val(arg_name, val)
else:
arg_name, val = arg, args[i + 1]
try:
_parse_val(arg_name, val)
skip_index = i + 1
except TypeError:
click.secho("Error parsing arguments!", fg="yellow")
execute = False
break
except ValueError:
click.secho("Error parsing argument", fg="yellow")
execute = False
break
return kwargs, execute


def cli(client): # noqa: C901 pylint: disable=too-complex
"""Run client definition."""
use_keys = KeyBindings()
Expand All @@ -142,44 +187,6 @@ def _(event):
buffer = event.cli.current_buffer
buffer.complete_state = None

def _process_args(args, string=True):
kwargs = {}
execute = True
skip_index = None
for i, arg in enumerate(args):
if i == skip_index:
continue
arg = arg.strip()
if "=" in arg:
arg_name, val = arg.split("=")
if not string:
if "," in val:
val = val.split(",")
val = [int(v) for v in val]
else:
val = int(val)
kwargs[arg_name] = val
else:
arg_name, val = arg, args[i + 1]
try:
if not string:
if "," in val:
val = val.split(",")
val = [int(v) for v in val]
else:
val = int(val)
kwargs[arg_name] = val
skip_index = i + 1
except TypeError:
click.secho("Error parsing arguments!", fg="yellow")
execute = False
break
except ValueError:
click.secho("Error parsing argument", fg="yellow")
execute = False
break
return kwargs, execute

session = PromptSession(
lexer=PygmentsLexer(PythonLexer),
completer=CmdCompleter(client),
Expand Down
49 changes: 49 additions & 0 deletions test/test_repl_client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
"""Test client sync."""
import pytest

from pymodbus.repl.client.main import _process_args as process_args


def test_repl_client_process_args():
"""Test argument processing in repl.client.main (_process_args function)."""
resp = process_args(["address=11"], False)
assert resp == ({"address": 11}, True)

resp = process_args(["address=0x11"], False)
assert resp == ({"address": 17}, True)

resp = process_args(["address=0b11"], False)
assert resp == ({"address": 3}, True)

resp = process_args(["address=0o11"], False)
assert resp == ({"address": 9}, True)

resp = process_args(["address=11", "value=0x10"], False)
assert resp == ({"address": 11, "value": 16}, True)

resp = process_args(["value=11", "address=0x10"], False)
assert resp == ({"address": 16, "value": 11}, True)

resp = process_args(["address=0b11", "value=0x10"], False)
assert resp == ({"address": 3, "value": 16}, True)

try:
resp = process_args(["address=0xhj", "value=0x10"], False)
except ValueError:
pass
except Exception as exc: # pylint: disable=broad-except
pytest.fail(f"Exception in _process_args: {exc}")

try:
resp = process_args(["address=11ah", "value=0x10"], False)
except ValueError:
pass
except Exception as exc: # pylint: disable=broad-except
pytest.fail(f"Exception in _process_args: {exc}")

try:
resp = process_args(["address=0b12", "value=0x10"], False)
except ValueError:
pass
except Exception as exc: # pylint: disable=broad-except
pytest.fail(f"Exception in _process_args: {exc}")