diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 7e3c4785..41d59443 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -14,7 +14,7 @@ jobs: strategy: matrix: - python-version: ['3.10', '3.11', '3.12', '3.13'] + python-version: ['3.10', '3.11', '3.12', '3.13', '3.14'] group: [1, 2, 3, 4, 5] steps: - uses: actions/checkout@v2 diff --git a/Makefile b/Makefile index 6537e95a..a7e23825 100644 --- a/Makefile +++ b/Makefile @@ -9,7 +9,7 @@ BUILD_CMD = $(PYTHON) setup.py bdist_wheel --dist-dir=$(DIST_DIR) PYPIREPO = pypitest -DEPENDENCIES = robotframework pyyaml dill coverage Sphinx \ +DEPENDENCIES = robotframework pyyaml dill coverage Sphinx==7.4.7 \ sphinxcontrib-napoleon sphinxcontrib-mockautodoc \ sphinx-rtd-theme asyncssh PrettyTable "cryptography>=43.0" diff --git a/docs/changelog/2026/march.rst b/docs/changelog/2026/march.rst new file mode 100644 index 00000000..ed865568 --- /dev/null +++ b/docs/changelog/2026/march.rst @@ -0,0 +1,67 @@ +March 2026 +========== + +March 31 - Unicon v26.3 +------------------------ + + + +.. csv-table:: Module Versions + :header: "Modules", "Versions" + + ``unicon.plugins``, v26.3 + ``unicon``, v26.3 + + + + +Changelogs +^^^^^^^^^^ +-------------------------------------------------------------------------------- + Fix +-------------------------------------------------------------------------------- + +* bases/router/connection_provider + * Added logic to raise a traceback when when HA sync does not complete within POST_BOOT_TIMEOUT + +* removed unused `pyats` and `genie` imports + + +-------------------------------------------------------------------------------- + New +-------------------------------------------------------------------------------- + +* iosxe/ie9k + * Added plugin settings for IE9k platform. + +* iosxe/ie3k + * Added plugin settings for IE3k platform. + + +-------------------------------------------------------------------------------- + Fix +-------------------------------------------------------------------------------- + +* iosxe + * Fixed boot image to support multiple filesystems. + * Fixed encryption selection criteria on boot. + +* iosxe/iec3400 + * Removed this platform as it must be ie3k. + * Related state machine and test cases were not needed hence weren't moved to ie3k. + +* pid_tokens.csv + * Modified PID tokens + * Added IE9k family PID token mappings. + * Added ESS3300/ESS9300 family PID token mappings. + +* pid_tokens + * Added PID tokens for IE 3100, 3500 series + +* iosxe/settings + * Increased POST_BOOT_TIMEOUT to allow for longer HA sync times. + +* iosxe/patterns + * Modified want_continue pattern to match the prompt "Continue? [no]" + + diff --git a/docs/changelog/index.rst b/docs/changelog/index.rst index c0c7357e..64f23864 100644 --- a/docs/changelog/index.rst +++ b/docs/changelog/index.rst @@ -4,6 +4,7 @@ Changelog .. toctree:: :maxdepth: 2 + 2026/march 2026/february 2026/january 2025/december diff --git a/docs/changelog_plugins/2026/march.rst b/docs/changelog_plugins/2026/march.rst new file mode 100644 index 00000000..29ca9e4b --- /dev/null +++ b/docs/changelog_plugins/2026/march.rst @@ -0,0 +1,78 @@ +March 2026 +========== + +March 31 - Unicon.Plugins v26.3 +------------------------ + + + +.. csv-table:: Module Versions + :header: "Modules", "Versions" + + ``unicon.plugins``, v26.3 + ``unicon``, v26.3 + + + + +Changelogs +^^^^^^^^^^ +-------------------------------------------------------------------------------- + Fix +-------------------------------------------------------------------------------- + +* generic + * Configure service + * Refactored banner handling logic to improve maintainability. + * Updated banner processing to send lines sequentially with appropriate delays for device processing. + * HAReloadService + * Fixed command fallback check and added guard to skip sendline when command is empty. + * SwitchoverService + * Fixed command fallback check and added guard to skip sendline when command is empty. + +* iosxe + * Configure service + * Updated ACM configlet implementation to use connection context for acm_configlet parameter. + * Ensures proper persistence of acm_configlet during state transitions. + * Fixed multiline banner configuration to support variable delimiters + * HASwitchover + * Changed default command parameter from [] to None. + +* iosxe/cat9k + * stackwise_virtual + * Updated the logic to detect current state before during state change + * Updated the logic of designate handles to correctly identify active and standby state after svl configuration. + * 9500x/stackwise_virtual + * Updated the logic to detect current state before during state change + +* iosxe/cat4k + * Reload + * Fixed reload_command fallback check and added guard to skip sendline when empty. + +* iosxe/cat8k + * SwitchoverService + * Fixed command fallback check and added guard to skip sendline when empty. + +* iosxe/stack + * StackSwitchover + * Fixed command fallback check and added guard to skip sendline when empty. + * StackReload + * Fixed reload_command fallback check and added guard to skip sendline when empty. + +* iosxe/quad + * QuadSwitchover + * Fixed command fallback check and added guard to skip sendline when empty. + * QuadReload + * Fixed reload_command fallback check and added guard to skip sendline when empty. + +* iosxe/cat9k/c9350/stack + * C9350StackReload + * Fixed reload_command fallback check and added guard to skip sendline when empty. + +* iosxe/cat9k/c9500x/stackwise_virtual + * SVLStackReload + * Fixed reload_command fallback check, added guard for empty command and improved post-reload recovery and reconnection handling. + * SVLStackSwitchover + * Fixed command fallback check and added guard to skip sendline when empty. + + diff --git a/docs/changelog_plugins/index.rst b/docs/changelog_plugins/index.rst index 8e5f4bb3..6b9290fb 100644 --- a/docs/changelog_plugins/index.rst +++ b/docs/changelog_plugins/index.rst @@ -4,6 +4,7 @@ Plugins Changelog .. toctree:: :maxdepth: 2 + 2026/march 2026/february 2026/january 2025/december diff --git a/src/unicon/plugins/__init__.py b/src/unicon/plugins/__init__.py index 735a174a..352cc5d8 100644 --- a/src/unicon/plugins/__init__.py +++ b/src/unicon/plugins/__init__.py @@ -1,4 +1,4 @@ -__version__ = "26.2" +__version__ = "26.3" supported_chassis = [ 'single_rp', diff --git a/src/unicon/plugins/generic/service_implementation.py b/src/unicon/plugins/generic/service_implementation.py index 98cfe2cf..2fffaef1 100644 --- a/src/unicon/plugins/generic/service_implementation.py +++ b/src/unicon/plugins/generic/service_implementation.py @@ -934,9 +934,11 @@ def config_state_change(spawn, from_state, sm): else: sm.update_cur_state(from_state) + # Flatten multi-line input into one-command-per-line list. + flat_cmd_list = list(self.utils.flatten_splitlines_command(command) if command else []) + self.result = '' - if command: - flat_cmd = self.utils.flatten_splitlines_command(command) + if flat_cmd_list: dialog = self.dialog + self.service_dialog(handle=handle, service_dialog=reply) # Add all known states to detect state changes. for state in sm.states: @@ -957,22 +959,35 @@ def config_state_change(spawn, from_state, sm): matched_retry_sleep=self.state_change_matched_retry_sleep )) - banner_lines, command_lines, banner_delim = self.get_banner_lines(flat_cmd) + # Use flattened command list + pre_lines, banner_lines, post_lines, banner_delim = self.get_banner_lines(flat_cmd_list) # Populate context for banner_text_handler only if banner was detected if banner_lines: self.connection.log.info('Banner detected, configuring banners without state detection') + for cmd in pre_lines: + handle.spawn.sendline(cmd) + self.update_hostname_if_needed([cmd]) + self.process_dialog_on_handle(handle, dialog, timeout) + # Send banner lines for line in banner_lines: handle.spawn.sendline(line) time.sleep(0.1) handle.spawn.read_update_buffer() + self.process_dialog_on_handle(handle, dialog, timeout) - if bulk: + post_cmds = chain(post_lines, [self.commit_cmd]) if self.commit_cmd else post_lines + for cmd in post_cmds: + handle.spawn.sendline(cmd) + self.update_hostname_if_needed([cmd]) + self.process_dialog_on_handle(handle, dialog, timeout) + + elif bulk: indicator = handle.settings.BULK_CONFIG_END_INDICATOR - cmd_lst = list(chain(command_lines, [indicator])) + cmd_lst = list(chain(flat_cmd_list, [indicator])) if bulk_chunk_lines == 0: chunks = [cmd_lst] else: @@ -998,8 +1013,8 @@ def config_state_change(spawn, from_state, sm): handle.spawn.sendline(self.commit_cmd) self.process_dialog_on_handle(handle, dialog, timeout) else: - cmds = chain(command_lines, [self.commit_cmd]) \ - if self.commit_cmd else command_lines + cmds = chain(flat_cmd_list, [self.commit_cmd]) \ + if self.commit_cmd else flat_cmd_list for cmd in cmds: handle.spawn.sendline(cmd) self.update_hostname_if_needed([cmd]) @@ -1032,34 +1047,49 @@ def config_state_change(spawn, from_state, sm): def get_banner_lines(self, config_lines): - """ Process lines related to the banner command + """ Process lines related to the banner command. + Handles detection and separation of banner configuration blocks from + regular configuration commands. Supports the first banner block only; + subsequent banners (if any) will be processed sequentially. Args: - config_lines (list): list of config lines + config_lines (list): Configuration command lines Returns: - tuple: (banner_lines, command_lines, banner_delim) + tuple: (pre_lines, banner_lines, post_lines, banner_delim) + - pre_lines: Commands before the banner block + - banner_lines: The banner initiation line and content + - post_lines: Commands after the banner block + - banner_delim: The delimiter character used for the banner """ - banner_lines = [] - command_lines = [] + pre_lines, banner_lines, post_lines = [], [], [] banner_delim = None + in_banner = False + banner_seen = False for line in config_lines: - match = re.match(r'^\s*banner\s+(login|motd|exec|incoming)\s+(\S)', line) - if match: - banner_lines.append(line) - banner_delim = match.group(2) + if not in_banner and not banner_seen: + match = re.match(r'^\s*banner\s+(login|motd|exec|incoming)\s+(\S+)', line) + if match: + banner_lines.append(line) + raw_delim = match.group(2) + # Use '^C' token when present, else first character (e.g. '%') + banner_delim = '^C' if raw_delim.startswith('^C') else raw_delim[0] + in_banner = True + banner_seen = True + continue + pre_lines.append(line) continue - if banner_delim: + if in_banner: banner_lines.append(line) # End of banner when delimiter repeats as a full line if line.strip() == banner_delim: - banner_delim = None + in_banner = False continue - command_lines.append(line) + post_lines.append(line) - return banner_lines, command_lines, banner_delim + return pre_lines, banner_lines, post_lines, banner_delim def process_dialog_on_handle(self, handle, dialog, timeout): try: @@ -2175,7 +2205,7 @@ def call_service(self, # noqa: C901 if command and reload_command: raise SubCommandFailure( "Please use either 'command' or 'reload_command' parameter") - command = command or reload_command or self.command + command = command if command is not None else (reload_command if reload_command is not None else self.command) # TODO counter value must be moved to settings counter = 0 @@ -2197,7 +2227,8 @@ def call_service(self, # noqa: C901 dialog += Dialog(custom_auth_stmt) # Issue reload command - con.active.spawn.sendline(command) + if command: + con.active.spawn.sendline(command) try: reload_output = dialog.process(con.active.spawn, context=context, @@ -2373,7 +2404,7 @@ def call_service(self, command=None, # noqa: C901 category=DeprecationWarning) timeout = timeout or self.timeout - command = command or self.command + command = command if command is not None else self.command switchover_counter = con.settings.SWITCHOVER_COUNTER con.log.debug("+++ Issuing switchover on %s with " "switchover_command %s and timeout is %s +++" @@ -2403,7 +2434,8 @@ def call_service(self, command=None, # noqa: C901 context = con.standby.context # Issue switchover command - con.active.spawn.sendline(command) + if command: + con.active.spawn.sendline(command) try: dialog.process(con.active.spawn, timeout=timeout, diff --git a/src/unicon/plugins/generic/statements.py b/src/unicon/plugins/generic/statements.py index 28ae7da2..19732b21 100644 --- a/src/unicon/plugins/generic/statements.py +++ b/src/unicon/plugins/generic/statements.py @@ -339,6 +339,7 @@ def enable_secret_handler(spawn, context, session): spawn.log.warning('Using enable secret from TEMP_ENABLE_SECRET setting') enable_secret = spawn.settings.TEMP_ENABLE_SECRET context['setup_selection'] = 0 + context['encryption_selection'] = 2 spawn.sendline(enable_secret) @@ -352,6 +353,16 @@ def setup_enter_selection(spawn, context): spawn.sendline('2') +def setup_enter_encryption_selection(spawn, context): + selection = context.get('encryption_selection', context.get('setup_selection')) + if selection is not None: + if str(selection) == '0': + spawn.log.warning('Not saving setup configuration') + spawn.sendline(f'{selection}') + else: + spawn.sendline('2') + + def ssh_tacacs_handler(spawn, context): result = False start_cmd = spawn.spawn_command @@ -781,7 +792,7 @@ def __init__(self): continue_timer=False) self.enter_your_encryption_selection_stmt = Statement(pattern=pat.enter_your_encryption_selection_2, - action=setup_enter_selection, + action=setup_enter_encryption_selection, args=None, loop_continue=True, continue_timer=True) diff --git a/src/unicon/plugins/iosxe/cat4k/service_implementation.py b/src/unicon/plugins/iosxe/cat4k/service_implementation.py index 18eab61c..b50e5f14 100644 --- a/src/unicon/plugins/iosxe/cat4k/service_implementation.py +++ b/src/unicon/plugins/iosxe/cat4k/service_implementation.py @@ -105,7 +105,7 @@ def call_service(self, # noqa: C901 timeout = timeout or self.timeout - command = reload_command or self.command + command = reload_command if reload_command is not None else self.command fmt_str = "+++ reloading %s with reload_command %s and timeout is %s +++" con.log.info(fmt_str % (con.hostname, command, timeout)) @@ -125,7 +125,8 @@ def call_service(self, # noqa: C901 dialog += Dialog(custom_auth_stmt) # Issue reload command - con.active.spawn.sendline(command) + if command: + con.active.spawn.sendline(command) try: dialog.process(con.active.spawn, context=context, diff --git a/src/unicon/plugins/iosxe/cat8k/service_implementation.py b/src/unicon/plugins/iosxe/cat8k/service_implementation.py index 6e8c7af0..e3771d26 100644 --- a/src/unicon/plugins/iosxe/cat8k/service_implementation.py +++ b/src/unicon/plugins/iosxe/cat8k/service_implementation.py @@ -66,7 +66,7 @@ def call_service(self, command=None, # create an alias for connection. con = self.connection timeout = timeout or self.timeout - command = command or self.command + command = command if command is not None else self.command if (reply is None) or (reply == []): reply = Dialog([]) @@ -93,7 +93,8 @@ def call_service(self, command=None, # Issue switchover command - con.spawn.sendline(command) + if command: + con.spawn.sendline(command) try: reply.process(con.spawn, timeout=timeout, diff --git a/src/unicon/plugins/iosxe/cat9k/c9350/stack/service_implementation.py b/src/unicon/plugins/iosxe/cat9k/c9350/stack/service_implementation.py index b09ee40e..b7789885 100644 --- a/src/unicon/plugins/iosxe/cat9k/c9350/stack/service_implementation.py +++ b/src/unicon/plugins/iosxe/cat9k/c9350/stack/service_implementation.py @@ -67,7 +67,7 @@ def call_service(self, if member: reload_command = f'reload slot {member}' - reload_cmd = reload_command or self.reload_command + reload_cmd = reload_command if reload_command is not None else self.reload_command timeout = timeout or self.timeout conn = self.connection.active @@ -122,7 +122,8 @@ def call_service(self, conn.context['post_reload_wait_time'] = timedelta(seconds= self.post_reload_wait_time) conn.log.info('Processing on active rp %s-%s with timeout %s' % (conn.hostname, conn.alias, timeout)) - conn.sendline(reload_cmd) + if reload_cmd: + conn.sendline(reload_cmd) conn_list = self.connection.subconnections reload_cmd_output = None diff --git a/src/unicon/plugins/iosxe/cat9k/c9500x/stackwise_virtual/service_implementation.py b/src/unicon/plugins/iosxe/cat9k/c9500x/stackwise_virtual/service_implementation.py index 384c49e2..8b5db254 100644 --- a/src/unicon/plugins/iosxe/cat9k/c9500x/stackwise_virtual/service_implementation.py +++ b/src/unicon/plugins/iosxe/cat9k/c9500x/stackwise_virtual/service_implementation.py @@ -66,7 +66,7 @@ def call_service(self, self.result = False if member: reload_command = f'reload slot {member}' - reload_cmd = reload_command or self.reload_command + reload_cmd = reload_command if reload_command is not None else self.reload_command timeout = timeout or self.timeout conn = self.connection.active @@ -118,7 +118,8 @@ def call_service(self, conn.context['post_reload_wait_time'] = timedelta(seconds= self.post_reload_wait_time) conn.log.info('Processing on active rp %s-%s with timeout %s' % (conn.hostname, conn.alias, timeout)) - conn.sendline(reload_cmd) + if reload_cmd: + conn.sendline(reload_cmd) conn_list = self.connection.subconnections reload_cmd_output = None @@ -223,13 +224,13 @@ def boot(con): raise SubCommandFailure('Error during reload', e) from e else: try: - # bring device to enable mode - conn.sendline() - conn.log.info("Bringing device to any state") + conn.log.info("Bring device to any state") conn.state_machine.go_to('any', conn.spawn, timeout=timeout, prompt_recovery=self.prompt_recovery, context=conn.context) - + conn.state_machine.go_to('enable', conn.spawn, timeout=timeout, + prompt_recovery=self.prompt_recovery, + context=conn.context) except Exception as e: raise SubCommandFailure('Failed to bring device to disable mode.', e) from e @@ -257,8 +258,9 @@ def boot(con): self.connection.settings.STACK_POST_RELOAD_SLEEP) sleep(self.connection.settings.STACK_POST_RELOAD_SLEEP) - self.connection.log.info('Initialize the connection after reload') - self.connection.connection_provider.init_connection() + self.connection.log.info('Disconnecting and reconnecting') + self.connection.disconnect() + self.connection.connect() self.connection.log.info("+++ Reload Completed Successfully +++") @@ -312,7 +314,7 @@ def call_service(self, command=None, timeout=None, *args, **kwargs): - switchover_cmd = command or self.command + switchover_cmd = command if command is not None else self.command timeout = timeout or self.timeout conn = self.connection.active @@ -327,7 +329,8 @@ def call_service(self, command=None, dialog += connect_dialog conn.log.info('Processing on active rp %s-%s' % (conn.hostname, conn.alias)) - conn.sendline(switchover_cmd) + if switchover_cmd: + conn.sendline(switchover_cmd) try: # A loop has been implemented to handle the # "Press RETURN to get started" prompt twice. Based on extensive diff --git a/src/unicon/plugins/iosxe/cat9k/stackwise_virtual/connection_provider.py b/src/unicon/plugins/iosxe/cat9k/stackwise_virtual/connection_provider.py index 0a4dcd98..97d56cb0 100644 --- a/src/unicon/plugins/iosxe/cat9k/stackwise_virtual/connection_provider.py +++ b/src/unicon/plugins/iosxe/cat9k/stackwise_virtual/connection_provider.py @@ -51,15 +51,19 @@ def designate_handles(self): for subcon in [subcon1, subcon2]: try: - subcon.state_machine.go_to( - 'enable', - subcon.spawn, - context=subcon.context, - timeout=con.settings.BOOT_TIMEOUT, - dialog=standby_locked_dialog, - ) - except Exception: - pass + # Attempt to detect the current state of the subcon and go to enable if not already there + subcon.state_machine.detect_state(subcon.spawn, subcon.context) + if subcon.state_machine.current_state != 'enable': + subcon.sendline() + subcon.state_machine.go_to( + 'enable', + subcon.spawn, + context=subcon.context, + timeout=con.settings.BOOT_TIMEOUT, + dialog=standby_locked_dialog, + ) + except Exception as e: + con.log.exception('Failed to go to enable on %s: %s', subcon.alias, e) con.log.debug('{} in state: {}'.format(subcon.alias, subcon.state_machine.current_state)) if subcon1.state_machine.current_state == 'enable': @@ -85,13 +89,14 @@ def designate_handles(self): output = {} stack_info = output.get("switch", {}).get("stack", {}) roles = [switch_info.get("role") for switch_info in stack_info.values()] + roles_lower = [str(role).lower() for role in roles if role is not None] - if "active" in roles and "standby" in roles: + if "active" in roles_lower and "standby" in roles_lower: # Only designate handle when in SVL state # There are case when in non-SVL the device connection # becomes active for both connection and there isn't a standby state # it would have either active and member state or just active state - + # Verify the active and standby target_con.spawn.sendline(target_con.spawn.settings.SHOW_REDUNDANCY_CMD) output = target_con.spawn.expect( diff --git a/src/unicon/plugins/iosxe/ie3k/__init__.py b/src/unicon/plugins/iosxe/ie3k/__init__.py new file mode 100644 index 00000000..f15d9549 --- /dev/null +++ b/src/unicon/plugins/iosxe/ie3k/__init__.py @@ -0,0 +1,13 @@ +from unicon.plugins.iosxe import IosXEDualRPConnection, IosXESingleRpConnection + +from .settings import IosXEIe3kSettings + + +class IosXEIe3kSingleRpConnection(IosXESingleRpConnection): + platform = 'ie3k' + settings = IosXEIe3kSettings() + + +class IosXEIe3kDualRPConnection(IosXEDualRPConnection): + platform = 'ie3k' + settings = IosXEIe3kSettings() diff --git a/src/unicon/plugins/iosxe/ie3k/settings.py b/src/unicon/plugins/iosxe/ie3k/settings.py new file mode 100644 index 00000000..dd8c54ec --- /dev/null +++ b/src/unicon/plugins/iosxe/ie3k/settings.py @@ -0,0 +1,9 @@ +from unicon.plugins.iosxe.settings import IosXESettings + + +class IosXEIe3kSettings(IosXESettings): + + def __init__(self): + super().__init__() + + self.BOOT_FILESYSTEM = ["sdflash:", "flash:"] diff --git a/src/unicon/plugins/iosxe/ie9k/__init__.py b/src/unicon/plugins/iosxe/ie9k/__init__.py new file mode 100644 index 00000000..6d655bb8 --- /dev/null +++ b/src/unicon/plugins/iosxe/ie9k/__init__.py @@ -0,0 +1,13 @@ +from unicon.plugins.iosxe import IosXEDualRPConnection, IosXESingleRpConnection + +from .settings import IosXEIe9kSettings + + +class IosXEIe9kSingleRpConnection(IosXESingleRpConnection): + platform = 'ie9k' + settings = IosXEIe9kSettings() + + +class IosXEIe9kDualRPConnection(IosXEDualRPConnection): + platform = 'ie9k' + settings = IosXEIe9kSettings() diff --git a/src/unicon/plugins/iosxe/ie9k/settings.py b/src/unicon/plugins/iosxe/ie9k/settings.py new file mode 100644 index 00000000..0cd9f25a --- /dev/null +++ b/src/unicon/plugins/iosxe/ie9k/settings.py @@ -0,0 +1,9 @@ +from unicon.plugins.iosxe.settings import IosXESettings + + +class IosXEIe9kSettings(IosXESettings): + + def __init__(self): + super().__init__() + + self.BOOT_FILESYSTEM = ["sdflash:", "flash:"] diff --git a/src/unicon/plugins/iosxe/iec3400/__init__.py b/src/unicon/plugins/iosxe/iec3400/__init__.py deleted file mode 100644 index a62e7b4d..00000000 --- a/src/unicon/plugins/iosxe/iec3400/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ - -from unicon.plugins.iosxe import IosXEServiceList, IosXESingleRpConnection - -from .settings import IosXEIec3400Settings -from . import service_implementation as svc -from .statemachine import IosXEIec3400SingleRpStateMachine - - -class IosXEIec3400ServiceList(IosXEServiceList): - def __init__(self): - super().__init__() - self.reload = svc.Reload - - -class IosXEIec3400SingleRpConnection(IosXESingleRpConnection): - os = 'iosxe' - platform = 'iec3400' - chassis_type = 'single_rp' - state_machine_class = IosXEIec3400SingleRpStateMachine - subcommand_list = IosXEIec3400ServiceList - settings = IosXEIec3400Settings() diff --git a/src/unicon/plugins/iosxe/iec3400/service_implementation.py b/src/unicon/plugins/iosxe/iec3400/service_implementation.py deleted file mode 100644 index 3ebd85f8..00000000 --- a/src/unicon/plugins/iosxe/iec3400/service_implementation.py +++ /dev/null @@ -1,93 +0,0 @@ - -from unicon.bases.routers.services import BaseService -from unicon.plugins.generic.service_implementation import ReloadResult -from unicon.eal.dialogs import Dialog -from unicon.core.errors import SubCommandFailure -from unicon.utils import AttributeDict - -from .service_statements import reload_statement_list - - -class Reload(BaseService): - """Service to reload the device. - - Arguments: - reload_command: reload command to be issued on device. - default reload_command is "reload" - dialog: Dialog which include list of Statements for - additional dialogs prompted by reload command, in-case - it is not in the current list. - timeout: Timeout value in sec, Default Value is 400 sec - image_to_boot: image to be used if the device stops in rommon mode - - Returns: - bool: True on success False otherwise - - Raises: - SubCommandFailure: on failure. - - Example: - .. code-block:: python - - uut.reload() - """ - - def __init__(self, connection, context, **kwargs): - super().__init__(connection, context, **kwargs) - self.start_state = 'enable' - self.end_state = 'enable' - self.timeout = connection.settings.RELOAD_TIMEOUT - self.dialog = Dialog(reload_statement_list) - - def call_service(self, - reload_command='reload', - dialog=Dialog([]), - timeout=None, - return_output=False, - error_pattern=None, - append_error_pattern=None, - *args, - **kwargs): - - con = self.connection - timeout = timeout or self.timeout - - if error_pattern is None: - self.error_pattern = con.settings.ERROR_PATTERN - else: - self.error_pattern = error_pattern - - if not isinstance(self.error_pattern, list): - raise ValueError('error_pattern should be a list') - if append_error_pattern: - if not isinstance(append_error_pattern, list): - raise ValueError('append_error_pattern should be a list') - self.error_pattern += append_error_pattern - sm = self.get_sm() - assert isinstance(dialog, - Dialog), "dialog passed must be an instance of Dialog" - dialog += self.dialog - - con.log.debug( - "+++ reloading {} with reload_command {} and timeout is {} +++" - .format(self.connection.hostname, reload_command, timeout)) - - context = AttributeDict(self.context) - dialog = self.service_dialog(service_dialog=dialog) - dialog += Dialog([[sm.get_state('disable').pattern]]) - con.spawn.sendline(reload_command) - try: - reload_op=dialog.process(con.spawn, context=context, timeout=timeout, - prompt_recovery=self.prompt_recovery) - sm.detect_state(con.spawn, context=context) - con.state_machine.go_to('enable', con.spawn, - context=context, - timeout=con.connection_timeout, - prompt_recovery=self.prompt_recovery) - except Exception as err: - raise SubCommandFailure("Reload failed : {}".format(err)) - - con.log.debug("+++ Reload Completed Successfully +++") - self.result = True - if return_output: - self.result = ReloadResult(self.result, reload_op.match_output.replace(reload_command, '', 1)) diff --git a/src/unicon/plugins/iosxe/iec3400/service_statements.py b/src/unicon/plugins/iosxe/iec3400/service_statements.py deleted file mode 100644 index 73f15087..00000000 --- a/src/unicon/plugins/iosxe/iec3400/service_statements.py +++ /dev/null @@ -1,13 +0,0 @@ - -from unicon.eal.dialogs import Statement - - -reload_proceed_stmt = Statement(pattern=r'.*Proceed with reload\?\[y/n]\s*$', - action='sendline(y)', - loop_continue=True, - continue_timer=False) - - -reload_statement_list = [ - reload_proceed_stmt -] diff --git a/src/unicon/plugins/iosxe/iec3400/settings.py b/src/unicon/plugins/iosxe/iec3400/settings.py deleted file mode 100644 index 3710bd0b..00000000 --- a/src/unicon/plugins/iosxe/iec3400/settings.py +++ /dev/null @@ -1,13 +0,0 @@ - - -from unicon.plugins.iosxe.settings import IosXESettings - - -class IosXEIec3400Settings(IosXESettings): - - def __init__(self): - super().__init__() - self.RELOAD_TIMEOUT = 120 - - self.HA_INIT_EXEC_COMMANDS = [] - self.HA_INIT_CONFIG_COMMANDS = [] diff --git a/src/unicon/plugins/iosxe/iec3400/statemachine.py b/src/unicon/plugins/iosxe/iec3400/statemachine.py deleted file mode 100644 index c0cbcc18..00000000 --- a/src/unicon/plugins/iosxe/iec3400/statemachine.py +++ /dev/null @@ -1,14 +0,0 @@ - -from unicon.plugins.generic.service_statements import generic_statements - -from unicon.plugins.iosxe.statemachine import IosXESingleRpStateMachine - - -class IosXEIec3400SingleRpStateMachine(IosXESingleRpStateMachine): - - def create(self): - super().create() - config_to_enable = self.get_path('config', 'enable') - config_to_enable.command = 'exit' - - self.add_default_statements([generic_statements.terminal_position_stmt]) diff --git a/src/unicon/plugins/iosxe/patterns.py b/src/unicon/plugins/iosxe/patterns.py index 7ce5b8a5..d19b8e98 100644 --- a/src/unicon/plugins/iosxe/patterns.py +++ b/src/unicon/plugins/iosxe/patterns.py @@ -19,7 +19,7 @@ def __init__(self): r'^.*Are you sure you want to continue\? \(y\/n\)\[y\]:?\s?$' self.delete_filename = r'^.*Delete filename \[.*\]\?\s*$' self.wish_continue = r'^.*Do you wish to continue\? \[yes\]:\s*$' - self.want_continue = r'^.*Do you want to continue\? \[no\]:\s*$' + self.want_continue = r'^.*(Do you want to )?[Cc]ontinue\? \[no\]:\s*$' self.want_continue_confirm = r'.*Do you want to continue\?\s*\[confirm]\s*$' self.want_continue_yes = r'.*Do you want to continue\?\s*\[y/n]\?\s*\[yes]:\s*$' self.disable_prompt = \ diff --git a/src/unicon/plugins/iosxe/quad/service_implementation.py b/src/unicon/plugins/iosxe/quad/service_implementation.py index 10c89e37..85f79388 100644 --- a/src/unicon/plugins/iosxe/quad/service_implementation.py +++ b/src/unicon/plugins/iosxe/quad/service_implementation.py @@ -121,7 +121,7 @@ def call_service(self, command=None, *args, **kwargs): self.result = False - switchover_cmd = command or self.command + switchover_cmd = command if command is not None else self.command timeout = timeout or self.timeout conn = self.connection.active @@ -137,7 +137,8 @@ def call_service(self, command=None, self.connection.log.info('Processing on original Global Active rp ' '%s-%s' % (conn.hostname, conn.alias)) - conn.sendline(switchover_cmd) + if switchover_cmd: + conn.sendline(switchover_cmd) try: active_output = dialog.process(conn.spawn, timeout=timeout, prompt_recovery=self.prompt_recovery, @@ -262,7 +263,7 @@ def call_service(self, **kwargs): self.result = False - reload_cmd = reload_command or self.reload_command + reload_cmd = reload_command if reload_command is not None else self.reload_command timeout = timeout or self.timeout conn = self.connection.active @@ -289,7 +290,8 @@ def call_service(self, self.connection.log.info('Processing on rp %s-%s' % (conn.hostname, conn.alias)) - conn.sendline(reload_cmd) + if reload_cmd: + conn.sendline(reload_cmd) try: reload_output = reload_dialog.process(conn.spawn, timeout=timeout, prompt_recovery=self.prompt_recovery, diff --git a/src/unicon/plugins/iosxe/service_implementation.py b/src/unicon/plugins/iosxe/service_implementation.py index 63758745..dc249002 100644 --- a/src/unicon/plugins/iosxe/service_implementation.py +++ b/src/unicon/plugins/iosxe/service_implementation.py @@ -7,7 +7,6 @@ from unicon.core.errors import SubCommandFailure from unicon.bases.routers.services import BaseService - from unicon.plugins.generic.service_implementation import ( Configure as GenericConfigure, Execute as GenericExecute, @@ -65,6 +64,10 @@ def pre_service(self, *args, **kwargs): self.rules = kwargs.pop('rules', False) self.prompt_recovery = kwargs.get('prompt_recovery', True) + self.connection.context.pop('acm_configlet', None) + self.start_state = 'config' + self.end_state = 'enable' + if self.acm_configlet: self.connection.state_machine.go_to('acm', self.connection.spawn,context={'acm_configlet': self.acm_configlet}) self.start_state = 'acm' @@ -222,7 +225,7 @@ def call_service(self, command=[], reload_command=[], reply=Dialog([]), timeout= class HASwitchover(GenericHASwitchover): - def call_service(self, command=[], reply=Dialog([]), timeout=None, *args, + def call_service(self, command=None, reply=Dialog([]), timeout=None, *args, **kwargs): super().call_service(command, reply=reply + Dialog([confirm]), timeout=timeout, *args, **kwargs) diff --git a/src/unicon/plugins/iosxe/settings.py b/src/unicon/plugins/iosxe/settings.py index 485d2059..db4b2634 100644 --- a/src/unicon/plugins/iosxe/settings.py +++ b/src/unicon/plugins/iosxe/settings.py @@ -51,7 +51,7 @@ def __init__(self): self.CONFIG_LOCK_RETRY_SLEEP = 30 self.CONFIG_LOCK_RETRIES = 10 - self.POST_BOOT_TIMEOUT = 300 + self.POST_BOOT_TIMEOUT = 900 self.BOOT_POSTCHECK_INTERVAL = 30 self.SERVICE_PROMPT_CONFIG_CMD = 'service prompt config' diff --git a/src/unicon/plugins/iosxe/stack/service_implementation.py b/src/unicon/plugins/iosxe/stack/service_implementation.py index 52ef3557..f1abdf47 100644 --- a/src/unicon/plugins/iosxe/stack/service_implementation.py +++ b/src/unicon/plugins/iosxe/stack/service_implementation.py @@ -105,7 +105,7 @@ def call_service(self, command=None, timeout=None, *args, **kwargs): - switchover_cmd = command or self.command + switchover_cmd = command if command is not None else self.command timeout = timeout or self.timeout conn = self.connection.active @@ -120,7 +120,8 @@ def call_service(self, command=None, dialog += connect_dialog conn.log.info('Processing on active rp %s-%s' % (conn.hostname, conn.alias)) - conn.sendline(switchover_cmd) + if switchover_cmd: + conn.sendline(switchover_cmd) try: match_object = dialog.process(conn.spawn, timeout=timeout, prompt_recovery=self.prompt_recovery, @@ -218,7 +219,7 @@ def call_service(self, if member: reload_command = f'reload slot {member}' - reload_cmd = reload_command or self.reload_command + reload_cmd = reload_command if reload_command is not None else self.reload_command timeout = timeout or self.timeout conn = self.connection.active @@ -275,7 +276,8 @@ def call_service(self, conn.context['post_reload_wait_time'] = timedelta(seconds= self.post_reload_wait_time) conn.log.info('Processing on active rp %s-%s with timeout %s' % (conn.hostname, conn.alias, timeout)) - conn.sendline(reload_cmd) + if reload_cmd: + conn.sendline(reload_cmd) conn_list = self.connection.subconnections diff --git a/src/unicon/plugins/iosxe/statements.py b/src/unicon/plugins/iosxe/statements.py index 9a95668c..9f07dfe4 100644 --- a/src/unicon/plugins/iosxe/statements.py +++ b/src/unicon/plugins/iosxe/statements.py @@ -158,19 +158,37 @@ def boot_image(spawn, context, session): elif context.get('image_to_boot', '').strip(): cmd = "boot {}".format(context['image_to_boot']).strip() elif spawn.settings.FIND_BOOT_IMAGE: - filesystem = spawn.settings.BOOT_FILESYSTEM if \ - hasattr(spawn.settings, 'BOOT_FILESYSTEM') else 'flash:' - spawn.buffer = '' - spawn.sendline('dir {}'.format(filesystem)) - dir_listing = spawn.expect(patterns.rommon_prompt).match_output - boot_file_regex = spawn.settings.BOOT_FILE_REGEX if \ - hasattr(spawn.settings, 'BOOT_FILE_REGEX') else r'(\S+\.bin)' - m = re.search(boot_file_regex, dir_listing) - if m: - boot_image = m.group(1) - cmd = "boot {}{}".format(filesystem, boot_image) + if context.get('filesystem_images'): + # Attempt to boot the next image from the list of images + # collected from the possible boot filesystems. + cmd = "boot {}".format(context['filesystem_images'].pop(0)) else: - cmd = "boot" + if hasattr(spawn.settings, 'BOOT_FILESYSTEM'): + if isinstance(spawn.settings.BOOT_FILESYSTEM, list): + filesystems = spawn.settings.BOOT_FILESYSTEM + else: + filesystems = [spawn.settings.BOOT_FILESYSTEM] + else: + filesystems = ["flash:"] + # Collect all possible boot images from the filesystems and + # attempt to boot each one until successful / max attempts reached. + context['filesystem_images'] = [] + for fs in filesystems: + spawn.buffer = '' + spawn.sendline('dir {}'.format(fs)) + dir_listing = spawn.expect(patterns.rommon_prompt).match_output + boot_file_regex = spawn.settings.BOOT_FILE_REGEX if \ + hasattr(spawn.settings, 'BOOT_FILE_REGEX') else r'(\S+\.bin)' + matches = re.findall(boot_file_regex, dir_listing) + if matches: + context['filesystem_images'].extend( + [fs + image_name for image_name in matches] + ) + # Attempt to boot the first image from the list. + if context['filesystem_images']: + cmd = "boot {}".format(context['filesystem_images'].pop(0)) + else: + cmd = "boot" else: cmd = "boot" spawn.sendline(cmd) diff --git a/src/unicon/plugins/pid_tokens.csv b/src/unicon/plugins/pid_tokens.csv index bf068896..3306befd 100644 --- a/src/unicon/plugins/pid_tokens.csv +++ b/src/unicon/plugins/pid_tokens.csv @@ -711,6 +711,29 @@ CW9178I,cheetah,ap,cw9100, CW9800H1,iosxe,cat9k,c9800, CW9800H2,iosxe,cat9k,c9800, CW9800L,iosxe,cat9k,c9800, +ESS-3300-24T-CON-A,iosxe,ess3k,ess3300, +ESS-3300-24T-CON-E,iosxe,ess3k,ess3300, +ESS-3300-24T-NCP-A,iosxe,ess3k,ess3300, +ESS-3300-24T-NCP-E,iosxe,ess3k,ess3300, +ESS-3300-CON-A,iosxe,ess3k,ess3300, +ESS-3300-CON-E,iosxe,ess3k,ess3300, +ESS-3300-NCP-A,iosxe,ess3k,ess3300, +ESS-3300-NCP-E,iosxe,ess3k,ess3300, +ESS-9300-10X-E,iosxe,ess9k,es9300, +ESS-9300-8X16T-W,iosxe,ess9k,es9300, +IE-3100-18T2C-CC-E,iosxe,ie3k,ie3100, +IE-3100-18T2C-E,iosxe,ie3k,ie3100, +IE-3100-3P1U2S-E,iosxe,ie3k,ie3100, +IE-3100-4P2S-E,iosxe,ie3k,ie3100, +IE-3100-4T2S-E,iosxe,ie3k,ie3100, +IE-3100-6P2U2C-E,iosxe,ie3k,ie3100, +IE-3100-8P2C-E,iosxe,ie3k,ie3100, +IE-3100-8T2C-E,iosxe,ie3k,ie3100, +IE-3100-8T4S-E,iosxe,ie3k,ie3100, +IE-3100H-6FT2T-E,iosxe,ie3k,ie3100h, +IE-3100H-8T-E,iosxe,ie3k,ie3100h, +IE-3105-18T2C-E,iosxe,ie3k,ie3105, +IE-3105-8T2C-E,iosxe,ie3k,ie3105, IE-3200-8P2S-E,iosxe,ie3k,ie3200, IE-3200-8T2S-E,iosxe,ie3k,ie3200, IE-3300-8P2S-A,iosxe,ie3k,ie3300, @@ -725,19 +748,63 @@ IE-3400-8P2S-A,iosxe,ie3k,ie3400, IE-3400-8P2S-E,iosxe,ie3k,ie3400, IE-3400-8T2S-A,iosxe,ie3k,ie3400, IE-3400-8T2S-E,iosxe,ie3k,ie3400, -IE-3400H-16FT-A,iosxe,ie3k,ie3400, -IE-3400H-16FT-E,iosxe,ie3k,ie3400, -IE-3400H-16T-A,iosxe,ie3k,ie3400, -IE-3400H-16T-E,iosxe,ie3k,ie3400, -IE-3400H-24FT-A,iosxe,ie3k,ie3400, -IE-3400H-24FT-E,iosxe,ie3k,ie3400, -IE-3400H-24T-A,iosxe,ie3k,ie3400, -IE-3400H-24T-E,iosxe,ie3k,ie3400, -IE-3400H-8FT-A,iosxe,ie3k,ie3400, -IE-3400H-8FT-E,iosxe,ie3k,ie3400, -IE-3400H-8T-A,iosxe,ie3k,ie3400, -IE-3400H-8T-E,iosxe,ie3k,ie3400, -IR1101-K9,iosxe,ir1k,ir1100, +IE-3400H-16FT-A,iosxe,ie3k,ie3400h, +IE-3400H-16FT-E,iosxe,ie3k,ie3400h, +IE-3400H-16T-A,iosxe,ie3k,ie3400h, +IE-3400H-16T-E,iosxe,ie3k,ie3400h, +IE-3400H-24FT-A,iosxe,ie3k,ie3400h, +IE-3400H-24FT-E,iosxe,ie3k,ie3400h, +IE-3400H-24T-A,iosxe,ie3k,ie3400h, +IE-3400H-24T-E,iosxe,ie3k,ie3400h, +IE-3400H-8FT-A,iosxe,ie3k,ie3400h, +IE-3400H-8FT-E,iosxe,ie3k,ie3400h, +IE-3400H-8T-A,iosxe,ie3k,ie3400h, +IE-3400H-8T-E,iosxe,ie3k,ie3400h, +IE-3500-8P3S-A,iosxe,ie3k,ie3500, +IE-3500-8P3S-E,iosxe,ie3k,ie3500, +IE-3500-8T3S-A,iosxe,ie3k,ie3500, +IE-3500-8T3S-E,iosxe,ie3k,ie3500, +IE-3500-8T3X-A,iosxe,ie3k,ie3500, +IE-3500-8T3X-E,iosxe,ie3k,ie3500, +IE-3500-8U3X-A,iosxe,ie3k,ie3500, +IE-3500-8U3X-E,iosxe,ie3k,ie3500, +IE-3500H-12FT4T-A,iosxe,ie3k,ie3500h, +IE-3500H-12FT4T-E,iosxe,ie3k,ie3500h, +IE-3500H-12P2MU2XA,iosxe,ie3k,ie3500h, +IE-3500H-12P2MU2XE,iosxe,ie3k,ie3500h, +IE-3500H-14P2T-A,iosxe,ie3k,ie3500h, +IE-3500H-14P2T-E,iosxe,ie3k,ie3500h, +IE-3500H-16T-A,iosxe,ie3k,ie3500h, +IE-3500H-16T-E,iosxe,ie3k,ie3500h, +IE-3500H-20FT4T-A,iosxe,ie3k,ie3500h, +IE-3500H-20FT4T-E,iosxe,ie3k,ie3500h, +IE-3500H-24T-A,iosxe,ie3k,ie3500h, +IE-3500H-24T-E,iosxe,ie3k,ie3500h, +IE-3500H-8T-A,iosxe,ie3k,ie3500h, +IE-3500H-8T-E,iosxe,ie3k,ie3500h, +IE-3505-8P3S-A,iosxe,ie3k,ie3505, +IE-3505-8P3S-E,iosxe,ie3k,ie3505, +IE-3505-8T3S-A,iosxe,ie3k,ie3505, +IE-3505-8T3S-E,iosxe,ie3k,ie3505, +IE-3505H-16T-A,iosxe,ie3k,ie3505h, +IE-3505H-16T-E,iosxe,ie3k,ie3505h, +IE-9310-16P8S4X-A,iosxe,ie9k,ie9310, +IE-9310-16P8S4X-E,iosxe,ie9k,ie9310, +IE-9310-26S2C-A,iosxe,ie9k,ie9310, +IE-9310-26S2C-E,iosxe,ie9k,ie9310, +IE-9320-16P8U4X-A,iosxe,ie9k,ie9320, +IE-9320-16P8U4X-E,iosxe,ie9k,ie9320, +IE-9320-22S2C4X-A,iosxe,ie9k,ie9320, +IE-9320-22S2C4X-E,iosxe,ie9k,ie9320, +IE-9320-24P4S-A,iosxe,ie9k,ie9320, +IE-9320-24P4S-E,iosxe,ie9k,ie9320, +IE-9320-24P4X-A,iosxe,ie9k,ie9320, +IE-9320-24P4X-E,iosxe,ie9k,ie9320, +IE-9320-24T4X-A,iosxe,ie9k,ie9320, +IE-9320-24T4X-E,iosxe,ie9k,ie9320, +IE-9320-26S2C-A,iosxe,ie9k,ie9320, +IE-9320-26S2C-E,iosxe,ie9k,ie9320, +IR1101-K9,iosxe,ir1k,ir1101, IR1833-K9,iosxe,ir1k,ir1800, ISR1100-4G,iosxe,isr1k,isr1100, ISR1100-4GLTEGB,iosxe,isr1k,isr1100, diff --git a/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data.yaml b/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data.yaml index 0d6ab9f4..db5e2e1e 100644 --- a/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data.yaml +++ b/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data.yaml @@ -494,6 +494,18 @@ general_config: new_state: wsma_agent "banner login *": new_state: banner_config_state + "vrf definition Mgmt": "" + "interface loopback 2": "" + "description test": "" + "description plain test": "" + " ": " " + " interface loopback 1 ": "" + "description test device": "" + " no shutdown ": "" + "banner login *": + new_state: banner_config_state + "end": + new_state: general_enable wsma_agent: prompt: "%N(wsma-exec-agent)#" @@ -1583,6 +1595,7 @@ tclsh: banner_config_state: prompt: "" commands: + "": "" "Updated - Unauthorized access to this device is strictly prohibited.": "" "Please contact corpnet@fb.com with any access requests.": "" "Done": "" @@ -1606,6 +1619,10 @@ acm_configlet: "banner login %": response: "Enter TEXT message. End with the character '%'." new_state: acm_configlet_banner_input_percent + "interface loopback 2": + new_state: acm_if + "invalid command example": + response: "% Invalid input detected at '^' marker." "end": new_state: general_enable diff --git a/src/unicon/plugins/tests/test_plugin_iosxe.py b/src/unicon/plugins/tests/test_plugin_iosxe.py index ed562305..2ef289f2 100644 --- a/src/unicon/plugins/tests/test_plugin_iosxe.py +++ b/src/unicon/plugins/tests/test_plugin_iosxe.py @@ -11,6 +11,7 @@ import os import time import unittest +from types import SimpleNamespace from unittest.mock import patch, Mock, MagicMock from pyats.topology import loader @@ -20,12 +21,78 @@ from unicon.eal.dialogs import Dialog, Statement from unicon.eal.utils import ExpectMatch, MatchMode from unicon.core.errors import SubCommandFailure, StateMachineError, UniconAuthenticationError, ConnectionError as UniconConnectionError +from unicon.plugins.iosxe.statements import boot_image from unicon.plugins.tests.mock.mock_device_iosxe import MockDeviceTcpWrapperIOSXE unicon.settings.Settings.POST_DISCONNECT_WAIT_SEC = 0 unicon.settings.Settings.GRACEFUL_DISCONNECT_WAIT_SEC = 0.2 +class TestIosXEStatements(unittest.TestCase): + + def test_boot_image_finds_images_from_multiple_filesystems(self): + spawn = Mock() + spawn.settings = SimpleNamespace( + MAX_BOOT_ATTEMPTS=5, + FIND_BOOT_IMAGE=True, + BOOT_FILESYSTEM=['bootflash:', 'flash:'], + BOOT_FILE_REGEX=r'(\S+\.bin)' + ) + spawn.expect.side_effect = [ + Mock(match_output='bootflash: cat9k_gold.bin cat9k_prod.bin'), + Mock(match_output='flash: emergency.bin') + ] + + context = {} + session = {} + + boot_image(spawn, context, session) + + self.assertEqual( + spawn.sendline.call_args_list, + [ + unittest.mock.call('dir bootflash:'), + unittest.mock.call('dir flash:'), + unittest.mock.call('boot bootflash:cat9k_gold.bin') + ] + ) + self.assertEqual( + context['filesystem_images'], + ['bootflash:cat9k_prod.bin', 'flash:emergency.bin'] + ) + self.assertEqual(context['boot_prompt_count'], 2) + + def test_boot_image_uses_cached_filesystem_images_fifo_order(self): + spawn = Mock() + spawn.settings = SimpleNamespace( + MAX_BOOT_ATTEMPTS=5, + FIND_BOOT_IMAGE=True, + BOOT_FILESYSTEM=['bootflash:', 'flash:'], + BOOT_FILE_REGEX=r'(\S+\.bin)' + ) + spawn.expect.side_effect = [ + Mock(match_output='bootflash: cat9k_gold.bin cat9k_prod.bin'), + Mock(match_output='flash: emergency.bin') + ] + + context = {} + session = {} + + # First call builds the filesystem image list and boots first image. + boot_image(spawn, context, session) + + spawn.sendline.reset_mock() + spawn.expect.reset_mock() + + # Second call should use cached images in FIFO order without re-running dir. + boot_image(spawn, context, session) + + spawn.expect.assert_not_called() + spawn.sendline.assert_called_once_with('boot bootflash:cat9k_prod.bin') + self.assertEqual(context['filesystem_images'], ['flash:emergency.bin']) + self.assertEqual(context['boot_prompt_count'], 3) + + class TestIosXEPluginConnect(unittest.TestCase): def test_asr_login_connect(self): @@ -1710,6 +1777,111 @@ def test_acm_configure_invalid_input(self): c.disconnect() +class TestIosxeAcmInvalidCommand(unittest.TestCase): + + def test_invalid_command_configure(self): + """Test plain configure after failed ACM - should not enter ACM mode""" + c = Connection( + hostname='PE1', + start=['mock_device_cli --os iosxe --state general_enable --hostname PE1'], + os='iosxe', + mit=True + ) + c.connect() + + config_txt_fail = [ + 'interface loopback 1', + 'invalid command example', #invalid commad + ] + with patch.object(c.spawn, 'sendline', wraps=c.spawn.sendline) as mock_sendline_fail: + with self.assertRaises(SubCommandFailure): + c.configure(config_txt_fail, acm_configlet='my_config') + + sent_fail = [call.args[0] for call in mock_sendline_fail.call_args_list if call.args] + expected_fail = ['acm configlet create my_config', 'interface loopback 1', 'invalid command example', 'end'] + self.assertEqual(sent_fail, expected_fail, f"Expected commands {expected_fail}, but got {sent_fail}") + + config_txt_plain = ['interface loopback 2', 'description test'] + with patch.object(c.spawn, 'sendline', wraps=c.spawn.sendline) as mock_sendline_plain: + c.configure(config_txt_plain) + + sent_plain = [call.args[0] for call in mock_sendline_plain.call_args_list if call.args] + expected_plain = ['config term', 'interface loopback 2', 'description test', 'end'] + self.assertEqual(sent_plain, expected_plain, f"Expected commands {expected_plain}, but got {sent_plain}") + self.assertNotIn('acm configlet create my_config', sent_plain) + + c.disconnect() + + +class TestConfigureEmptySpacesHandling(unittest.TestCase): + + def test_configure_with_leading_trailing_spaces(self): + """Test configure commands with leading and trailing spaces""" + c = Connection( + hostname='PE1', + start=['mock_device_cli --os iosxe --state general_enable --hostname PE1'], + os='iosxe', + mit=True + ) + c.connect() + + config_txt = [ + " ", + " interface loopback 1 ", + "description test device", + " no shutdown " # Leading space + ] + + with patch.object(c.spawn, 'sendline', wraps=c.spawn.sendline) as mock_sendline: + c.configure(config_txt) + + sent_commands = [call.args[0] for call in mock_sendline.call_args_list if call.args] + # Validate exact expected commands with preserved leading/trailing spaces + expected_config = ['config term', ' ', ' interface loopback 1 ', 'description test device', ' no shutdown ', 'end'] + self.assertEqual(sent_commands, expected_config, f"Expected {expected_config}, but got {sent_commands}") + self.assertEqual( + c.state_machine.current_state, + 'enable', + f"Device should be in enable state, but is in {c.state_machine.current_state}" + ) + + c.disconnect() + +class TestBannerWithPreCommands(unittest.TestCase): + + def test_banner_login_with_pre_commands(self): + """Test banner login configuration with pre-commands before banner content""" + c = Connection( + hostname='PE1', + start=['mock_device_cli --os iosxe --state general_enable --hostname PE1'], + os='iosxe', + mit=True + ) + c.connect() + + config_txt = ( + "vrf definition Mgmt\n" + "banner login *\n" + "\n" + "Updated - Unauthorized access to this device is strictly prohibited.\n" + "\n" + "Please contact corpnet@fb.com with any access requests.\n" + "Done\n" + "\n" + "*" + ) + with patch.object(c.spawn, 'sendline', wraps=c.spawn.sendline) as mock_sendline: + c.configure(config_txt) + + sent_commands = [call.args[0] for call in mock_sendline.call_args_list if call.args] + # Check complete list of expected commands including empty lines + expected_commands = ['config term', 'vrf definition Mgmt', 'banner login *', '', + 'Updated - Unauthorized access to this device is strictly prohibited.', '', + 'Please contact corpnet@fb.com with any access requests.', 'Done', '', '*', 'end'] + self.assertEqual(sent_commands, expected_commands, f"Expected exact command sequence {expected_commands}, but got {sent_commands}") + + c.disconnect() + class TestIosxeSyntaxConfigure(unittest.TestCase): def test_syntax_configure(self): diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_c9500x_svl_designate_handles.py b/src/unicon/plugins/tests/test_plugin_iosxe_c9500x_svl_designate_handles.py new file mode 100644 index 00000000..8ebe00b0 --- /dev/null +++ b/src/unicon/plugins/tests/test_plugin_iosxe_c9500x_svl_designate_handles.py @@ -0,0 +1,62 @@ +""" +Unit tests for C9500X SVL designate_handles logic. +""" + +import unittest +from unittest.mock import MagicMock + +from unicon.plugins.iosxe.cat9k.c9500x.stackwise_virtual import ( + StackwiseVirtualConnectionProvider, +) + + +class TestIosxeC9500xSVLDesignateHandles(unittest.TestCase): + + def test_c9500x_svl_designate_handles(self): + con = MagicMock() + con.settings.BOOT_TIMEOUT = 1 + con.settings.EXEC_TIMEOUT = 1 + con.device = MagicMock() + con.device.parse.return_value = { + 'switch': { + 'stack': { + '1': {'role': 'active'}, + '2': {'role': 'standby'} + } + } + } + + redundancy_output = "my state = 8 -STANDBY HOT \n" + show_redundancy_cmd = "show redundancy states" + redundancy_state_pattern = r"my state = (.*?)\s*$" + + subcon_a = MagicMock() + subcon_a.alias = 'a' + subcon_a.state_machine.current_state = 'enable' + subcon_a.spawn.expect.return_value.match_output = redundancy_output + subcon_a.spawn.settings.SHOW_REDUNDANCY_CMD = show_redundancy_cmd + subcon_a.spawn.settings.REDUNDANCY_STATE_PATTERN = redundancy_state_pattern + + subcon_b = MagicMock() + subcon_b.alias = 'b' + subcon_b.state_machine.current_state = 'enable' + subcon_b.spawn.expect.return_value.match_output = redundancy_output + subcon_b.spawn.settings.SHOW_REDUNDANCY_CMD = show_redundancy_cmd + subcon_b.spawn.settings.REDUNDANCY_STATE_PATTERN = redundancy_state_pattern + + con._subconnections = {'a': subcon_a, 'b': subcon_b} + con.subconnections = [subcon_a, subcon_b] + con._set_active_alias = MagicMock() + con._set_standby_alias = MagicMock() + + provider = StackwiseVirtualConnectionProvider(con) + provider.designate_handles() + + con.device.parse.assert_called_with("show switch") + subcon_a.spawn.sendline.assert_called_with(show_redundancy_cmd) + self.assertEqual(con._set_active_alias.call_args_list[-1].args[0], 'b') + self.assertEqual(con._set_standby_alias.call_args_list[-1].args[0], 'a') + + +if __name__ == "__main__": + unittest.main() diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_c9500x_svl_reload.py b/src/unicon/plugins/tests/test_plugin_iosxe_c9500x_svl_reload.py new file mode 100644 index 00000000..d672285b --- /dev/null +++ b/src/unicon/plugins/tests/test_plugin_iosxe_c9500x_svl_reload.py @@ -0,0 +1,84 @@ +""" +Unittest for IOSXE/c9500x SVLStackReload service implementation. +""" +import unittest + +from unittest.mock import MagicMock, patch +from unicon.eal.dialogs import Dialog +from unicon.plugins.iosxe.cat9k.c9500x.stackwise_virtual.service_implementation import SVLStackReload + + +class TestSVLStackReload(unittest.TestCase): + """Test SVLStackReload service implementation.""" + + def _make_mock_connection(self): + """Return a MagicMock connection wired for SVLStackReload.""" + con = MagicMock() + con.hostname = 'Router' + con.connection_timeout = 60 + + settings = MagicMock() + settings.STACK_RELOAD_TIMEOUT = 900 + settings.STACK_POST_RELOAD_SLEEP = 0 + settings.POST_RELOAD_WAIT = 0 + settings.RELOAD_POSTCHECK_INTERVAL = 30 + settings.ERROR_PATTERN = [] + settings.LOGIN_PROMPT = 'Username:' + settings.PASSWORD_PROMPT = 'Password:' + con.settings = settings + + conn_active = MagicMock() + conn_active.spawn = MagicMock() + conn_active.sendline = conn_active.spawn.sendline + conn_active.context = {'hostname': 'Router'} + conn_active.alias = 'peer_1' + conn_active.hostname = 'Router' + conn_active.state_machine = MagicMock() + conn_active.state_machine.default_dialog = Dialog([]) + conn_active.settings = settings + con.active = conn_active + con.subconnections = [conn_active] + + con.connection_provider = MagicMock() + con.connection_provider.get_connection_dialog = MagicMock( + return_value=Dialog([])) + return con + + @patch('unicon.plugins.iosxe.cat9k.c9500x.stackwise_virtual' + '.service_implementation.utils') + @patch('unicon.plugins.iosxe.cat9k.c9500x.stackwise_virtual' + '.service_implementation.sleep') + @patch('unicon.eal.dialogs.Dialog.process', + return_value=MagicMock(match_output='')) + def test_svl_stack_reload_complete_flow(self, mock_process, + mock_sleep, mock_utils): + """Test complete flow of SVLStackReload service implementation.""" + mock_utils.is_active_standby_ready.return_value = True + con = self._make_mock_connection() + + svc = SVLStackReload(con, con.active.context) + svc.prompt_recovery = False + + svc.call_service(reload_command=None) + con.active.sendline.assert_called_with('redundancy reload shelf') + self.assertEqual(mock_process.call_count, 2) + go_to_calls = con.active.state_machine.go_to.call_args_list + self.assertEqual(go_to_calls[0][0][0], 'any') + self.assertEqual(go_to_calls[1][0][0], 'enable') + con.active.state_machine.detect_state.assert_not_called() + + # Standby readiness check with correct interval + mock_utils.is_active_standby_ready.assert_called_once() + + # post-reload sleep called + mock_sleep.assert_called() + con.disconnect.assert_called_once() + con.connect.assert_called_once() + con.connection_provider.init_connection.assert_not_called() + con.log.addHandler.assert_called() + con.log.removeHandler.assert_called() + self.assertTrue(svc.result) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_iec3400.py b/src/unicon/plugins/tests/test_plugin_iosxe_iec3400.py deleted file mode 100644 index 1b881cc8..00000000 --- a/src/unicon/plugins/tests/test_plugin_iosxe_iec3400.py +++ /dev/null @@ -1,56 +0,0 @@ -import unittest -from unittest.mock import patch - -import unicon -from unicon import Connection -from unicon.core.errors import SubCommandFailure -from unicon.eal.dialogs import Statement, Dialog - -unicon.settings.Settings.POST_DISCONNECT_WAIT_SEC = 0 -unicon.settings.Settings.GRACEFUL_DISCONNECT_WAIT_SEC = 0.2 - - -class TestIec3400Plugin(unittest.TestCase): - - def test_terminal_position_handler(self): - c = Connection( - hostname='PE1', - start=['mock_device_cli --os iosxe --state general_enable --hostname PE1'], - os='iosxe', - platform='iec3400' - ) - c.connect() - c.execute('get terminal position') - self.assertIn('^[[0;0R', c.spawn.match.match_output) - self.assertTrue(c.spawn.match.match_output.endswith('PE1#')) - c.disconnect() - - def test_reload_with_error_pattern(self): - - c = Connection( - hostname='PE1', - start=['mock_device_cli --os iosxe --state general_enable --hostname PE1'], - os='iosxe', - platform='iec3400' - ) - - install_add_one_shot_dialog = Dialog([ - Statement(pattern=r"FAILED:.* ", - action=None, - loop_continue=False, - continue_timer=False), - ]) - - error_pattern=[r"FAILED:.* ",] - try: - c.connect() - c.settings.POST_RELOAD_WAIT = 1 - c.reload('active_install_add', - reply=install_add_one_shot_dialog, - error_pattern=error_pattern) - self.assertEqual(c.reload.error_pattern, error_pattern) - finally: - c.disconnect() - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_svl_designate_handles.py b/src/unicon/plugins/tests/test_plugin_iosxe_svl_designate_handles.py new file mode 100644 index 00000000..eb770c79 --- /dev/null +++ b/src/unicon/plugins/tests/test_plugin_iosxe_svl_designate_handles.py @@ -0,0 +1,62 @@ +""" +Unit tests for SVL designate_handles logic. +""" + +import unittest +from unittest.mock import MagicMock + +from unicon.plugins.iosxe.cat9k.stackwise_virtual.connection_provider import ( + StackwiseVirtualConnectionProvider, +) + + +class TestIosxeSVLDesignateHandles(unittest.TestCase): + + def test_svl_designate_handles_9500(self): + con = MagicMock() + con.settings.BOOT_TIMEOUT = 1 + con.settings.EXEC_TIMEOUT = 1 + con.device = MagicMock() + con.device.parse.return_value = { + 'switch': { + 'stack': { + '1': {'role': 'active'}, + '2': {'role': 'standby'} + } + } + } + + redundancy_output = "my state = 8 -STANDBY HOT \n" + show_redundancy_cmd = "show redundancy states" + redundancy_state_pattern = r"my state = (.*?)\s*$" + + subcon_a = MagicMock() + subcon_a.alias = 'a' + subcon_a.state_machine.current_state = 'enable' + subcon_a.spawn.expect.return_value.match_output = redundancy_output + subcon_a.spawn.settings.SHOW_REDUNDANCY_CMD = show_redundancy_cmd + subcon_a.spawn.settings.REDUNDANCY_STATE_PATTERN = redundancy_state_pattern + + subcon_b = MagicMock() + subcon_b.alias = 'b' + subcon_b.state_machine.current_state = 'enable' + subcon_b.spawn.expect.return_value.match_output = redundancy_output + subcon_b.spawn.settings.SHOW_REDUNDANCY_CMD = show_redundancy_cmd + subcon_b.spawn.settings.REDUNDANCY_STATE_PATTERN = redundancy_state_pattern + + con._subconnections = {'a': subcon_a, 'b': subcon_b} + con.subconnections = [subcon_a, subcon_b] + con._set_active_alias = MagicMock() + con._set_standby_alias = MagicMock() + + provider = StackwiseVirtualConnectionProvider(con) + provider.designate_handles() + + con.device.parse.assert_called_with("show switch") + subcon_a.spawn.sendline.assert_called_with(show_redundancy_cmd) + self.assertEqual(con._set_active_alias.call_args_list[-1].args[0], 'b') + self.assertEqual(con._set_standby_alias.call_args_list[-1].args[0], 'a') + + +if __name__ == "__main__": + unittest.main()