From 666a7dbc2ea33659425d2f695835c4a0a1fef723 Mon Sep 17 00:00:00 2001 From: guoyuan Date: Tue, 27 Jan 2026 02:03:03 +0000 Subject: [PATCH 1/3] add framework --- gnmi_cli_converter/__init__.py | 49 +++++ gnmi_cli_converter/commands/__init__.py | 16 ++ gnmi_cli_converter/commands/buffer_pool.py | 80 ++++++++ gnmi_cli_converter/commands/uptime.py | 28 +++ gnmi_cli_converter/converter.py | 105 ++++++++++ gnmi_cli_converter/exceptions.py | 30 +++ gnmi_cli_converter/registry.py | 127 ++++++++++++ gnmi_cli_converter/utils.py | 94 +++++++++ tests/gnmi_cli_converter/__init__.py | 3 + tests/gnmi_cli_converter/common.py | 39 ++++ tests/gnmi_cli_converter/test_buffer_pool.py | 191 ++++++++++++++++++ tests/gnmi_cli_converter/test_converter.py | 131 ++++++++++++ tests/gnmi_cli_converter/test_exceptions.py | 45 +++++ tests/gnmi_cli_converter/test_uptime.py | 76 +++++++ tests/gnmi_cli_converter/test_utils.py | 79 ++++++++ .../buffer_pool_persistent_watermark.txt | 5 + .../expected/buffer_pool_watermark.txt | 6 + .../buffer_pool_persistent_watermark.json | 4 + .../inputs/buffer_pool_watermark.json | 5 + 19 files changed, 1113 insertions(+) create mode 100644 gnmi_cli_converter/__init__.py create mode 100644 gnmi_cli_converter/commands/__init__.py create mode 100644 gnmi_cli_converter/commands/buffer_pool.py create mode 100644 gnmi_cli_converter/commands/uptime.py create mode 100644 gnmi_cli_converter/converter.py create mode 100644 gnmi_cli_converter/exceptions.py create mode 100644 gnmi_cli_converter/registry.py create mode 100644 gnmi_cli_converter/utils.py create mode 100644 tests/gnmi_cli_converter/__init__.py create mode 100644 tests/gnmi_cli_converter/common.py create mode 100644 tests/gnmi_cli_converter/test_buffer_pool.py create mode 100644 tests/gnmi_cli_converter/test_converter.py create mode 100644 tests/gnmi_cli_converter/test_exceptions.py create mode 100644 tests/gnmi_cli_converter/test_uptime.py create mode 100644 tests/gnmi_cli_converter/test_utils.py create mode 100644 tests/gnmi_cli_converter/testdata/expected/buffer_pool_persistent_watermark.txt create mode 100644 tests/gnmi_cli_converter/testdata/expected/buffer_pool_watermark.txt create mode 100644 tests/gnmi_cli_converter/testdata/inputs/buffer_pool_persistent_watermark.json create mode 100644 tests/gnmi_cli_converter/testdata/inputs/buffer_pool_watermark.json 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..1817d587f --- /dev/null +++ b/gnmi_cli_converter/commands/buffer_pool.py @@ -0,0 +1,80 @@ +""" +Buffer Pool Related Command Render Functions + +Contains: +- show buffer_pool watermark +- show buffer_pool persistent-watermark +""" + +from typing import Any, List +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]) -> str: + """ + show buffer_pool watermark + + Args: + json_data: JSON data returned by GNMI + path_elems: Path element array (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]) -> 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) + """ + 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..767b166f7 --- /dev/null +++ b/gnmi_cli_converter/commands/uptime.py @@ -0,0 +1,28 @@ +""" +Uptime Command Render Function + +Contains: +- show uptime +""" + +from typing import Any, List +from ..utils import passthrough +from ..registry import register + + +@register(["uptime"]) +def render_uptime(json_data: Any, path_elems: List[str]) -> str: + """ + show uptime + + Args: + json_data: JSON data returned by GNMI + path_elems: Path element array (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..2ac17c68b --- /dev/null +++ b/gnmi_cli_converter/converter.py @@ -0,0 +1,105 @@ +""" +CLI Converter Main Class + +Provides interface for converting GNMI JSON to CLI format. +""" + +from typing import List, Any +from dataclasses import dataclass +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] # ["buffer_pool", "watermark"] or ["interfaces", "errors", "*"] + command: str # "show buffer_pool watermark" + + +class CLIConverter: + """ + Converter for GNMI JSON to CLI format. + + Example: + >>> converter = CLIConverter() + >>> json_data = {"egress_lossless_pool": {"Bytes": "12345"}} + >>> output = converter.convert(["buffer_pool", "watermark"], json_data) + >>> print(output) + Shared pool maximum occupancy: + Pool Bytes + --------------------- ----- + egress_lossless_pool 12345 + """ + + def convert(self, path_elems: List[str], json_data: Any) -> str: + """ + Convert GNMI JSON to CLI format text. + + Args: + path_elems: Path element array containing the full path (with parameter values) + - Without parameters: ["buffer_pool", "watermark"] + - With parameters: ["interfaces", "errors", "Ethernet0"] + (without "SHOW" prefix) + json_data: JSON data returned by GNMI (already parsed to Python object) + + Returns: + CLI format text output + + Raises: + PathNotFoundError: Path not registered + RenderError: Rendering failed + """ + 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 + log.log_debug(f"Rendering JSON data for path: {path_elems}") + output = render_func(json_data, path_elems) + + 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: List[str]) -> bool: + """ + Check if the path is supported. + + Args: + path_elems: Path element array + + Returns: + True if supported, False otherwise + """ + 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..b88c97ff4 --- /dev/null +++ b/gnmi_cli_converter/registry.py @@ -0,0 +1,127 @@ +""" +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]], str] # Render function (unified signature) + 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]) -> str + + Usage: + # Command without parameters (path_elems can be ignored) + @register(["buffer_pool", "watermark"]) + def render_buffer_pool_watermark(json_data, path_elems): + ... + + # Command with parameters (use * to indicate parameter position) + @register(["interfaces", "errors", "*"]) + def render_interfaces_errors(json_data, path_elems): + interface = path_elems[2] # Extract parameter + ... + + 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]], str]: + """ + Find render function. + + Args: + path_elems: Full path, e.g. ["interfaces", "errors", "Ethernet0"] + + Returns: + Render function (unified signature) + + 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..10c109b26 --- /dev/null +++ b/tests/gnmi_cli_converter/__init__.py @@ -0,0 +1,3 @@ +""" +GNMI CLI Converter 测试模块初始化文件 +""" 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..2ef3b564f --- /dev/null +++ b/tests/gnmi_cli_converter/test_buffer_pool.py @@ -0,0 +1,191 @@ +""" +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_basic_output(self, converter, buffer_pool_watermark_json): + """Test basic output format""" + output = converter.convert( + ["buffer_pool", "watermark"], + buffer_pool_watermark_json + ) + + # Check title + assert "Shared pool maximum occupancy:" in output + + # Check headers + assert "Pool" in output + assert "Bytes" in output + + # Check data + assert "egress_lossless_pool" in output + assert "12345" in output + + def test_output_matches_expected(self, converter, buffer_pool_watermark_json): + """Test output matches expected file exactly""" + output = converter.convert( + ["buffer_pool", "watermark"], + buffer_pool_watermark_json + ) + 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( + ["buffer_pool", "watermark"], + json_data + ) + + # 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( + ["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( + ["buffer_pool", "watermark"], + json_data + ) + + assert "pool_without_bytes" in output + assert "N/A" in output + + +class TestBufferPoolPersistentWatermark: + """Tests for show buffer_pool persistent-watermark command""" + + def test_basic_output(self, converter, buffer_pool_persistent_watermark_json): + """Test basic output format""" + output = converter.convert( + ["buffer_pool", "persistent-watermark"], + buffer_pool_persistent_watermark_json + ) + + # Check title + assert "Shared pool maximum occupancy:" in output + + # Check data + assert "egress_lossless_pool" in output + assert "11111" in output + assert "ingress_lossless_pool" in output + assert "22222" in output + + 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"], + buffer_pool_persistent_watermark_json + ) + expected = load_expected("buffer_pool_persistent_watermark.txt") + assert output == expected + + def test_empty_data(self, converter): + """Test empty data handling""" + output = converter.convert( + ["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(["buffer_pool", "watermark"]) is True + assert converter.is_supported(["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..3eab73b54 --- /dev/null +++ b/tests/gnmi_cli_converter/test_converter.py @@ -0,0 +1,131 @@ +""" +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"], + buffer_pool_watermark_json + ) + + # 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_not_found(self, converter): + """Test unregistered path throws PathNotFoundError""" + with pytest.raises(PathNotFoundError) as exc_info: + converter.convert(["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(["buffer_pool", "watermark"]) is True + assert converter.is_supported(["buffer_pool", "persistent-watermark"]) is True + + def test_is_supported_not_registered(self, converter): + """Test unregistered path returns False""" + assert converter.is_supported(["unknown", "path"]) is False + assert converter.is_supported([]) 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): + 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..0f0ab9931 --- /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(["uptime"], json_data) + + 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(["uptime"], json_data) + + 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(["uptime"], json_data) + + assert output == "" + + def test_missing_key(self, converter): + """Test missing uptime key""" + json_data = {} + + output = converter.convert(["uptime"], json_data) + + assert output == "" + + def test_is_supported(self, converter): + """Test uptime command is registered""" + assert converter.is_supported(["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"} +} From 16abc0607efb236220633016ee10201fe8cf285c Mon Sep 17 00:00:00 2001 From: guoyuan Date: Tue, 27 Jan 2026 02:19:23 +0000 Subject: [PATCH 2/3] Fix typo --- tests/gnmi_cli_converter/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/gnmi_cli_converter/__init__.py b/tests/gnmi_cli_converter/__init__.py index 10c109b26..cd405948a 100644 --- a/tests/gnmi_cli_converter/__init__.py +++ b/tests/gnmi_cli_converter/__init__.py @@ -1,3 +1,3 @@ """ -GNMI CLI Converter 测试模块初始化文件 +GNMI CLI Converter test module initialization file """ From 83969c28e56b669892c6f2c6acf6682ea338f8fd Mon Sep 17 00:00:00 2001 From: guoyuan Date: Wed, 28 Jan 2026 07:00:37 +0000 Subject: [PATCH 3/3] enrich the interface --- gnmi_cli_converter/commands/buffer_pool.py | 8 +- gnmi_cli_converter/commands/uptime.py | 5 +- gnmi_cli_converter/converter.py | 182 +++++++++++++++++-- gnmi_cli_converter/registry.py | 19 +- tests/gnmi_cli_converter/test_buffer_pool.py | 62 ++----- tests/gnmi_cli_converter/test_converter.py | 149 ++++++++++++++- tests/gnmi_cli_converter/test_uptime.py | 10 +- 7 files changed, 344 insertions(+), 91 deletions(-) diff --git a/gnmi_cli_converter/commands/buffer_pool.py b/gnmi_cli_converter/commands/buffer_pool.py index 1817d587f..91dca9005 100644 --- a/gnmi_cli_converter/commands/buffer_pool.py +++ b/gnmi_cli_converter/commands/buffer_pool.py @@ -6,20 +6,21 @@ - show buffer_pool persistent-watermark """ -from typing import Any, List +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]) -> str: +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: { @@ -54,7 +55,7 @@ def render_buffer_pool_watermark(json_data: Any, path_elems: List[str]) -> str: @register(["buffer_pool", "persistent-watermark"]) -def render_buffer_pool_persistent_watermark(json_data: Any, path_elems: List[str]) -> str: +def render_buffer_pool_persistent_watermark(json_data: Any, path_elems: List[str], options: Dict[str, Any]) -> str: """ show buffer_pool persistent-watermark @@ -63,6 +64,7 @@ def render_buffer_pool_persistent_watermark(json_data: Any, path_elems: List[str 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" diff --git a/gnmi_cli_converter/commands/uptime.py b/gnmi_cli_converter/commands/uptime.py index 767b166f7..003e521c5 100644 --- a/gnmi_cli_converter/commands/uptime.py +++ b/gnmi_cli_converter/commands/uptime.py @@ -5,19 +5,20 @@ - show uptime """ -from typing import Any, List +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]) -> str: +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"} diff --git a/gnmi_cli_converter/converter.py b/gnmi_cli_converter/converter.py index 2ac17c68b..d788e18eb 100644 --- a/gnmi_cli_converter/converter.py +++ b/gnmi_cli_converter/converter.py @@ -4,8 +4,9 @@ Provides interface for converting GNMI JSON to CLI format. """ -from typing import List, Any +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 @@ -19,8 +20,8 @@ @dataclass class CommandInfo: """Command information (for callers to query supported command list)""" - path_elems: List[str] # ["buffer_pool", "watermark"] or ["interfaces", "errors", "*"] - command: str # "show buffer_pool watermark" + path_elems: List[str] # Pattern with wildcards: ["interfaces", "errors", "*"] + command: str # "show interfaces errors " class CLIConverter: @@ -30,7 +31,10 @@ class CLIConverter: Example: >>> converter = CLIConverter() >>> json_data = {"egress_lossless_pool": {"Bytes": "12345"}} - >>> output = converter.convert(["buffer_pool", "watermark"], json_data) + >>> # 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 @@ -38,32 +42,164 @@ class CLIConverter: egress_lossless_pool 12345 """ - def convert(self, path_elems: List[str], json_data: Any) -> str: + @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: - path_elems: Path element array containing the full path (with parameter values) - - Without parameters: ["buffer_pool", "watermark"] - - With parameters: ["interfaces", "errors", "Ethernet0"] - (without "SHOW" prefix) 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 registered + 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 - log.log_debug(f"Rendering JSON data for path: {path_elems}") - output = render_func(json_data, 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 @@ -76,16 +212,34 @@ def convert(self, path_elems: List[str], json_data: Any) -> str: 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: List[str]) -> bool: + 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]: diff --git a/gnmi_cli_converter/registry.py b/gnmi_cli_converter/registry.py index b88c97ff4..1b0d431fb 100644 --- a/gnmi_cli_converter/registry.py +++ b/gnmi_cli_converter/registry.py @@ -15,9 +15,9 @@ @dataclass class CommandEntry: """Command registry entry""" - path_pattern: List[str] # Registration pattern: ["interfaces", "errors", "*"] - render_func: Callable[[Any, List[str]], str] # Render function (unified signature) - command: str # "show interfaces errors " + 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]): @@ -25,18 +25,19 @@ 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]) -> str + def render_func(json_data: Any, path_elems: List[str], options: Dict[str, Any]) -> str Usage: - # Command without parameters (path_elems can be ignored) + # Command without parameters (path_elems and options can be ignored) @register(["buffer_pool", "watermark"]) - def render_buffer_pool_watermark(json_data, path_elems): + 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): + def render_interfaces_errors(json_data, path_elems, options): interface = path_elems[2] # Extract parameter + is_printall = options.get("printall", False) ... Args: @@ -62,7 +63,7 @@ def decorator(func: Callable[[Any, List[str]], str]): return decorator -def get_renderer(path_elems: List[str]) -> Callable[[Any, List[str]], str]: +def get_renderer(path_elems: List[str]) -> Callable[[Any, List[str], Dict[str, Any]], str]: """ Find render function. @@ -70,7 +71,7 @@ def get_renderer(path_elems: List[str]) -> Callable[[Any, List[str]], str]: path_elems: Full path, e.g. ["interfaces", "errors", "Ethernet0"] Returns: - Render function (unified signature) + Render function with signature: (json_data, path_elems, options) -> str Raises: PathNotFoundError: If path is not registered diff --git a/tests/gnmi_cli_converter/test_buffer_pool.py b/tests/gnmi_cli_converter/test_buffer_pool.py index 2ef3b564f..6286462a8 100644 --- a/tests/gnmi_cli_converter/test_buffer_pool.py +++ b/tests/gnmi_cli_converter/test_buffer_pool.py @@ -44,29 +44,11 @@ def buffer_pool_persistent_watermark_json(): class TestBufferPoolWatermark: """Tests for show buffer_pool watermark command""" - def test_basic_output(self, converter, buffer_pool_watermark_json): - """Test basic output format""" - output = converter.convert( - ["buffer_pool", "watermark"], - buffer_pool_watermark_json - ) - - # Check title - assert "Shared pool maximum occupancy:" in output - - # Check headers - assert "Pool" in output - assert "Bytes" in output - - # Check data - assert "egress_lossless_pool" in output - assert "12345" in output - def test_output_matches_expected(self, converter, buffer_pool_watermark_json): """Test output matches expected file exactly""" output = converter.convert( - ["buffer_pool", "watermark"], - buffer_pool_watermark_json + buffer_pool_watermark_json, + path_elems=["buffer_pool", "watermark"] ) expected = load_expected("buffer_pool_watermark.txt") assert output == expected @@ -80,8 +62,8 @@ def test_natural_sorting(self, converter): } output = converter.convert( - ["buffer_pool", "watermark"], - json_data + json_data, + path_elems=["buffer_pool", "watermark"] ) # Check all pools are in output @@ -102,8 +84,8 @@ def test_natural_sorting(self, converter): def test_empty_data(self, converter): """Test empty data handling""" output = converter.convert( - ["buffer_pool", "watermark"], - {} + {}, + path_elems=["buffer_pool", "watermark"] ) assert "Shared pool maximum occupancy:" in output @@ -116,8 +98,8 @@ def test_missing_bytes_field(self, converter): } output = converter.convert( - ["buffer_pool", "watermark"], - json_data + json_data, + path_elems=["buffer_pool", "watermark"] ) assert "pool_without_bytes" in output @@ -127,27 +109,11 @@ def test_missing_bytes_field(self, converter): class TestBufferPoolPersistentWatermark: """Tests for show buffer_pool persistent-watermark command""" - def test_basic_output(self, converter, buffer_pool_persistent_watermark_json): - """Test basic output format""" - output = converter.convert( - ["buffer_pool", "persistent-watermark"], - buffer_pool_persistent_watermark_json - ) - - # Check title - assert "Shared pool maximum occupancy:" in output - - # Check data - assert "egress_lossless_pool" in output - assert "11111" in output - assert "ingress_lossless_pool" in output - assert "22222" in output - 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"], - buffer_pool_persistent_watermark_json + buffer_pool_persistent_watermark_json, + path_elems=["buffer_pool", "persistent-watermark"] ) expected = load_expected("buffer_pool_persistent_watermark.txt") assert output == expected @@ -155,8 +121,8 @@ def test_output_matches_expected(self, converter, buffer_pool_persistent_waterma def test_empty_data(self, converter): """Test empty data handling""" output = converter.convert( - ["buffer_pool", "persistent-watermark"], - {} + {}, + path_elems=["buffer_pool", "persistent-watermark"] ) assert "Shared pool maximum occupancy:" in output @@ -168,8 +134,8 @@ class TestBufferPoolIntegration: def test_both_commands_registered(self, converter): """Test both commands are registered""" - assert converter.is_supported(["buffer_pool", "watermark"]) is True - assert converter.is_supported(["buffer_pool", "persistent-watermark"]) is 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_commands_in_list(self, converter): """Test commands appear in list""" diff --git a/tests/gnmi_cli_converter/test_converter.py b/tests/gnmi_cli_converter/test_converter.py index 3eab73b54..cf1470dde 100644 --- a/tests/gnmi_cli_converter/test_converter.py +++ b/tests/gnmi_cli_converter/test_converter.py @@ -41,8 +41,8 @@ class TestCLIConverter: def test_convert_basic(self, converter, buffer_pool_watermark_json): """Test basic convert functionality""" output = converter.convert( - ["buffer_pool", "watermark"], - buffer_pool_watermark_json + buffer_pool_watermark_json, + path_elems=["buffer_pool", "watermark"] ) # Check output contains title @@ -58,22 +58,151 @@ def test_convert_basic(self, converter, buffer_pool_watermark_json): 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(["unknown", "path"], {}) + 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(["buffer_pool", "watermark"]) is True - assert converter.is_supported(["buffer_pool", "persistent-watermark"]) is 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(["unknown", "path"]) is False - assert converter.is_supported([]) is 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""" @@ -115,15 +244,15 @@ def test_wildcard_matching(self): try: # Register a command with wildcard @register(["test", "wildcard", "*"]) - def render_test_wildcard(json_data, path_elems): + 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" + assert renderer({"data": "test"}, ["test", "wildcard", "Ethernet0"], {}) == "param: Ethernet0" renderer = get_renderer(["test", "wildcard", "Ethernet4"]) - assert renderer({"data": "test"}, ["test", "wildcard", "Ethernet4"]) == "param: Ethernet4" + assert renderer({"data": "test"}, ["test", "wildcard", "Ethernet4"], {}) == "param: Ethernet4" finally: # Restore original registry diff --git a/tests/gnmi_cli_converter/test_uptime.py b/tests/gnmi_cli_converter/test_uptime.py index 0f0ab9931..ae53df16d 100644 --- a/tests/gnmi_cli_converter/test_uptime.py +++ b/tests/gnmi_cli_converter/test_uptime.py @@ -35,7 +35,7 @@ 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(["uptime"], json_data) + output = converter.convert(json_data, path_elems=["uptime"]) assert output == "up 3 weeks, 4 days, 10 hours, 15 minutes" @@ -43,7 +43,7 @@ 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(["uptime"], json_data) + output = converter.convert(json_data, path_elems=["uptime"]) assert "07:42:51 up 16 days" in output assert "load average" in output @@ -52,7 +52,7 @@ def test_empty_uptime(self, converter): """Test empty uptime""" json_data = {"uptime": ""} - output = converter.convert(["uptime"], json_data) + output = converter.convert(json_data, path_elems=["uptime"]) assert output == "" @@ -60,13 +60,13 @@ def test_missing_key(self, converter): """Test missing uptime key""" json_data = {} - output = converter.convert(["uptime"], 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(["uptime"]) is True + assert converter.is_supported(path_elems=["uptime"]) is True def test_command_in_list(self, converter): """Test uptime appears in command list"""