diff --git a/gnmi_cli_converter/__init__.py b/gnmi_cli_converter/__init__.py new file mode 100644 index 000000000..19ab5a208 --- /dev/null +++ b/gnmi_cli_converter/__init__.py @@ -0,0 +1,49 @@ +""" +GNMI CLI Converter + +A framework for converting GNMI JSON to CLI format output. + +Main interfaces: +- CLIConverter: Main converter class +- CommandInfo: Command information dataclass + +Usage example: + >>> from gnmi_cli_converter import CLIConverter + >>> converter = CLIConverter() + >>> json_data = { + ... "egress_lossless_pool": {"Bytes": "12345"}, + ... "ingress_lossless_pool": {"Bytes": "67890"} + ... } + >>> output = converter.convert(["buffer_pool", "watermark"], json_data) + >>> print(output) + Shared pool maximum occupancy: + Pool Bytes + --------------------- ----- + egress_lossless_pool 12345 + ingress_lossless_pool 67890 +""" + +# Import commands module to trigger all command registrations +from . import commands + +# Export main interfaces +from .converter import CLIConverter, CommandInfo +from .exceptions import ( + CLIConverterError, + PathNotFoundError, + InvalidJSONError, + RenderError, + MissingFieldError +) + +__all__ = [ + 'CLIConverter', + 'CommandInfo', + 'CLIConverterError', + 'PathNotFoundError', + 'InvalidJSONError', + 'RenderError', + 'MissingFieldError', +] + +__version__ = '1.0.0' diff --git a/gnmi_cli_converter/commands/__init__.py b/gnmi_cli_converter/commands/__init__.py new file mode 100644 index 000000000..51e1aa93e --- /dev/null +++ b/gnmi_cli_converter/commands/__init__.py @@ -0,0 +1,16 @@ +""" +Commands Module + +Import all command modules to trigger @register decorator registration. +""" + +# Import all command modules to trigger @register decorators +from . import buffer_pool +from . import uptime + +# More command modules can be added in the future: +# from . import interfaces +# from . import queue +# from . import ipv6 +# from . import processes +# ... diff --git a/gnmi_cli_converter/commands/buffer_pool.py b/gnmi_cli_converter/commands/buffer_pool.py new file mode 100644 index 000000000..91dca9005 --- /dev/null +++ b/gnmi_cli_converter/commands/buffer_pool.py @@ -0,0 +1,82 @@ +""" +Buffer Pool Related Command Render Functions + +Contains: +- show buffer_pool watermark +- show buffer_pool persistent-watermark +""" + +from typing import Any, List, Dict +from natsort import natsorted +from ..utils import tabulate_dict +from ..registry import register + + +@register(["buffer_pool", "watermark"]) +def render_buffer_pool_watermark(json_data: Any, path_elems: List[str], options: Dict[str, Any]) -> str: + """ + show buffer_pool watermark + + Args: + json_data: JSON data returned by GNMI + path_elems: Path element array (not used by this command) + options: Options dict (not used by this command) + + Input JSON example: + { + "egress_lossless_pool": {"Bytes": "12345"}, + "egress_lossy_pool": {"Bytes": "67890"}, + "ingress_lossless_pool": {"Bytes": "24680"} + } + + Output example: + Shared pool maximum occupancy: + Pool Bytes + --------------------- ----- + egress_lossless_pool 12345 + egress_lossy_pool 67890 + ingress_lossless_pool 24680 + """ + if not json_data: + return "Shared pool maximum occupancy:\nNo data available" + + # Use natsorted for natural sorting + rows = [ + (pool, data.get("Bytes", "N/A")) + for pool, data in natsorted(json_data.items()) + ] + + return tabulate_dict( + title="Shared pool maximum occupancy:", + headers=["Pool", "Bytes"], + rows=rows, + stralign="right" + ) + + +@register(["buffer_pool", "persistent-watermark"]) +def render_buffer_pool_persistent_watermark(json_data: Any, path_elems: List[str], options: Dict[str, Any]) -> str: + """ + show buffer_pool persistent-watermark + + Format is the same as watermark, only the data source is different. + + Args: + json_data: JSON data returned by GNMI + path_elems: Path element array (not used by this command) + options: Options dict (not used by this command) + """ + if not json_data: + return "Shared pool maximum occupancy:\nNo data available" + + rows = [ + (pool, data.get("Bytes", "N/A")) + for pool, data in natsorted(json_data.items()) + ] + + return tabulate_dict( + title="Shared pool maximum occupancy:", + headers=["Pool", "Bytes"], + rows=rows, + stralign="right" + ) diff --git a/gnmi_cli_converter/commands/uptime.py b/gnmi_cli_converter/commands/uptime.py new file mode 100644 index 000000000..003e521c5 --- /dev/null +++ b/gnmi_cli_converter/commands/uptime.py @@ -0,0 +1,29 @@ +""" +Uptime Command Render Function + +Contains: +- show uptime +""" + +from typing import Any, List, Dict +from ..utils import passthrough +from ..registry import register + + +@register(["uptime"]) +def render_uptime(json_data: Any, path_elems: List[str], options: Dict[str, Any]) -> str: + """ + show uptime + + Args: + json_data: JSON data returned by GNMI + path_elems: Path element array (not used by this command) + options: Options dict (not used by this command) + + Input JSON example: + {"uptime": "up 3 weeks, 4 days, 10 hours, 15 minutes"} + + Output example: + up 3 weeks, 4 days, 10 hours, 15 minutes + """ + return passthrough(json_data, key="uptime") diff --git a/gnmi_cli_converter/converter.py b/gnmi_cli_converter/converter.py new file mode 100644 index 000000000..d788e18eb --- /dev/null +++ b/gnmi_cli_converter/converter.py @@ -0,0 +1,259 @@ +""" +CLI Converter Main Class + +Provides interface for converting GNMI JSON to CLI format. +""" + +from typing import List, Any, Dict, Optional, Union, Tuple +from dataclasses import dataclass +import re +from .registry import get_renderer, list_commands as registry_list_commands, is_supported as registry_is_supported +from .exceptions import PathNotFoundError, RenderError + +# Logger instance +from sonic_py_common.logger import Logger + +SYSLOG_IDENTIFIER = "gnmi_cli_converter" +log = Logger(SYSLOG_IDENTIFIER) + + +@dataclass +class CommandInfo: + """Command information (for callers to query supported command list)""" + path_elems: List[str] # Pattern with wildcards: ["interfaces", "errors", "*"] + command: str # "show interfaces errors " + + +class CLIConverter: + """ + Converter for GNMI JSON to CLI format. + + Example: + >>> converter = CLIConverter() + >>> json_data = {"egress_lossless_pool": {"Bytes": "12345"}} + >>> # Three equivalent ways to call convert: + >>> output = converter.convert(json_data, path_elems=["buffer_pool", "watermark"]) + >>> output = converter.convert(json_data, xpath="/buffer_pool/watermark") + >>> output = converter.convert(json_data, command="show buffer_pool watermark") + >>> print(output) + Shared pool maximum occupancy: + Pool Bytes + --------------------- ----- + egress_lossless_pool 12345 + """ + + @staticmethod + def _xpath_to_path_elems(xpath: str) -> Tuple[List[str], Dict[str, Any]]: + """ + Convert xpath string to path_elems list and options dict. + + Args: + xpath: XPath string, e.g."/interfaces/counters[printall=true][period=5]" + + Returns: + Tuple of (path_elems, options) + - path_elems: e.g. ["interfaces", "counters"] + - options: e.g. {"printall": True, "period": 5} + + Examples: + >>> CLIConverter._xpath_to_path_elems("/buffer_pool/watermark") + (['buffer_pool', 'watermark'], {}) + >>> CLIConverter._xpath_to_path_elems("/interfaces/counters[printall=true]") + (['interfaces', 'counters'], {'printall': True}) + """ + options = {} + + # Extract options from brackets: [key=value] + # Pattern matches [key=value] at any position + option_pattern = r'\[([^=\]]+)=([^\]]+)\]' + matches = re.findall(option_pattern, xpath) + for key, value in matches: + # Convert string values to appropriate types + key = key.strip() + value = value.strip() + if value.lower() == 'true': + options[key] = True + elif value.lower() == 'false': + options[key] = False + elif value.isdigit(): + options[key] = int(value) + else: + options[key] = value + + # Remove options from xpath to get clean path + clean_xpath = re.sub(option_pattern, '', xpath) + + # Remove leading/trailing slashes and split + path = clean_xpath.strip("/") + if not path: + return [], options + return path.split("/"), options + + @staticmethod + def _command_to_path_elems(command: str) -> Tuple[List[str], Dict[str, Any]]: + """ + Convert show command string to path_elems list and options dict. + + Args: + command: Show command string, e.g. "show interfaces counters --printall --period=5" + Path parameters (like interface names) are part of the path: + "show interfaces errors Ethernet0" -> ["interfaces", "errors", "Ethernet0"] + + Returns: + Tuple of (path_elems, options) + - path_elems: e.g. ["interfaces", "counters"] + - options: e.g. {"printall": True, "period": 5} + + Examples: + >>> CLIConverter._command_to_path_elems("show buffer_pool watermark") + (['buffer_pool', 'watermark'], {}) + >>> CLIConverter._command_to_path_elems("show interfaces counters --printall") + (['interfaces', 'counters'], {'printall': True}) + >>> CLIConverter._command_to_path_elems("show interfaces errors Ethernet0") + (['interfaces', 'errors', 'Ethernet0'], {}) + """ + options = {} + path_parts = [] + + parts = command.strip().split() + # Remove "show" prefix if present + if parts and parts[0].lower() == "show": + parts = parts[1:] + + for part in parts: + if part.startswith("--"): + # Parse option: --key=value or --flag + opt = part[2:] # Remove -- + if "=" in opt: + key, value = opt.split("=", 1) + # Convert string values to appropriate types + if value.lower() == 'true': + options[key] = True + elif value.lower() == 'false': + options[key] = False + elif value.isdigit(): + options[key] = int(value) + else: + options[key] = value + else: + # Flag without value, treat as True + options[opt] = True + else: + path_parts.append(part) + + return path_parts, options + + def convert( + self, + json_data: Any, + *, + path_elems: Optional[List[str]] = None, + xpath: Optional[str] = None, + command: Optional[str] = None, + options: Optional[Dict[str, Any]] = None + ) -> str: + """ + Convert GNMI JSON to CLI format text. + + Must provide exactly one of: path_elems, xpath, or command. + + Args: + json_data: JSON data returned by GNMI (already parsed to Python object) + path_elems: Path element list, e.g. ["interfaces", "counters"] + or with path parameter: ["interfaces", "errors", "Ethernet0"] + xpath: XPath string, e.g. "/interfaces/counters" + or with path parameter: "/interfaces/errors/Ethernet0" + or with options: "/interfaces/counters[printall=true]" + command: CLI command string, e.g. "show interfaces counters" + or with path parameter: "show interfaces errors Ethernet0" + or with options: "show interfaces counters --printall" + options: Optional dict of options, used with path_elems. + e.g. {"printall": True} + + Returns: + CLI format text output + + Raises: + PathNotFoundError: Path not supported + RenderError: Rendering failed + """ + # Validate: exactly one path specifier must be provided + specifiers = [path_elems is not None, xpath is not None, command is not None] + if sum(specifiers) != 1: + raise ValueError( + "Must provide exactly one of: path_elems, xpath, or command" + ) + + # Convert to path_elems (internal canonical format) and extract options + if xpath is not None: + path_elems, final_options = self._xpath_to_path_elems(xpath) + elif command is not None: + path_elems, final_options = self._command_to_path_elems(command) + else: + final_options = options or {} + + try: + # 1. Find render function + log.log_debug(f"Looking up renderer for path: {path_elems}") + render_func = get_renderer(path_elems) + + # 2. Call render function with options + log.log_debug(f"Rendering JSON data for path: {path_elems}, options: {final_options}") + output = render_func(json_data, path_elems, final_options) + + log.log_debug(f"Successfully rendered path: {path_elems}") + return output + + except PathNotFoundError: + log.log_warning(f"Path not found: {path_elems}") + raise + + except Exception as e: + log.log_error(f"Failed to render path {path_elems}: {e}") + raise RenderError(f"Failed to render path {path_elems}: {e}") from e + + def is_supported( + self, + *, + path_elems: Optional[List[str]] = None, + xpath: Optional[str] = None, + command: Optional[str] = None + ) -> bool: + """ + Check if the path is supported. + + Must provide exactly one of: path_elems, xpath, or command. + + Args: + path_elems: Path element array + xpath: XPath string + command: Show command string + + Returns: + True if supported, False otherwise + """ + # Convert to path_elems + if xpath is not None: + path_elems, _ = self._xpath_to_path_elems(xpath) + elif command is not None: + path_elems, _ = self._command_to_path_elems(command) + elif path_elems is None: + return False + + return registry_is_supported(path_elems) + + def list_commands(self) -> List[CommandInfo]: + """ + Get list of all supported commands. + + Returns: + List of command information + """ + entries = registry_list_commands() + return [ + CommandInfo( + path_elems=entry.path_pattern, + command=entry.command + ) + for entry in entries + ] diff --git a/gnmi_cli_converter/exceptions.py b/gnmi_cli_converter/exceptions.py new file mode 100644 index 000000000..b0befefad --- /dev/null +++ b/gnmi_cli_converter/exceptions.py @@ -0,0 +1,30 @@ +""" +CLI Converter Custom Exception Classes +""" + + +class CLIConverterError(Exception): + """CLI Converter base exception""" + pass + + +class PathNotFoundError(CLIConverterError): + """Path not registered""" + def __init__(self, path_elems): + self.path_elems = path_elems + super().__init__(f"Path not registered: {path_elems}") + + +class InvalidJSONError(CLIConverterError): + """Invalid JSON data""" + pass + + +class RenderError(CLIConverterError): + """Render failed""" + pass + + +class MissingFieldError(CLIConverterError): + """Required field missing""" + pass diff --git a/gnmi_cli_converter/registry.py b/gnmi_cli_converter/registry.py new file mode 100644 index 000000000..1b0d431fb --- /dev/null +++ b/gnmi_cli_converter/registry.py @@ -0,0 +1,128 @@ +""" +Command Registry + +Provides @register decorator and command lookup functionality. +""" + +from typing import List, Callable, Any, Dict +from dataclasses import dataclass +from .exceptions import PathNotFoundError + +# Global registry +_REGISTRY: Dict[str, 'CommandEntry'] = {} + + +@dataclass +class CommandEntry: + """Command registry entry""" + path_pattern: List[str] # Registration pattern: ["interfaces", "errors", "*"] + render_func: Callable[[Any, List[str], Dict[str, Any]], str] # Render function (unified signature with options) + command: str # "show interfaces errors " + + +def register(path_pattern: List[str]): + """ + Decorator: Register render function. + + All render functions use a unified signature: + def render_func(json_data: Any, path_elems: List[str], options: Dict[str, Any]) -> str + + Usage: + # Command without parameters (path_elems and options can be ignored) + @register(["buffer_pool", "watermark"]) + def render_buffer_pool_watermark(json_data, path_elems, options): + ... + + # Command with parameters (use * to indicate parameter position) + @register(["interfaces", "errors", "*"]) + def render_interfaces_errors(json_data, path_elems, options): + interface = path_elems[2] # Extract parameter + is_printall = options.get("printall", False) + ... + + Args: + path_pattern: Path pattern, use "*" to indicate parameter position + """ + def decorator(func: Callable[[Any, List[str]], str]): + key = "/".join(path_pattern) + + # Generate command description + cmd_parts = [] + for elem in path_pattern: + if elem == "*": + cmd_parts.append("") + else: + cmd_parts.append(elem) + + _REGISTRY[key] = CommandEntry( + path_pattern=path_pattern, + render_func=func, + command=f"show {' '.join(cmd_parts)}" + ) + return func + return decorator + + +def get_renderer(path_elems: List[str]) -> Callable[[Any, List[str], Dict[str, Any]], str]: + """ + Find render function. + + Args: + path_elems: Full path, e.g. ["interfaces", "errors", "Ethernet0"] + + Returns: + Render function with signature: (json_data, path_elems, options) -> str + + Raises: + PathNotFoundError: If path is not registered + + Matching strategy: + 1. Exact match: interfaces/errors/Ethernet0 + 2. Wildcard match: interfaces/errors/* + """ + # 1. Try exact match + exact_key = "/".join(path_elems) + if exact_key in _REGISTRY: + return _REGISTRY[exact_key].render_func + + # 2. Try wildcard match (replace last element with *) + if len(path_elems) > 0: + wildcard_key = "/".join(path_elems[:-1] + ["*"]) + if wildcard_key in _REGISTRY: + return _REGISTRY[wildcard_key].render_func + + raise PathNotFoundError(path_elems) + + +def is_supported(path_elems: List[str]) -> bool: + """ + Check if the path is supported. + + Args: + path_elems: Path element array + + Returns: + True if supported, False otherwise + """ + try: + get_renderer(path_elems) + return True + except PathNotFoundError: + return False + + +def list_commands() -> List[CommandEntry]: + """ + List all registered commands. + + Returns: + List of command entries + """ + return list(_REGISTRY.values()) + + +def clear_registry(): + """ + Clear the registry (for testing only). + """ + _REGISTRY.clear() diff --git a/gnmi_cli_converter/utils.py b/gnmi_cli_converter/utils.py new file mode 100644 index 000000000..3c66418cf --- /dev/null +++ b/gnmi_cli_converter/utils.py @@ -0,0 +1,94 @@ +""" +Render Helper Functions + +Provides common rendering pattern helper functions to avoid duplicate code. +""" + +from typing import List, Tuple, Dict, Any, Optional + + +def passthrough(json_data: dict, key: str = "output") -> str: + """ + Passthrough render - directly return string value. + + Applicable for simple commands like show clock, show uptime, etc. + + Args: + json_data: JSON data + key: Key name to extract, default "output" + + Returns: + String value + + Example: + >>> passthrough({"output": "Thu Jan 23 10:30:00 UTC 2026"}) + 'Thu Jan 23 10:30:00 UTC 2026' + """ + return json_data.get(key, "") + + +def tabulate_dict( + rows: List[Tuple], + headers: List[str], + title: Optional[str] = None, + tablefmt: str = "simple", + stralign: str = "left" +) -> str: + """ + Table render - using tabulate library. + + Applicable for most table output commands. + + Args: + rows: Table row data (list of tuples) + headers: Header list + title: Optional title (displayed above table) + tablefmt: tabulate table format, default "simple" + stralign: String alignment, default "left" + + Returns: + Formatted table string + + Example: + >>> rows = [("pool1", "12345"), ("pool2", "67890")] + >>> headers = ["Pool", "Bytes"] + >>> print(tabulate_dict(rows, headers, title="Watermark:")) + Watermark: + Pool Bytes + ------ ----- + pool1 12345 + pool2 67890 + """ + from tabulate import tabulate + + output = tabulate(rows, headers=headers, tablefmt=tablefmt, stralign=stralign) + if title: + output = f"{title}\n{output}" + return output + + +def key_value_pairs(json_data: Dict[str, Any], mapping: Dict[str, str]) -> str: + """ + Key-value pair render. + + Applicable for key-value output like show mmu, show processes, etc. + + Args: + json_data: JSON data + mapping: Field mapping (JSON key -> display name) + + Returns: + Key-value pair formatted string + + Example: + >>> data = {"mmu_size": "12345678", "cell_size": "256"} + >>> mapping = {"mmu_size": "MMU Size", "cell_size": "Cell Size"} + >>> print(key_value_pairs(data, mapping)) + MMU Size: 12345678 + Cell Size: 256 + """ + lines = [] + for json_key, display_name in mapping.items(): + value = json_data.get(json_key, "N/A") + lines.append(f"{display_name}: {value}") + return "\n".join(lines) diff --git a/tests/gnmi_cli_converter/__init__.py b/tests/gnmi_cli_converter/__init__.py new file mode 100644 index 000000000..cd405948a --- /dev/null +++ b/tests/gnmi_cli_converter/__init__.py @@ -0,0 +1,3 @@ +""" +GNMI CLI Converter test module initialization file +""" diff --git a/tests/gnmi_cli_converter/common.py b/tests/gnmi_cli_converter/common.py new file mode 100644 index 000000000..93c1c6deb --- /dev/null +++ b/tests/gnmi_cli_converter/common.py @@ -0,0 +1,39 @@ +""" +Common test helpers for gnmi_cli_converter tests. + +Provides utilities for loading test data from files. +""" + +import json +from pathlib import Path + +# Test data directory +TESTDATA_DIR = Path(__file__).parent / "testdata" +INPUTS_DIR = TESTDATA_DIR / "inputs" +EXPECTED_DIR = TESTDATA_DIR / "expected" + + +def load_json(filename: str) -> dict: + """Load JSON test data from inputs directory. + + Args: + filename: Name of the JSON file (e.g., "buffer_pool_watermark.json") + + Returns: + Parsed JSON data as dict + """ + with open(INPUTS_DIR / filename, 'r') as f: + return json.load(f) + + +def load_expected(filename: str) -> str: + """Load expected output from expected directory. + + Args: + filename: Name of the expected output file (e.g., "buffer_pool_watermark.txt") + + Returns: + Expected output string (trailing newline stripped) + """ + with open(EXPECTED_DIR / filename, 'r') as f: + return f.read().rstrip('\n') diff --git a/tests/gnmi_cli_converter/test_buffer_pool.py b/tests/gnmi_cli_converter/test_buffer_pool.py new file mode 100644 index 000000000..6286462a8 --- /dev/null +++ b/tests/gnmi_cli_converter/test_buffer_pool.py @@ -0,0 +1,157 @@ +""" +Buffer Pool Command Tests +""" + +import pytest +import sys +import os +from unittest.mock import MagicMock + +# Add gnmi_cli_converter to sys.path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) + +# Add current test directory to sys.path for common import +sys.path.insert(0, os.path.dirname(__file__)) + +# Mock sonic_py_common.logger before importing gnmi_cli_converter +mock_logger = MagicMock() +sys.modules['sonic_py_common'] = MagicMock() +sys.modules['sonic_py_common.logger'] = MagicMock() +sys.modules['sonic_py_common.logger'].Logger = MagicMock(return_value=mock_logger) + +from gnmi_cli_converter import CLIConverter +from common import load_json, load_expected + + +@pytest.fixture +def converter(): + """Fixture to create CLIConverter instance.""" + return CLIConverter() + + +@pytest.fixture +def buffer_pool_watermark_json(): + """Test JSON data for buffer_pool watermark command.""" + return load_json("buffer_pool_watermark.json") + + +@pytest.fixture +def buffer_pool_persistent_watermark_json(): + """Test JSON data for buffer_pool persistent-watermark command.""" + return load_json("buffer_pool_persistent_watermark.json") + + +class TestBufferPoolWatermark: + """Tests for show buffer_pool watermark command""" + + def test_output_matches_expected(self, converter, buffer_pool_watermark_json): + """Test output matches expected file exactly""" + output = converter.convert( + buffer_pool_watermark_json, + path_elems=["buffer_pool", "watermark"] + ) + expected = load_expected("buffer_pool_watermark.txt") + assert output == expected + + def test_natural_sorting(self, converter): + """Test natural sorting""" + json_data = { + "pool10": {"Bytes": "100"}, + "pool2": {"Bytes": "200"}, + "pool1": {"Bytes": "300"} + } + + output = converter.convert( + json_data, + path_elems=["buffer_pool", "watermark"] + ) + + # Check all pools are in output + assert "pool1" in output + assert "pool2" in output + assert "pool10" in output + + # Check natural sorting (pool1, pool2, pool10) + pos1 = output.find("pool1") + pos2 = output.find("pool2") + pos10 = output.find("pool10") + + # pool1 should be before pool2 (ignoring pool10) + assert pos1 < pos2 + # pool2 should be before pool10 + assert pos2 < pos10 + + def test_empty_data(self, converter): + """Test empty data handling""" + output = converter.convert( + {}, + path_elems=["buffer_pool", "watermark"] + ) + + assert "Shared pool maximum occupancy:" in output + assert "No data available" in output + + def test_missing_bytes_field(self, converter): + """Test handling of missing Bytes field""" + json_data = { + "pool_without_bytes": {"other_field": "value"} + } + + output = converter.convert( + json_data, + path_elems=["buffer_pool", "watermark"] + ) + + assert "pool_without_bytes" in output + assert "N/A" in output + + +class TestBufferPoolPersistentWatermark: + """Tests for show buffer_pool persistent-watermark command""" + + def test_output_matches_expected(self, converter, buffer_pool_persistent_watermark_json): + """Test output matches expected file exactly""" + output = converter.convert( + buffer_pool_persistent_watermark_json, + path_elems=["buffer_pool", "persistent-watermark"] + ) + expected = load_expected("buffer_pool_persistent_watermark.txt") + assert output == expected + + def test_empty_data(self, converter): + """Test empty data handling""" + output = converter.convert( + {}, + path_elems=["buffer_pool", "persistent-watermark"] + ) + + assert "Shared pool maximum occupancy:" in output + assert "No data available" in output + + +class TestBufferPoolIntegration: + """Buffer Pool command integration tests""" + + def test_both_commands_registered(self, converter): + """Test both commands are registered""" + assert converter.is_supported(path_elems=["buffer_pool", "watermark"]) is True + assert converter.is_supported(path_elems=["buffer_pool", "persistent-watermark"]) is True + + def test_commands_in_list(self, converter): + """Test commands appear in list""" + commands = converter.list_commands() + command_paths = [cmd.path_elems for cmd in commands] + + assert ["buffer_pool", "watermark"] in command_paths + assert ["buffer_pool", "persistent-watermark"] in command_paths + + def test_command_descriptions(self, converter): + """Test command descriptions are correct""" + commands = converter.list_commands() + + watermark_cmd = next( + (cmd for cmd in commands if cmd.path_elems == ["buffer_pool", "watermark"]), + None + ) + assert watermark_cmd is not None + assert "show buffer_pool watermark" in watermark_cmd.command diff --git a/tests/gnmi_cli_converter/test_converter.py b/tests/gnmi_cli_converter/test_converter.py new file mode 100644 index 000000000..cf1470dde --- /dev/null +++ b/tests/gnmi_cli_converter/test_converter.py @@ -0,0 +1,260 @@ +""" +CLIConverter Core Functionality Tests +""" + +import pytest +import sys +import os +from unittest.mock import MagicMock + +# Add gnmi_cli_converter to sys.path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) + +# Mock sonic_py_common.logger before importing gnmi_cli_converter +mock_logger = MagicMock() +sys.modules['sonic_py_common'] = MagicMock() +sys.modules['sonic_py_common.logger'] = MagicMock() +sys.modules['sonic_py_common.logger'].Logger = MagicMock(return_value=mock_logger) + +from gnmi_cli_converter import CLIConverter, PathNotFoundError, RenderError + + +@pytest.fixture +def converter(): + """Fixture to create CLIConverter instance.""" + return CLIConverter() + + +@pytest.fixture +def buffer_pool_watermark_json(): + """Test JSON data for buffer_pool watermark command.""" + return { + "egress_lossless_pool": {"Bytes": "12345"}, + "egress_lossy_pool": {"Bytes": "67890"}, + "ingress_lossless_pool": {"Bytes": "24680"} + } + + +class TestCLIConverter: + """Tests for CLIConverter class""" + + def test_convert_basic(self, converter, buffer_pool_watermark_json): + """Test basic convert functionality""" + output = converter.convert( + buffer_pool_watermark_json, + path_elems=["buffer_pool", "watermark"] + ) + + # Check output contains title + assert "Shared pool maximum occupancy:" in output + + # Check output contains all pool names + assert "egress_lossless_pool" in output + assert "egress_lossy_pool" in output + assert "ingress_lossless_pool" in output + + # Check output contains all values + assert "12345" in output + assert "67890" in output + assert "24680" in output + + def test_convert_with_xpath(self, converter, buffer_pool_watermark_json): + """Test convert using xpath""" + output = converter.convert( + buffer_pool_watermark_json, + xpath="/buffer_pool/watermark" + ) + + assert "Shared pool maximum occupancy:" in output + assert "egress_lossless_pool" in output + + def test_convert_with_xpath_no_leading_slash(self, converter, buffer_pool_watermark_json): + """Test convert using xpath without leading slash""" + output = converter.convert( + buffer_pool_watermark_json, + xpath="buffer_pool/watermark" + ) + + assert "Shared pool maximum occupancy:" in output + + def test_convert_with_command(self, converter, buffer_pool_watermark_json): + """Test convert using show command""" + output = converter.convert( + buffer_pool_watermark_json, + command="show buffer_pool watermark" + ) + + assert "Shared pool maximum occupancy:" in output + assert "egress_lossless_pool" in output + + def test_convert_with_command_no_show_prefix(self, converter, buffer_pool_watermark_json): + """Test convert using command without show prefix""" + output = converter.convert( + buffer_pool_watermark_json, + command="buffer_pool watermark" + ) + + assert "Shared pool maximum occupancy:" in output + + def test_convert_multiple_specifiers_error(self, converter, buffer_pool_watermark_json): + """Test error when multiple path specifiers provided""" + with pytest.raises(ValueError) as exc_info: + converter.convert( + buffer_pool_watermark_json, + path_elems=["buffer_pool", "watermark"], + xpath="/buffer_pool/watermark" + ) + + assert "exactly one" in str(exc_info.value) + + def test_convert_no_specifier_error(self, converter, buffer_pool_watermark_json): + """Test error when no path specifier provided""" + with pytest.raises(ValueError) as exc_info: + converter.convert(buffer_pool_watermark_json) + + assert "exactly one" in str(exc_info.value) + + def test_is_supported_with_xpath(self, converter): + """Test is_supported with xpath""" + assert converter.is_supported(xpath="/buffer_pool/watermark") is True + assert converter.is_supported(xpath="/unknown/path") is False + + def test_is_supported_with_command(self, converter): + """Test is_supported with command""" + assert converter.is_supported(command="show buffer_pool watermark") is True + assert converter.is_supported(command="show unknown path") is False + + def test_xpath_to_path_elems_basic(self, converter): + """Test _xpath_to_path_elems basic parsing""" + path_elems, options = CLIConverter._xpath_to_path_elems("/buffer_pool/watermark") + assert path_elems == ["buffer_pool", "watermark"] + assert options == {} + + def test_xpath_to_path_elems_with_options(self, converter): + """Test _xpath_to_path_elems with options""" + path_elems, options = CLIConverter._xpath_to_path_elems( + "/interfaces/counters[printall=true][interfaces=Ethernet0]" + ) + assert path_elems == ["interfaces", "counters"] + assert options == {"printall": True, "interfaces": "Ethernet0"} + + def test_xpath_to_path_elems_type_conversion(self, converter): + """Test _xpath_to_path_elems converts types correctly""" + path_elems, options = CLIConverter._xpath_to_path_elems( + "/queue/counters[nonzero=true][period=10][verbose=false]" + ) + assert options["nonzero"] is True + assert options["period"] == 10 + assert options["verbose"] is False + + def test_command_to_path_elems_basic(self, converter): + """Test _command_to_path_elems basic parsing""" + path_elems, options = CLIConverter._command_to_path_elems("show buffer_pool watermark") + assert path_elems == ["buffer_pool", "watermark"] + assert options == {} + + def test_command_to_path_elems_with_options(self, converter): + """Test _command_to_path_elems with options""" + path_elems, options = CLIConverter._command_to_path_elems( + "show interfaces counters --printall --interfaces=Ethernet0" + ) + assert path_elems == ["interfaces", "counters"] + assert options == {"printall": True, "interfaces": "Ethernet0"} + + def test_command_to_path_elems_type_conversion(self, converter): + """Test _command_to_path_elems converts types correctly""" + path_elems, options = CLIConverter._command_to_path_elems( + "show queue counters --nonzero --period=10 --verbose=false" + ) + assert options["nonzero"] is True + assert options["period"] == 10 + assert options["verbose"] is False + + def test_convert_xpath_with_options(self, converter, buffer_pool_watermark_json): + """Test convert using xpath with embedded options""" + # Note: buffer_pool doesn't use options, but we verify options are parsed + output = converter.convert( + buffer_pool_watermark_json, + xpath="/buffer_pool/watermark[verbose=true]" + ) + assert "Shared pool maximum occupancy:" in output + + def test_convert_command_with_options(self, converter, buffer_pool_watermark_json): + """Test convert using command with embedded options""" + output = converter.convert( + buffer_pool_watermark_json, + command="show buffer_pool watermark --verbose" + ) + assert "Shared pool maximum occupancy:" in output + + def test_convert_not_found(self, converter): + """Test unregistered path throws PathNotFoundError""" + with pytest.raises(PathNotFoundError) as exc_info: + converter.convert({}, path_elems=["unknown", "path"]) + + assert "unknown" in str(exc_info.value) + + def test_is_supported_registered(self, converter): + """Test registered path returns True""" + assert converter.is_supported(path_elems=["buffer_pool", "watermark"]) is True + assert converter.is_supported(path_elems=["buffer_pool", "persistent-watermark"]) is True + + def test_is_supported_not_registered(self, converter): + """Test unregistered path returns False""" + assert converter.is_supported(path_elems=["unknown", "path"]) is False + assert converter.is_supported(path_elems=[]) is False + + def test_list_commands(self, converter): + """Test listing all supported commands""" + commands = converter.list_commands() + + # Check returns a list + assert isinstance(commands, list) + + # Check contains at least buffer_pool commands + command_strs = [cmd.command for cmd in commands] + assert any("buffer_pool" in cmd for cmd in command_strs) + + # Check each command has path_elems and command attributes + for cmd in commands: + assert hasattr(cmd, 'path_elems') + assert hasattr(cmd, 'command') + assert isinstance(cmd.path_elems, list) + assert isinstance(cmd.command, str) + + +class TestRegistry: + """Registry functionality tests""" + + def test_register_decorator(self): + """Test @register decorator works correctly""" + from gnmi_cli_converter.registry import _REGISTRY + + # Check buffer_pool/watermark is registered + assert "buffer_pool/watermark" in _REGISTRY + assert "buffer_pool/persistent-watermark" in _REGISTRY + + def test_wildcard_matching(self): + """Test wildcard matching functionality""" + from gnmi_cli_converter.registry import register, get_renderer, clear_registry, _REGISTRY + + # Save original registry + original_registry = _REGISTRY.copy() + + try: + # Register a command with wildcard + @register(["test", "wildcard", "*"]) + def render_test_wildcard(json_data, path_elems, options): + return f"param: {path_elems[2]}" + + # Test wildcard matching + renderer = get_renderer(["test", "wildcard", "Ethernet0"]) + assert renderer({"data": "test"}, ["test", "wildcard", "Ethernet0"], {}) == "param: Ethernet0" + + renderer = get_renderer(["test", "wildcard", "Ethernet4"]) + assert renderer({"data": "test"}, ["test", "wildcard", "Ethernet4"], {}) == "param: Ethernet4" + + finally: + # Restore original registry + _REGISTRY.clear() + _REGISTRY.update(original_registry) diff --git a/tests/gnmi_cli_converter/test_exceptions.py b/tests/gnmi_cli_converter/test_exceptions.py new file mode 100644 index 000000000..ca5fd9fbe --- /dev/null +++ b/tests/gnmi_cli_converter/test_exceptions.py @@ -0,0 +1,45 @@ +""" +Exception Classes Tests +""" + +import pytest +import sys +import os +from unittest.mock import MagicMock + +# Add gnmi_cli_converter to sys.path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) + +# Mock sonic_py_common.logger before importing gnmi_cli_converter +mock_logger = MagicMock() +sys.modules['sonic_py_common'] = MagicMock() +sys.modules['sonic_py_common.logger'] = MagicMock() +sys.modules['sonic_py_common.logger'].Logger = MagicMock(return_value=mock_logger) + + +class TestExceptions: + """Exception classes tests""" + + def test_path_not_found_error(self): + """Test PathNotFoundError""" + from gnmi_cli_converter.exceptions import PathNotFoundError + + error = PathNotFoundError(["test", "path"]) + assert error.path_elems == ["test", "path"] + assert "test" in str(error) + assert "path" in str(error) + + def test_exception_hierarchy(self): + """Test exception inheritance hierarchy""" + from gnmi_cli_converter.exceptions import ( + CLIConverterError, + PathNotFoundError, + InvalidJSONError, + RenderError, + MissingFieldError + ) + + assert issubclass(PathNotFoundError, CLIConverterError) + assert issubclass(InvalidJSONError, CLIConverterError) + assert issubclass(RenderError, CLIConverterError) + assert issubclass(MissingFieldError, CLIConverterError) diff --git a/tests/gnmi_cli_converter/test_uptime.py b/tests/gnmi_cli_converter/test_uptime.py new file mode 100644 index 000000000..ae53df16d --- /dev/null +++ b/tests/gnmi_cli_converter/test_uptime.py @@ -0,0 +1,76 @@ +""" +Uptime Command Tests +""" + +import pytest +import sys +import os +from unittest.mock import MagicMock + +# Add gnmi_cli_converter to sys.path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) + +# Add current test directory to sys.path for common import +sys.path.insert(0, os.path.dirname(__file__)) + +# Mock sonic_py_common.logger before importing gnmi_cli_converter +mock_logger = MagicMock() +sys.modules['sonic_py_common'] = MagicMock() +sys.modules['sonic_py_common.logger'] = MagicMock() +sys.modules['sonic_py_common.logger'].Logger = MagicMock(return_value=mock_logger) + +from gnmi_cli_converter import CLIConverter + + +@pytest.fixture +def converter(): + """Fixture to create CLIConverter instance.""" + return CLIConverter() + + +class TestUptime: + """Tests for show uptime command""" + + def test_basic_output(self, converter): + """Test basic uptime output""" + json_data = {"uptime": "up 3 weeks, 4 days, 10 hours, 15 minutes"} + + output = converter.convert(json_data, path_elems=["uptime"]) + + assert output == "up 3 weeks, 4 days, 10 hours, 15 minutes" + + def test_detailed_uptime(self, converter): + """Test detailed uptime format""" + json_data = {"uptime": "07:42:51 up 16 days, 14:51, 2 users, load average: 0.00, 0.00, 0.00"} + + output = converter.convert(json_data, path_elems=["uptime"]) + + assert "07:42:51 up 16 days" in output + assert "load average" in output + + def test_empty_uptime(self, converter): + """Test empty uptime""" + json_data = {"uptime": ""} + + output = converter.convert(json_data, path_elems=["uptime"]) + + assert output == "" + + def test_missing_key(self, converter): + """Test missing uptime key""" + json_data = {} + + output = converter.convert(json_data, path_elems=["uptime"]) + + assert output == "" + + def test_is_supported(self, converter): + """Test uptime command is registered""" + assert converter.is_supported(path_elems=["uptime"]) is True + + def test_command_in_list(self, converter): + """Test uptime appears in command list""" + commands = converter.list_commands() + command_paths = [cmd.path_elems for cmd in commands] + + assert ["uptime"] in command_paths diff --git a/tests/gnmi_cli_converter/test_utils.py b/tests/gnmi_cli_converter/test_utils.py new file mode 100644 index 000000000..83b913c46 --- /dev/null +++ b/tests/gnmi_cli_converter/test_utils.py @@ -0,0 +1,79 @@ +""" +Utils Module Tests +""" + +import pytest +import sys +import os +from unittest.mock import MagicMock + +# Add gnmi_cli_converter to sys.path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) + +# Mock sonic_py_common.logger before importing gnmi_cli_converter +mock_logger = MagicMock() +sys.modules['sonic_py_common'] = MagicMock() +sys.modules['sonic_py_common.logger'] = MagicMock() +sys.modules['sonic_py_common.logger'].Logger = MagicMock(return_value=mock_logger) + + +class TestUtils: + """Utils functions tests""" + + def test_passthrough(self): + """Test passthrough function""" + from gnmi_cli_converter.utils import passthrough + + json_data = {"output": "Hello, World!"} + assert passthrough(json_data) == "Hello, World!" + + json_data = {"output": ""} + assert passthrough(json_data) == "" + + json_data = {} + assert passthrough(json_data) == "" + + def test_passthrough_custom_key(self): + """Test passthrough function with custom key""" + from gnmi_cli_converter.utils import passthrough + + json_data = {"result": "Custom value"} + assert passthrough(json_data, key="result") == "Custom value" + + def test_tabulate_dict(self): + """Test tabulate_dict function""" + from gnmi_cli_converter.utils import tabulate_dict + + rows = [("pool1", "100"), ("pool2", "200")] + headers = ["Pool", "Bytes"] + + output = tabulate_dict(rows, headers) + + assert "Pool" in output + assert "Bytes" in output + assert "pool1" in output + assert "100" in output + + def test_tabulate_dict_with_title(self): + """Test tabulate_dict function with title""" + from gnmi_cli_converter.utils import tabulate_dict + + rows = [("pool1", "100")] + headers = ["Pool", "Bytes"] + + output = tabulate_dict(rows, headers, title="My Title:") + + assert "My Title:" in output + assert output.startswith("My Title:") + + def test_key_value_pairs(self): + """Test key_value_pairs function""" + from gnmi_cli_converter.utils import key_value_pairs + + json_data = {"mmu_size": "12345678", "cell_size": "256"} + mapping = {"mmu_size": "MMU Size", "cell_size": "Cell Size"} + + output = key_value_pairs(json_data, mapping) + + assert "MMU Size: 12345678" in output + assert "Cell Size: 256" in output diff --git a/tests/gnmi_cli_converter/testdata/expected/buffer_pool_persistent_watermark.txt b/tests/gnmi_cli_converter/testdata/expected/buffer_pool_persistent_watermark.txt new file mode 100644 index 000000000..66d945e34 --- /dev/null +++ b/tests/gnmi_cli_converter/testdata/expected/buffer_pool_persistent_watermark.txt @@ -0,0 +1,5 @@ +Shared pool maximum occupancy: + Pool Bytes +--------------------- ------- + egress_lossless_pool 11111 +ingress_lossless_pool 22222 diff --git a/tests/gnmi_cli_converter/testdata/expected/buffer_pool_watermark.txt b/tests/gnmi_cli_converter/testdata/expected/buffer_pool_watermark.txt new file mode 100644 index 000000000..36fb8fa43 --- /dev/null +++ b/tests/gnmi_cli_converter/testdata/expected/buffer_pool_watermark.txt @@ -0,0 +1,6 @@ +Shared pool maximum occupancy: + Pool Bytes +--------------------- ------- + egress_lossless_pool 12345 + egress_lossy_pool 67890 +ingress_lossless_pool 24680 diff --git a/tests/gnmi_cli_converter/testdata/inputs/buffer_pool_persistent_watermark.json b/tests/gnmi_cli_converter/testdata/inputs/buffer_pool_persistent_watermark.json new file mode 100644 index 000000000..ea02c4613 --- /dev/null +++ b/tests/gnmi_cli_converter/testdata/inputs/buffer_pool_persistent_watermark.json @@ -0,0 +1,4 @@ +{ + "egress_lossless_pool": {"Bytes": "11111"}, + "ingress_lossless_pool": {"Bytes": "22222"} +} diff --git a/tests/gnmi_cli_converter/testdata/inputs/buffer_pool_watermark.json b/tests/gnmi_cli_converter/testdata/inputs/buffer_pool_watermark.json new file mode 100644 index 000000000..d05a29a02 --- /dev/null +++ b/tests/gnmi_cli_converter/testdata/inputs/buffer_pool_watermark.json @@ -0,0 +1,5 @@ +{ + "egress_lossless_pool": {"Bytes": "12345"}, + "egress_lossy_pool": {"Bytes": "67890"}, + "ingress_lossless_pool": {"Bytes": "24680"} +}