diff --git a/gnmi_api_client/__init__.py b/gnmi_api_client/__init__.py new file mode 100644 index 000000000..9ab1b13d2 --- /dev/null +++ b/gnmi_api_client/__init__.py @@ -0,0 +1,3 @@ +from gnmi_api_client.client import GnmiApiClient, main + +__all__ = ["GnmiApiClient", "main"] diff --git a/gnmi_api_client/client.py b/gnmi_api_client/client.py new file mode 100644 index 000000000..2495ad496 --- /dev/null +++ b/gnmi_api_client/client.py @@ -0,0 +1,72 @@ +import argparse +import shlex +import sys + +from gnmi_api_client.input_formatter import ShowCliToGnmiPathConverter +from gnmi_api_client.output_formatter import ( + DummyTabularFormatter, + EXAMPLE_INTERFACES_STATUS_JSON, + EXAMPLE_INTERFACES_STATUS_PATH, +) + + +DEFAULT_HOST = "localhost" +DEFAULT_PORT = 50052 + + +class GnmiApiClient: + def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, is_test_flag: bool = True): + self.host = host + self.port = port + self.is_test_flag = is_test_flag + self._output_formatter = DummyTabularFormatter() + + def format_input(self, show_cli_str: str) -> str: + tokens = shlex.split(show_cli_str) + return ShowCliToGnmiPathConverter(tokens).convert() + + def get_gnmi_result(self, gnmi_path: str) -> dict: + if self.is_test_flag: + if gnmi_path != EXAMPLE_INTERFACES_STATUS_PATH: + raise NotImplementedError( + f"test-mode stub only supports {EXAMPLE_INTERFACES_STATUS_PATH}, got {gnmi_path}" + ) + return EXAMPLE_INTERFACES_STATUS_JSON + + # Real gNMI transport against f"{self.host}:{self.port}" is not wired up yet. + raise NotImplementedError( + f"real gNMI client against {self.host}:{self.port} is not implemented yet" + ) + + def format_output(self, gnmi_path: str, gnmi_json: dict) -> str: + return self._output_formatter.format(gnmi_path, gnmi_json) + + def run(self, show_cli_str: str) -> str: + gnmi_path = self.format_input(show_cli_str) + gnmi_json = self.get_gnmi_result(gnmi_path) + return self.format_output(gnmi_path, gnmi_json) + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser( + description="Convert a SONiC show CLI command into a gNMI SHOW path, fetch the result, and format it into CLI tabular format." + ) + parser.add_argument("command", help="show CLI command string, e.g. 'show interfaces status'") + parser.add_argument("--host", default=DEFAULT_HOST, help="gNMI server host (default: localhost)") + parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="gNMI server port (default: 50052)") + parser.add_argument( + "--no-test-flag", + dest="is_test_flag", + action="store_false", + help="disable the hard-coded test fixture and attempt a real gNMI connection", + ) + parser.set_defaults(is_test_flag=True) + args = parser.parse_args(argv) + + client = GnmiApiClient(host=args.host, port=args.port, is_test_flag=args.is_test_flag) + print(client.run(args.command)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/gnmi_api_client/input_formatter.py b/gnmi_api_client/input_formatter.py new file mode 100644 index 000000000..657a372d5 --- /dev/null +++ b/gnmi_api_client/input_formatter.py @@ -0,0 +1,76 @@ +NO_COMMAND_ERROR = "No command provided for conversion" +EMPTY_COMMAND_ERROR = "Empty command" +INVALID_COMMAND_ERROR = "Command must start with 'show'" +NO_PATH_ERROR = "No path segments after 'show'" +SHORT_OPTION_ERROR = "Short options are not supported" +INVALID_OPTION_TOKEN_ERROR = "Invalid option token: --" +INVALID_LONG_OPTION_ERROR = "Invalid long option '--'" + + +class OptionException(Exception): + pass + + +def has_special_char(text: str) -> bool: + return "=" in text or "]" in text or "[" in text + + +def escape_gnmi(text: str) -> str: + # Escape only '/' → '\/' + return text.replace("/", r"\/") + + +class ShowCliToGnmiPathConverter: + def __init__(self, tokens): + self.tokens = tokens + + def parseLongOption(self, token: str): + # --flag -> ('flag', 'True') + # --key=value -> ('key', 'value') + if token == "--": + raise OptionException(INVALID_OPTION_TOKEN_ERROR) + + body = token[2:] + if not body: + raise OptionException(INVALID_LONG_OPTION_ERROR) + + if "=" in body: + name, value = body.split("=", 1) + if not name: + raise OptionException("Invalid long option: missing name before '='") + if has_special_char(name) or has_special_char(value): + raise OptionException("Invalid long option: key/value cannot contain =,[,]") + return name, escape_gnmi(value) + + return body, "True" + + def convert(self) -> str: + tokens = self.tokens + if not tokens: + raise OptionException(EMPTY_COMMAND_ERROR) + if tokens[0].lower() != "show": + raise OptionException(INVALID_COMMAND_ERROR) + + tokens = tokens[1:] # drop 'show' + out = ["SHOW"] + + for tok in tokens: + if tok.startswith("-") and not tok.startswith("--"): + raise OptionException(f"{SHORT_OPTION_ERROR}: '{tok}'") + + if tok.startswith("--"): + if len(out) == 1: + raise ValueError("Option before first path segment") + key, val = self.parseLongOption(tok) + out.append(f"[{key}={val}]") + continue + + if has_special_char(tok): + raise ValueError("Invalid characters inside of non option") + + out.append("/") + out.append(escape_gnmi(tok)) + + if len(out) == 1: + raise OptionException(NO_PATH_ERROR) + return "".join(out) diff --git a/gnmi_api_client/output_formatter.py b/gnmi_api_client/output_formatter.py new file mode 100644 index 000000000..db414a545 --- /dev/null +++ b/gnmi_api_client/output_formatter.py @@ -0,0 +1,45 @@ +from tabulate import tabulate + + +EXAMPLE_INTERFACES_STATUS_PATH = "SHOW/interfaces/status" + +EXAMPLE_INTERFACES_STATUS_JSON = { + "interfaces": [ + { + "Interface": "Ethernet0", + "Lanes": "0,1,2,3", + "Speed": "100G", + "MTU": "9100", + "FEC": "rs", + "Alias": "etp1", + "Admin": "up", + "Oper": "up", + }, + { + "Interface": "Ethernet4", + "Lanes": "4,5,6,7", + "Speed": "100G", + "MTU": "9100", + "FEC": "rs", + "Alias": "etp2", + "Admin": "down", + "Oper": "down", + }, + ] +} + +_INTERFACES_STATUS_HEADERS = [ + "Interface", "Lanes", "Speed", "MTU", "FEC", "Alias", "Admin", "Oper", +] + + +class DummyTabularFormatter: + def format(self, gnmi_path: str, gnmi_json: dict) -> str: + if gnmi_path != EXAMPLE_INTERFACES_STATUS_PATH: + raise ValueError(f"no dummy formatter for path {gnmi_path}") + + rows = [ + [entry.get(col, "") for col in _INTERFACES_STATUS_HEADERS] + for entry in gnmi_json.get("interfaces", []) + ] + return tabulate(rows, headers=_INTERFACES_STATUS_HEADERS, tablefmt="simple") diff --git a/setup.py b/setup.py index ea0e949ab..b975c2b65 100644 --- a/setup.py +++ b/setup.py @@ -58,6 +58,7 @@ 'crm', 'debug', 'generic_config_updater', + 'gnmi_api_client', 'dump', 'dump.plugins', 'pfcwd',