diff --git a/netmiko/base_connection.py b/netmiko/base_connection.py index 740986229..f6fbff937 100644 --- a/netmiko/base_connection.py +++ b/netmiko/base_connection.py @@ -1516,6 +1516,7 @@ def send_command_timing( ttp_template: Optional[str] = None, use_genie: bool = False, cmd_verify: bool = False, + raise_parsing_error: bool = False, ) -> Union[str, List[Any], Dict[str, Any]]: """Execute command_string on the SSH channel using a delay-based mechanism. Generally used for show commands. @@ -1553,6 +1554,8 @@ def send_command_timing( :param use_genie: Process command output through PyATS/Genie parser (default: False). :param cmd_verify: Verify command echo before proceeding (default: False). + + :param raise_parsing_error: Raise exception when parsing output to structured data fails. """ if delay_factor is not None or max_loops is not None: warnings.warn(DELAY_FACTOR_DEPR_SIMPLE_MSG, DeprecationWarning) @@ -1586,6 +1589,7 @@ def send_command_timing( use_genie=use_genie, textfsm_template=textfsm_template, ttp_template=ttp_template, + raise_parsing_error=raise_parsing_error, ) return return_data @@ -1666,6 +1670,7 @@ def send_command( ttp_template: Optional[str] = None, use_genie: bool = False, cmd_verify: bool = True, + raise_parsing_error: bool = False, ) -> Union[str, List[Any], Dict[str, Any]]: """Execute command_string on the SSH channel using a pattern-based mechanism. Generally used for show commands. By default this method will keep waiting to receive data until the @@ -1705,6 +1710,8 @@ def send_command( :param use_genie: Process command output through PyATS/Genie parser (default: False). :param cmd_verify: Verify command echo before proceeding (default: True). + + :param raise_parsing_error: Raise exception when parsing output to structured data fails. """ # Time to delay in each read loop @@ -1840,6 +1847,7 @@ def send_command( use_genie=use_genie, textfsm_template=textfsm_template, ttp_template=ttp_template, + raise_parsing_error=raise_parsing_error, ) return return_val diff --git a/netmiko/exceptions.py b/netmiko/exceptions.py index 8c5f25694..8a2be8eb0 100644 --- a/netmiko/exceptions.py +++ b/netmiko/exceptions.py @@ -54,3 +54,9 @@ class ReadTimeout(ReadException): """General exception indicating an error occurred during a Netmiko read operation.""" pass + + +class NetmikoParsingException(ReadException): + """Exception raised when there is a generic parsing error.""" + + pass diff --git a/netmiko/utilities.py b/netmiko/utilities.py index d425db761..db2e6aac4 100644 --- a/netmiko/utilities.py +++ b/netmiko/utilities.py @@ -25,6 +25,7 @@ from textfsm import clitable from textfsm.clitable import CliTableError from netmiko import log +from netmiko.exceptions import NetmikoParsingException # For decorators F = TypeVar("F", bound=Callable[..., Any]) @@ -341,6 +342,7 @@ def _textfsm_parse( raw_output: str, attrs: Dict[str, str], template_file: Optional[str] = None, + raise_parsing_error: bool = False, ) -> Union[str, List[Dict[str, str]]]: """Perform the actual TextFSM parsing using the CliTable object.""" tfsm_parse: Callable[..., Any] = textfsm_obj.ParseCmd @@ -353,10 +355,16 @@ def _textfsm_parse( structured_data = clitable_to_dict(textfsm_obj) if structured_data == []: + if raise_parsing_error: + raise NetmikoParsingException( + "Failed to parse CLI output with TextFSM, empty list returned." + ) return raw_output else: return structured_data - except (FileNotFoundError, CliTableError): + except (FileNotFoundError, CliTableError) as error: + if raise_parsing_error: + raise error return raw_output @@ -365,6 +373,7 @@ def get_structured_data_textfsm( platform: Optional[str] = None, command: Optional[str] = None, template: Optional[str] = None, + raise_parsing_error: bool = False, ) -> Union[str, List[Dict[str, str]]]: """ Convert raw CLI output to structured data using TextFSM template. @@ -385,7 +394,12 @@ def get_structured_data_textfsm( template_dir = get_template_dir() index_file = os.path.join(template_dir, "index") textfsm_obj = clitable.CliTable(index_file, template_dir) - output = _textfsm_parse(textfsm_obj, raw_output, attrs) + output = _textfsm_parse( + textfsm_obj, + raw_output, + attrs, + raise_parsing_error=raise_parsing_error, + ) # Retry the output if "cisco_xe" and not structured data if platform and "cisco_xe" in platform: @@ -400,7 +414,11 @@ def get_structured_data_textfsm( # CliTable with no index will fall-back to a TextFSM parsing behavior textfsm_obj = clitable.CliTable(template_dir=template_dir_alt) return _textfsm_parse( - textfsm_obj, raw_output, attrs, template_file=template_file + textfsm_obj, + raw_output, + attrs, + template_file=template_file, + raise_parsing_error=raise_parsing_error, ) @@ -408,7 +426,9 @@ def get_structured_data_textfsm( get_structured_data = get_structured_data_textfsm -def get_structured_data_ttp(raw_output: str, template: str) -> Union[str, List[Any]]: +def get_structured_data_ttp( + raw_output: str, template: str, raise_parsing_error: bool = False +) -> Union[str, List[Any]]: """ Convert raw CLI output to structured data using TTP template. @@ -423,7 +443,9 @@ def get_structured_data_ttp(raw_output: str, template: str) -> Union[str, List[A ttp_parser.parse(one=True) result: List[Any] = ttp_parser.result(format="raw") return result - except Exception: + except Exception as exception: + if raise_parsing_error: + raise exception return raw_output @@ -496,7 +518,7 @@ def run_ttp_template( def get_structured_data_genie( - raw_output: str, platform: str, command: str + raw_output: str, platform: str, command: str, raise_parsing_error: bool = False ) -> Union[str, Dict[str, Any]]: if not sys.version_info >= (3, 4): raise ValueError("Genie requires Python >= 3.4") @@ -544,7 +566,9 @@ def get_structured_data_genie( get_parser(command, device) parsed_output: Dict[str, Any] = device.parse(command, output=raw_output) return parsed_output - except Exception: + except Exception as exception: + if raise_parsing_error: + raise exception return raw_output @@ -557,16 +581,22 @@ def structured_data_converter( use_genie: bool = False, textfsm_template: Optional[str] = None, ttp_template: Optional[str] = None, + raise_parsing_error: bool = False, ) -> Union[str, List[Any], Dict[str, Any]]: """ Try structured data converters in the following order: TextFSM, TTP, Genie. - Return the first structured data found, else return the raw_data as-is. + Return the first structured data found, else return the raw_data as-is unless + `raise_parsing_error` is True, then bubble up the exception to the caller. """ command = command.strip() if use_textfsm: structured_output_tfsm = get_structured_data_textfsm( - raw_data, platform=platform, command=command, template=textfsm_template + raw_data, + platform=platform, + command=command, + template=textfsm_template, + raise_parsing_error=raise_parsing_error, ) if not isinstance(structured_output_tfsm, str): return structured_output_tfsm diff --git a/tests/unit/test_utilities.py b/tests/unit/test_utilities.py index 51e99fb05..6480e73d4 100755 --- a/tests/unit/test_utilities.py +++ b/tests/unit/test_utilities.py @@ -4,10 +4,13 @@ import sys from os.path import dirname, join, relpath import pytest +from textfsm.clitable import CliTableError from netmiko import utilities from textfsm import clitable +from netmiko.exceptions import NetmikoParsingException + RESOURCE_FOLDER = join(dirname(dirname(__file__)), "etc") RELATIVE_RESOURCE_FOLDER = join(dirname(dirname(relpath(__file__))), "etc") CONFIG_FILENAME = join(RESOURCE_FOLDER, ".netmiko.yml") @@ -299,6 +302,35 @@ def test_textfsm_missing_template(): assert result == raw_output +@pytest.mark.parametrize( + "raw_output,platform,command,exception", + [ + # Invalid output with valid template/platform + ("Unparsable output", "cisco_ios", "show version", NetmikoParsingException), + # Valid output with invalid template/platform + ( + "Cisco IOS Software, Catalyst 4500 L3 Switch Software", + "cisco_ios", + "show invalid-command", + CliTableError, + ), + # Missing platform (behaviour consistent between `raise_parsing_error=True|False` + ("", "", "show version", CliTableError), + ], +) +def test_textfsm_parsing_error_invalid_command( + raw_output, platform, command, exception +): + """Verify that an error is raised if TextFSM parsing fails and `raise_parsing_error` is set.""" + with pytest.raises(exception): + utilities.get_structured_data_textfsm( + raw_output, + platform, + command, + raise_parsing_error=True, + ) + + def test_get_structured_data_genie(): """Convert raw CLI output to structured data using Genie"""