From 55af2553836b8c3996282ee174be5712b6ab2f32 Mon Sep 17 00:00:00 2001 From: Thomas Ryan Date: Mon, 31 Aug 2026 14:38:52 -0400 Subject: [PATCH] Releasing v26.8 --- docs/changelog/2026/august.rst | 60 +++ docs/changelog/index.rst | 1 + docs/changelog_plugins/2026/august.rst | 55 +++ docs/changelog_plugins/index.rst | 1 + src/unicon/plugins/__init__.py | 2 +- src/unicon/plugins/generic/patterns.py | 4 +- src/unicon/plugins/generic/settings.py | 1 - src/unicon/plugins/generic/statements.py | 18 +- .../plugins/iosxe/cat8k/service_patterns.py | 5 +- .../plugins/iosxe/cat8k/service_statements.py | 12 +- src/unicon/plugins/iosxe/patterns.py | 1 + .../plugins/iosxe/service_implementation.py | 70 ++-- .../plugins/iosxe/service_statements.py | 7 +- .../iosxe/stack/connection_provider.py | 322 ++++++++++++++- .../iosxe/stack/service_implementation.py | 4 +- .../plugins/iosxe/stack/service_statements.py | 52 ++- src/unicon/plugins/iosxe/stack/settings.py | 30 +- src/unicon/plugins/iosxe/statemachine.py | 25 +- src/unicon/plugins/iosxe/statements.py | 51 ++- src/unicon/plugins/pid_tokens.csv | 1 + .../mock_data/iosxe/iosxe_mock_data.yaml | 46 +++ .../iosxe/iosxe_mock_data_asr1k_ha.yaml | 11 + .../iosxe_mock_data_cat9k_ha_reload.yaml | 7 + .../plugins/tests/test_plugin_generic.py | 17 +- src/unicon/plugins/tests/test_plugin_iosxe.py | 384 +++++++++++++++++- .../plugins/tests/test_plugin_iosxe_cat8k.py | 23 ++ .../plugins/tests/test_plugin_iosxe_cat9k.py | 32 ++ .../plugins/tests/test_plugin_iosxe_stack.py | 102 ++++- ...test_plugin_iosxe_stack_standby_console.py | 374 +++++++++++++++++ 29 files changed, 1640 insertions(+), 78 deletions(-) create mode 100644 docs/changelog/2026/august.rst create mode 100644 docs/changelog_plugins/2026/august.rst create mode 100644 src/unicon/plugins/tests/test_plugin_iosxe_stack_standby_console.py diff --git a/docs/changelog/2026/august.rst b/docs/changelog/2026/august.rst new file mode 100644 index 00000000..a24f1ecf --- /dev/null +++ b/docs/changelog/2026/august.rst @@ -0,0 +1,60 @@ +August 2026 +========== + +August 25 - Unicon v26.8 +------------------------ + + + +.. csv-table:: Module Versions + :header: "Modules", "Versions" + + ``unicon.plugins``, v26.8 + ``unicon``, v26.8 + + + + +Changelogs +^^^^^^^^^^ + +unicon +"""""" +-------------------------------------------------------------------------------- + Fix +-------------------------------------------------------------------------------- + +* unicon.bases.connection + * Modified Connection connect + * Clear stale previous_credential on the target context before the proxy/non-proxy branch so proxy_connections reconnects also restart credential selection from the first login_creds entry. + * Modified Connection proxy_connect + * Clear stale previous_credential on each proxy hop before its connection provider runs, so every hop of a multi-hop proxy chain restarts credential selection. Only hop 0 goes through setup_connection(); hops 1 and beyond are connected directly and previously retained relogin state from a prior session. + * Added Connection reset_previous_credential + * Helper that clears previous_credential on a connection and its subconnections. + +* unicon.bases.routers.connection + * Modified BaseSingleRpConnection and BaseMultiRpConnection setup_connection + * Clear stale previous_credential (parent and subconnection contexts) at the fresh-spawn boundary so every reconnect restarts credential selection from the first login_creds entry (e.g. terminal server), even for plugins that override disconnect() (hvrp, gaia, fxos/ftd). + +* unicon.bases.routers.connection_provider + * Modified BaseSingleRpConnectionProvider establish_connection + * Store the credential that actually answered the most recent password prompt as current_credentials, so a terminal-server post action that reaches the device prompt before later login_creds entries are consumed selects the correct credential. + * Modified disconnect cleanup + * Ensured all spawns are closed when sending the exit command fails. + +* unicon.bases + * Modified Connection, BaseMultiRpConnection, and BaseStackRpConnectionProvider + * Moved clear-console retries outside connection dialog callbacks so simultaneous stack member refusals are recovered once and retried from a fresh set of spawns. + * Avoided a redundant second stack state detection after a member has already reached a valid non-disable prompt. + * Preserved the second detection for disable prompts that may actually represent a standby-locked stack member. + * Moved the ``CONNECTION_REFUSED_MAX_COUNT`` default into core settings with the retry logic that consumes it. + * Preserved the originating device for proxy connection refusals so recovery never clears an unrelated target console. + * Preserved the originating proxy device while setting up multi-RP subconnections. + * Preserved the final target device when a refusal occurs after connecting through a proxy. + * Modified connection cleanup + * Prevented failed connections and reconnects from retaining stale PTY-backed spawns. + * Removed temporary connection log handlers after successful and failed connection attempts. + +* unicon.eal.backend.pty_backend + * Modified Spawn close + * Made PTY cleanup idempotent and added bounded child-process reaping. diff --git a/docs/changelog/index.rst b/docs/changelog/index.rst index 2f2fdc2f..3e4f2054 100644 --- a/docs/changelog/index.rst +++ b/docs/changelog/index.rst @@ -4,6 +4,7 @@ Changelog .. toctree:: :maxdepth: 2 + 2026/august 2026/july 2026/june 2026/may diff --git a/docs/changelog_plugins/2026/august.rst b/docs/changelog_plugins/2026/august.rst new file mode 100644 index 00000000..24294716 --- /dev/null +++ b/docs/changelog_plugins/2026/august.rst @@ -0,0 +1,55 @@ +August 2026 +========== + +August 25 - Unicon.Plugins v26.8 +-------------------------------- + + + +.. csv-table:: Module Versions + :header: "Modules", "Versions" + + ``unicon.plugins``, v26.8 + ``unicon``, v26.8 + + + + +Changelogs +^^^^^^^^^^ + +-------------------------------------------------------------------------------- + Fix +-------------------------------------------------------------------------------- + +* iosxe + * Updated the shared single-RP and dual-RP syntax transition to use ``config check syntax`` so syntax checking works consistently on IOS XE and CAT9k HA devices. + * Updated ``configure(config_syntax_check=True)`` and ``config_syntax()`` to enter syntax-check mode, validate the supplied commands, and return to enable mode. + * Removed the obsolete ``syntax_configlet`` handling and unregistered ``Syntaxsh`` service. + * Modified state machine + * Answer the uncommitted changes prompt when leaving configuration mode on controller-managed single- and dual-RP devices. + * Updated CAT8K reload-to-ROMMON handling to recognize ``System Bootstrap`` output. + * Saves the environment when prompted and sends Ctrl-C at the boot interruption point. + * Added the ``C8300-2N2S-4T2X`` PID mapping for CAT8K C8300 plugin selection. + * Fixed ROMMON boot dialogs to invoke wrapped statement callbacks according to their declared parameters. + * Copied generic connection statements before applying IOS XE-specific callback wrapping. + * Updated the configure service to automatically answer the interactive RSA key modulus prompt with 2048 bits while allowing a caller-provided reply to override the default. + * Added regression and mock-device coverage to verify RSA key generation completes without a caller-provided dialog. + +* iosxe/stack + * Modified init_standby + * Added STACK_STANDBY_CONSOLE_ATTACH setting, off by default. Enable it on stacks where every console connects to the active RP. + * Verify which console is on the standby RP using the '*' row of 'show switch' instead of assuming the designated standby handle. + * Attach to the standby RP console when no console reaches it, so device.standby, target='standby' and swap_roles() no longer operate on the active RP without reporting an error. + * Release a console attach left over from an earlier run before probing, so a stale attach is not mistaken for a real standby. + * Modified disconnect + * Detach the standby RP console so the next connect is not refused with 'Address already in use'. + * Modified set_init_commands + * Skip 'show version' on an attached standby console, whose output is rate limited and times out the connect. + * Added stack-specific reload handling for the transition from ``Press RETURN`` to an authentication prompt. The stack reload dialog now avoids an empty username and completes authentication in the same dialog flow. + +* generic + * Updated the syslog message pattern to match ``UNIX ERR:tcgetattr:Inappropriate ioctl for device`` output so it is handled as ignorable syslog noise. + * Modified connection_refused_handler + * Propagate a typed ``ConnectionRefusedError`` to Unicon core instead of clearing terminal lines and reconnecting recursively from a connection dialog callback. + * Removed the plugin-owned ``CONNECTION_REFUSED_MAX_COUNT`` default; it is now defined alongside the retry logic in Unicon core. diff --git a/docs/changelog_plugins/index.rst b/docs/changelog_plugins/index.rst index 2c7a1219..c8d53822 100644 --- a/docs/changelog_plugins/index.rst +++ b/docs/changelog_plugins/index.rst @@ -4,6 +4,7 @@ Plugins Changelog .. toctree:: :maxdepth: 2 + 2026/august 2026/july 2026/june 2026/may diff --git a/src/unicon/plugins/__init__.py b/src/unicon/plugins/__init__.py index 2f549f98..4bc77ea3 100644 --- a/src/unicon/plugins/__init__.py +++ b/src/unicon/plugins/__init__.py @@ -1,4 +1,4 @@ -__version__ = "26.7" +__version__ = "26.8" supported_chassis = [ 'single_rp', diff --git a/src/unicon/plugins/generic/patterns.py b/src/unicon/plugins/generic/patterns.py index a917a97d..e12cf270 100644 --- a/src/unicon/plugins/generic/patterns.py +++ b/src/unicon/plugins/generic/patterns.py @@ -76,6 +76,7 @@ def __init__(self): # % WARNING: The master key is not configured, so passwords/secrets might not be encrypted. # Configure the master key by using the following command: "key config-key password-encrypt # Applying config on Switch 3...[DONE] + # UNIX ERR:tcgetattr:Inappropriate ioctl for device self.syslog_message_pattern = ( r"^.*?(%\w+(-\S+)?-\d+-\w+|" r"yang-infra:|PKI_SSL_IPC:|Guestshell destroyed successfully|" @@ -86,7 +87,8 @@ def __init__(self): r"Failed to generate persistent self-signed certificate\.|" r"Secure server will use temporary self-signed certificate\.|" r"(LC|RP)/\d+/\d+/CPU\d+:\w+\s+\d+\s+\d{2}:\d{2}:\d{2}|" - r"\[OK\]" + r"\[OK\]|" + r"UNIX ERR:tcgetattr:Inappropriate ioctl for device" r").*\s*$" ) self.config_locked = r'Configuration (mode )?(is )?locked|Config mode cannot be entered' diff --git a/src/unicon/plugins/generic/settings.py b/src/unicon/plugins/generic/settings.py index e0e84d6a..fbc0c15a 100644 --- a/src/unicon/plugins/generic/settings.py +++ b/src/unicon/plugins/generic/settings.py @@ -55,7 +55,6 @@ def __init__(self): self.CONSOLE_TIMEOUT = 60 self.BOOT_TIMEOUT = 600 self.MAX_BOOT_ATTEMPTS = 3 - self.CONNECTION_REFUSED_MAX_COUNT = 3 # Temporary enable secret used during setup # this is used if no password is available diff --git a/src/unicon/plugins/generic/statements.py b/src/unicon/plugins/generic/statements.py index d59a526c..f1251997 100644 --- a/src/unicon/plugins/generic/statements.py +++ b/src/unicon/plugins/generic/statements.py @@ -46,18 +46,12 @@ def terminal_position_handler(spawn, session, context): def connection_refused_handler(spawn, context): """ handles connection refused scenarios """ - context.setdefault('connection_refused_count', 0) - context['connection_refused_count'] += 1 - if context.get('connection_refused_count') < spawn.settings.CONNECTION_REFUSED_MAX_COUNT: - if spawn.device: - spawn.device.api.execute_clear_line() - spawn.device.connect() - return - spawn.log.error( - 'Too many connection refused events: {}, update setting ' - 'CONNECTION_REFUSED_MAX_COUNT as needed.'.format( - context['connection_refused_count'])) - raise Exception('Connection refused to device %s' % (str(spawn))) + error = ConnectionRefusedError( + 'Connection refused to device %s' % str(spawn)) + error.spawn = spawn + error.context = context + error.device = getattr(spawn, 'device', None) + raise error def connection_failure_handler(spawn): diff --git a/src/unicon/plugins/iosxe/cat8k/service_patterns.py b/src/unicon/plugins/iosxe/cat8k/service_patterns.py index 51f9e420..3d2bc13e 100644 --- a/src/unicon/plugins/iosxe/cat8k/service_patterns.py +++ b/src/unicon/plugins/iosxe/cat8k/service_patterns.py @@ -5,7 +5,10 @@ class ReloadPatterns(IosXEPatterns): def __init__(self): super().__init__() - self.boot_interrupt_prompt = r'Preparing to autoboot. \[Press Ctrl-C to interrupt\]' + self.boot_interrupt_prompt = ( + r'(?:Preparing to autoboot\. \[Press Ctrl-C to interrupt\]|' + r'System Bootstrap, Version .*)' + ) class SwitchoverPatterns: def __init__(self): self.save_config = r'.*System configuration has been modified\.\s*Save\?\s*\[yes\/no\]:\s*$' diff --git a/src/unicon/plugins/iosxe/cat8k/service_statements.py b/src/unicon/plugins/iosxe/cat8k/service_statements.py index d132566d..a0d1b1e4 100644 --- a/src/unicon/plugins/iosxe/cat8k/service_statements.py +++ b/src/unicon/plugins/iosxe/cat8k/service_statements.py @@ -2,8 +2,9 @@ from unicon.eal.dialogs import Statement from unicon.plugins.iosxe.cat8k.service_patterns import SwitchoverPatterns, ReloadPatterns +from unicon.plugins.generic.service_patterns import reload_patterns from unicon.plugins.generic.service_statements import ( - save_env, confirm_reset, reload_confirm, reload_confirm_ios) + confirm_reset, reload_confirm, reload_confirm_ios) @@ -42,6 +43,13 @@ ############################################################################# patterns = ReloadPatterns() +save_env_to_rommon = Statement( + pattern=reload_patterns.savenv, + action='sendline(yes)', + args=None, + loop_continue=True, + continue_timer=False) + boot_interrupt_stmt = Statement( pattern=patterns.boot_interrupt_prompt, action='send(\x03)', @@ -50,7 +58,7 @@ continue_timer=False) -reload_to_rommon_statement_list = [save_env, +reload_to_rommon_statement_list = [save_env_to_rommon, confirm_reset, reload_confirm, reload_confirm_ios, diff --git a/src/unicon/plugins/iosxe/patterns.py b/src/unicon/plugins/iosxe/patterns.py index aaf1eecd..4bf13ce0 100644 --- a/src/unicon/plugins/iosxe/patterns.py +++ b/src/unicon/plugins/iosxe/patterns.py @@ -46,6 +46,7 @@ def __init__(self): self.acm_prompt = r'^(.*?)\(acm.*?\)#[\s\x07]*$' self.syntax_prompt = r'^(.*?)\(syntax.*?\)#[\s\x07]*$' self.rules_prompt = r'^(.*?)\(rules.*?\)#[\s\x07]*$' + self.rsa_modulus_prompt = r'^.*How many bits in the modulus \[\d+\]:\s*$' class IosXEReloadPatterns(ReloadPatterns): diff --git a/src/unicon/plugins/iosxe/service_implementation.py b/src/unicon/plugins/iosxe/service_implementation.py index 05454d2f..022ab092 100644 --- a/src/unicon/plugins/iosxe/service_implementation.py +++ b/src/unicon/plugins/iosxe/service_implementation.py @@ -27,6 +27,7 @@ execute_statement_list, configure_statement_list, confirm, + rsa_modulus, want_continue, ) @@ -36,6 +37,13 @@ from unicon.plugins.generic.service_implementation import BashService as GenericBashService +def _with_rsa_modulus_fallback(reply): + """Append the default RSA response after any caller-provided statements.""" + if isinstance(reply, Dialog): + return reply + Dialog([rsa_modulus]) + return reply + + # Simplex Services # ---------------- class Configure(GenericConfigure): @@ -61,10 +69,19 @@ def truncate_trailing_prompt(self, con_state, self.utils = ConfigUtils() + def call_service(self, command=[], reply=Dialog([]), *args, **kwargs): + reply = _with_rsa_modulus_fallback(reply) + return super().call_service(command, reply, *args, **kwargs) + def pre_service(self, *args, **kwargs): self.acm_configlet = kwargs.pop('acm_configlet', None) - self.syntax_configlet = kwargs.pop('syntax_configlet', None) + syntax_configlet = kwargs.pop('syntax_configlet', None) + if syntax_configlet: + raise SubCommandFailure( + "'syntax_configlet' is no longer supported. " + "Use configure(..., config_syntax_check=True) " + "or config_syntax() instead.") self.config_syntax_check = kwargs.pop('config_syntax_check', False) self.rules = kwargs.pop('rules', False) self.prompt_recovery = kwargs.get('prompt_recovery', True) @@ -84,12 +101,13 @@ def pre_service(self, *args, **kwargs): self.start_state = 'rules' self.end_state = 'rules' - elif self.syntax_configlet or self.config_syntax_check: - configlet_name = self.syntax_configlet if self.syntax_configlet else '' - self.connection.state_machine.go_to('syntax', self.connection.spawn, - context={'syntax_configlet': configlet_name}) + elif self.config_syntax_check: + target = kwargs.get('target') if self.connection.is_ha else None + handle = self.get_handle(target) + handle.state_machine.go_to( + 'syntax', handle.spawn, context=handle.context) self.start_state = 'syntax' - self.end_state = 'syntax' + self.end_state = 'enable' else: super().pre_service(*args, **kwargs) @@ -109,10 +127,9 @@ class Config(Configure): class ConfigSyntax(Configure): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.start_state = 'syntax' - self.end_state = 'enable' + def pre_service(self, *args, **kwargs): + kwargs['config_syntax_check'] = True + super().pre_service(*args, **kwargs) class Execute(GenericExecute): @@ -155,9 +172,19 @@ def __init__(self, connection, context, **kwargs): super().__init__(connection, context, **kwargs) self.dialog += Dialog(configure_statement_list) + def call_service(self, command=[], reply=Dialog([]), *args, **kwargs): + reply = _with_rsa_modulus_fallback(reply) + return super().call_service(command, reply, *args, **kwargs) + def pre_service(self, *args, **kwargs): self.acm_configlet = kwargs.pop('acm_configlet', None) - self.syntax_configlet = kwargs.pop('syntax_configlet', None) + syntax_configlet = kwargs.pop('syntax_configlet', None) + if syntax_configlet: + raise SubCommandFailure( + "'syntax_configlet' is no longer supported. " + "Use configure(..., config_syntax_check=True) " + "or config_syntax() instead.") + self.config_syntax_check = kwargs.pop('config_syntax_check', False) self.rules = kwargs.pop('rules', False) self.prompt_recovery = kwargs.get('prompt_recovery', True) @@ -166,10 +193,12 @@ def pre_service(self, *args, **kwargs): self.start_state = 'acm' self.end_state = 'acm' - if self.syntax_configlet: - self.connection.state_machine.go_to('syntax', self.connection.spawn,context={'syntax_configlet': self.syntax_configlet}) + if self.config_syntax_check: + handle = self.get_handle(kwargs.get('target')) + handle.state_machine.go_to( + 'syntax', handle.spawn, context=handle.context) self.start_state = 'syntax' - self.end_state = 'syntax' + self.end_state = 'enable' elif self.rules: if self.connection.connected: self.connection.state_machine.go_to('rules', self.connection.spawn) @@ -489,19 +518,6 @@ def __init__(self, connection, context, **kwargs): self.service_name = 'tclsh' self.__dict__.update(kwargs) -class Syntaxsh(BaseService): - - def __init__(self, connection, context, **kwargs): - super().__init__(connection, context, **kwargs) - self.start_state = 'enable' - self.end_state = 'syntax_check' - self.service_name = 'syntax_check' - - def call_service(self, syntax_file=None, **kwargs): - cmd = f"syntax configlet check {syntax_file}" if syntax_file else "syntax configlet check" - self.connection.spawn.sendline(cmd) - self.connection.state_machine.go_to('syntax_check', self.connection.spawn, **kwargs) - class MaintenanceMode(ContextMgrBaseService): def __init__(self, connection, context, **kwargs): diff --git a/src/unicon/plugins/iosxe/service_statements.py b/src/unicon/plugins/iosxe/service_statements.py index 4aad2275..73d284df 100644 --- a/src/unicon/plugins/iosxe/service_statements.py +++ b/src/unicon/plugins/iosxe/service_statements.py @@ -82,6 +82,11 @@ def send_response(spawn, response=""): macro_prompt = Statement(pattern=patterns.macro_prompt, loop_continue=False) +rsa_modulus = Statement(pattern=patterns.rsa_modulus_prompt, + action='sendline(2048)', + loop_continue=True, + continue_timer=False) + configure_statement_list = [ are_you_sure, @@ -117,4 +122,4 @@ def send_response(spawn, response=""): are_you_sure_confirm = Statement(pattern=pat.are_you_sure_confirm, action=send_response, args={'response': ''}, loop_continue=True, - continue_timer=False) \ No newline at end of file + continue_timer=False) diff --git a/src/unicon/plugins/iosxe/stack/connection_provider.py b/src/unicon/plugins/iosxe/stack/connection_provider.py index 5d78dc5a..caeed128 100644 --- a/src/unicon/plugins/iosxe/stack/connection_provider.py +++ b/src/unicon/plugins/iosxe/stack/connection_provider.py @@ -3,10 +3,47 @@ pyATS TEAM (pyats-support@cisco.com, pyats-support-ext@cisco.com) """ -from unicon.eal.dialogs import Dialog +import re +import time + +from unicon.eal.dialogs import Dialog, Statement from unicon.bases.routers.connection_provider import BaseStackRpConnectionProvider -from unicon.plugins.generic.statements import connection_statement_list, custom_auth_statements +from unicon.plugins.generic.statements import (connection_statement_list, + custom_auth_statements, + GenericStatements) + +statements = GenericStatements() + + +##show switch +# *1 Active 74e2.e72d.8180 1 V07 Ready +# 3 Standby 74e2.e72d.9080 1 V07 Ready +SHOW_SWITCH_ROW_RE = re.compile( + r'^\s*(?P\*)?\s*(?P\d+)\s+' + r'(?PActive|Standby|Member)\b', re.IGNORECASE | re.MULTILINE) + +# Only one session may hold the standby console TTY; a stale attach elsewhere +# makes the device answer 'ioucon:TTY open (remote): Address already in use'. +CONSOLE_BUSY_PATTERN = r'.*(Address already in use|TTY open.*in use).*' + +# The attached console stays silent after this banner until it gets a newline. +CONSOLE_BANNER_PATTERN = r'.*Enter Control-C to exit.*' + + +def _console_busy_handler(spawn, context): + """Mark the attach as refused so we can fail fast instead of timing out.""" + context['_standby_console_busy'] = True + + +def _console_banner_handler(spawn, context): + """Poke the attached console so it prints its prompt.""" + spawn.sendline() + + +def _console_rejected_handler(spawn, context): + """Mark the attach as not taken, so we stop instead of waiting it out.""" + context['_standby_console_rejected'] = True class StackRpConnectionProvider(BaseStackRpConnectionProvider): @@ -33,3 +70,284 @@ def get_connection_dialog(self): return con.connect_reply + \ Dialog(custom_auth_stmt + connection_statement_list if custom_auth_stmt else connection_statement_list) + + # ------------------------------------------------------------------ + # Standby console attach + # ------------------------------------------------------------------ + def _get_local_switch(self, con, timeout=30): + """Return (switch number, role) from the '*' row of 'show switch'. + + The '*' row is console-local, unlike 'show switch '. + """ + try: + output = con.execute('show switch', timeout=timeout) + except Exception as exc: + con.log.debug('Could not read stack roles: %s' % exc) + return None, None + + for match in SHOW_SWITCH_ROW_RE.finditer(output or ''): + if match.group('local'): + return match.group('sw_num'), match.group('role').capitalize() + return None, None + + def _attach_standby_console(self, con): + """Descend from an active-RP session into the standby RP console. + + Runs 'request platform software console attach switch standby ', + answers the login it raises, and leaves the handle in enable state. + """ + settings = self.connection.settings + plane = getattr(settings, 'STACK_STANDBY_CONSOLE_PLANE', 'R0') + timeout = getattr(settings, + 'STACK_STANDBY_CONSOLE_ATTACH_TIMEOUT', 120) + + command = ('request platform software console attach switch standby %s' + % plane) + con.log.info('+++ attaching to standby RP console +++') + + # Make sure we are at enable on the active before descending. + con.state_machine.go_to('enable', + con.spawn, + context=con.context, + prompt_recovery=self.prompt_recovery, + timeout=con.connection_timeout) + + # The attach banner is followed by a fresh 'User Access Verification' + # login. Reuse the standard auth statements so testbed credentials are + # applied, and tolerate the platform's informational/warning lines. + custom_auth_stmt = custom_auth_statements(settings.LOGIN_PROMPT, + settings.PASSWORD_PROMPT) + busy_stmt = Statement(pattern=CONSOLE_BUSY_PATTERN, + action=_console_busy_handler, + args=None, + loop_continue=False, + continue_timer=False) + banner_stmt = Statement(pattern=CONSOLE_BANNER_PATTERN, + action=_console_banner_handler, + args=None, + loop_continue=True, + continue_timer=True) + + # Terminate only on the standby prompt ('-stby>' or '-stby#'). + # The pattern requires '-stby' so it cannot match the active's prompt + # if the attach fails or is rejected. + hostname = con.context.get('hostname') or con.hostname + prompt_stmt = Statement(pattern=r'^.*%s[\w\-]*-stby[>#]\s?$' + % re.escape(hostname), + action=None, + args=None, + loop_continue=False, + continue_timer=False) + + # Landing back on the active's prompt means the attach was rejected or + # is unsupported; stop rather than poking until the timeout expires. + rejected_stmt = Statement(pattern=r'^.*%s(?#]\s?$' + % re.escape(hostname), + action=_console_rejected_handler, + args=None, + loop_continue=False, + continue_timer=False) + + dialog = Dialog((custom_auth_stmt or []) + [ + busy_stmt, + banner_stmt, + statements.login_stmt, + statements.password_stmt, + statements.enable_password_stmt, + statements.bad_password_stmt, + statements.syslog_stripper_stmt, + statements.syslog_msg_stmt, + prompt_stmt, + rejected_stmt, + ]) + + con.context.pop('_standby_console_busy', None) + con.context.pop('_standby_console_rejected', None) + con.spawn.buffer = '' + con.sendline(command) + + # The attached console prints nothing until it is poked with a newline, + # and a poke sent before it is ready is swallowed, so poke repeatedly + # with a short timeout rather than waiting out one long one. Prompt + # recovery is off here because its Ctrl-C would exit the very console + # being attached to. + poll = getattr(settings, 'STACK_STANDBY_CONSOLE_POLL_INTERVAL', 20) + deadline = time.time() + timeout + while True: + try: + dialog.process(con.spawn, + context=con.context, + prompt_recovery=False, + timeout=min(poll, + max(1, deadline - time.time()))) + break + except Exception as exc: + if con.context.get('_standby_console_busy') or \ + con.context.get('_standby_console_rejected'): + break + if time.time() >= deadline: + con.log.warning( + 'Standby console attach dialog ended: %s' % exc) + break + con.spawn.sendline() + + if con.context.pop('_standby_console_rejected', False): + con.log.error( + 'Standby console attach was not accepted; the session is still ' + 'on the active RP. device.standby and swap_roles() will not ' + 'reach the standby.') + con.state_machine.detect_state(con.spawn, context=con.context) + return + + if con.context.pop('_standby_console_busy', False): + con.log.error( + 'Standby console is already in use by another session ' + '("Address already in use"). Only one session may attach to ' + 'the standby RP console at a time. Close any manual telnet ' + 'session that ran "request platform software console attach" ' + '(send Ctrl-C to exit it), or clear the terminal-server line, ' + 'then reconnect.') + # Leave the handle on the active in a known state. + con.state_machine.detect_state(con.spawn, context=con.context) + return + + # The attached session starts in disable mode ('-stby>'). The + # iosxe disable/enable prompt patterns already tolerate the '-stby' + # suffix, so a normal state detection + go_to('enable') completes it. + con.state_machine.detect_state(con.spawn, context=con.context) + con.state_machine.go_to('enable', + con.spawn, + context=con.context, + prompt_recovery=self.prompt_recovery, + timeout=con.connection_timeout) + # Remember so disconnect() can release the console for the next run. + con._standby_console_attached = True + + def _release_stale_console_attach(self, con): + """Exit a console attach left behind by an earlier run. + + The attach lives on the terminal-server line, not the telnet session, + so it survives a disconnect and reconnecting lands back in '-stby#'. + Left in place, the probe below reads STANDBY and skips the attach -- + a false green. Ctrl-C is harmless at a normal prompt. + """ + try: + con.spawn.send('\x03') + time.sleep(2) + con.spawn.sendline() + time.sleep(1) + con.spawn.buffer = '' + con.state_machine.detect_state(con.spawn, context=con.context) + except Exception as exc: + con.log.debug('Stale console attach release skipped: %s' % exc) + finally: + con._standby_console_attached = False + + def _detach_standby_console(self, con): + """Ctrl-C out of the standby console so the next connect can attach. + + Only one session may hold the standby TTY. + """ + try: + con.log.info('+++ detaching from standby RP console +++') + con.spawn.send('\x03') + time.sleep(2) + con.spawn.sendline() + time.sleep(1) + con.spawn.buffer = '' + except Exception as exc: + con.log.warning('Failed to detach standby console: %s' % exc) + finally: + con._standby_console_attached = False + + def disconnect(self): + """Release the standby console (if we attached it) before disconnecting.""" + for subcon in getattr(self.connection, 'subconnections', []) or []: + if getattr(subcon, '_standby_console_attached', False): + self._detach_standby_console(subcon) + super().disconnect() + + def init_standby(self): + """Initialise the standby handle, attaching its console if needed. + + On a stack every member console connects to the active, so the + designated standby handle is a second session on the active and + device.standby / target='standby' / swap_roles() silently use it. + """ + con = self.connection + sby_con = con.standby + settings = con.settings + + if getattr(settings, 'STACK_STANDBY_CONSOLE_ATTACH', False): + standby_alias = self._find_standby_console() + + if standby_alias is not None: + con.log.info( + "Connection '%s' is on the standby RP; no console attach " + "needed." % standby_alias) + if standby_alias != getattr(sby_con, 'alias', None): + con.log.warning( + "Standby console is '%s' but the standby handle is " + "'%s'. device.standby will not reach the standby RP." + % (standby_alias, getattr(sby_con, 'alias', None))) + else: + sby_con.log.info( + 'No console reaches the standby RP; attaching to the ' + 'standby console.') + self._attach_standby_console(sby_con) + + # The first read straight after the attach can come back empty + # because the session has just switched prompts; retry once. + sw_num, role = self._get_local_switch(sby_con) + if role is None: + sw_num, role = self._get_local_switch(sby_con) + + if role == 'Standby': + sby_con.log.info( + 'Standby console attach successful (switch %s).' + % sw_num) + else: + sby_con.log.warning( + 'Standby console attach did not reach the standby RP ' + '(role = %s).' % role) + + super().init_standby() + + def set_init_commands(self): + """Drop init commands the attached standby console cannot sustain. + + The standby RP console is rate limited: 'show version' stalls part way + and times out the connect. The active handle keeps the full set. + """ + super().set_init_commands() + + sby_con = getattr(self.connection, 'standby', None) + if not getattr(sby_con, '_standby_console_attached', False): + return + + skip = getattr(self.connection.settings, + 'STACK_STANDBY_CONSOLE_SKIP_INIT_COMMANDS', []) + if skip: + self.init_exec_commands = [cmd for cmd in self.init_exec_commands + if cmd not in skip] + + def _find_standby_console(self): + """Return the alias of a console on the standby RP, else None. + + Probes every console, not just the designated standby handle: the + designation is positional and the standby can be any switch. Stale + attaches are cleared first so each console reports its own identity. + """ + con = self.connection + subcons = list(getattr(con, '_subconnections', {}).items()) or \ + [(getattr(con.standby, 'alias', 'standby'), con.standby)] + + for alias, subcon in subcons: + self._release_stale_console_attach(subcon) + + sw_num, role = self._get_local_switch(subcon) + subcon.log.info("Connection '%s' is on switch %s (%s)" + % (alias, sw_num, role)) + if role == 'Standby': + return alias + return None diff --git a/src/unicon/plugins/iosxe/stack/service_implementation.py b/src/unicon/plugins/iosxe/stack/service_implementation.py index b59b5897..cfe43d05 100644 --- a/src/unicon/plugins/iosxe/stack/service_implementation.py +++ b/src/unicon/plugins/iosxe/stack/service_implementation.py @@ -16,7 +16,7 @@ from unicon.plugins.generic.statements import custom_auth_statements, buffer_settled from unicon.plugins.generic.service_statements import standby_reset_rp_statement_list from .service_statements import (switch_prompt, - stack_reload_stmt_list, + stack_reload_auth_stmt_list, stack_reload_stmt_list_1, stack_switchover_stmt_list, stack_factory_reset_stmt_list) from unicon.plugins.generic.service_implementation import Enable as GenericEnable, Execute as GenericExecute @@ -219,7 +219,7 @@ def __init__(self, connection, context, *args, **kwargs): self.timeout = connection.settings.STACK_RELOAD_TIMEOUT self.reload_command = "redundancy reload shelf" self.log_buffer = io.StringIO() - self.dialog = Dialog(stack_reload_stmt_list) + self.dialog = Dialog(stack_reload_auth_stmt_list) def call_service(self, reload_command=None, diff --git a/src/unicon/plugins/iosxe/stack/service_statements.py b/src/unicon/plugins/iosxe/stack/service_statements.py index b2494ef4..150d84cf 100644 --- a/src/unicon/plugins/iosxe/stack/service_statements.py +++ b/src/unicon/plugins/iosxe/stack/service_statements.py @@ -1,4 +1,7 @@ """ Generic IOS-XE Stack Service Statements """ +import re +from datetime import datetime, timedelta + from unicon.eal.dialogs import Statement from unicon.plugins.generic.service_statements import (reload_statement_list, @@ -6,7 +9,9 @@ reload_confirm_ios, reload_confirm_iosxe, reload_entire_shelf, - reload_this_shelf) + reload_this_shelf, + press_enter, + press_return) from unicon.plugins.iosxe.service_statements import (factory_reset_confirm, are_you_sure_confirm) @@ -123,6 +128,42 @@ def send_boot_cmd(spawn, context): # reload service statements reload_pat = StackIosXEReloadPatterns() + +def stack_wait_and_enter(spawn, wait=0.5): + """Handle the stack reload ``Press RETURN``/authentication prompt race.""" + last_match = getattr(getattr(spawn, 'match', None), 'last_match', None) + match_start = last_match.start() if last_match is not None else 0 + + wait_time = timedelta(seconds=wait) + settle_time = current_time = datetime.now() + while (current_time - settle_time) < wait_time: + spawn.read_update_buffer() + current_time = datetime.now() + + settings = getattr(spawn, 'settings', None) + auth_patterns = ( + reload_pat.username, + reload_pat.password, + getattr(settings, 'LOGIN_PROMPT', None), + getattr(settings, 'PASSWORD_PROMPT', None), + ) + pending_output = (spawn.buffer or '')[match_start:] + if any(pattern and re.search(pattern, pending_output, re.DOTALL) + for pattern in auth_patterns): + spawn.log.debug( + 'Authentication prompt appeared while waiting; skipping RETURN') + return + + spawn.sendline() + + +stack_press_return = Statement( + pattern=reload_pat.press_return, + action=stack_wait_and_enter, + loop_continue=True, + continue_timer=False, +) + reload_shelf = Statement(pattern=reload_pat.reload_entire_shelf, action='sendline()', loop_continue=True, @@ -156,6 +197,7 @@ def send_boot_cmd(spawn, context): accelarating_discovery, fastreload_iosxeswitch, proceed_prompt_stmt] +# Some stack variants process each "Press RETURN" in separate dialog calls. stack_reload_stmt_list = list(reload_statement_list) # The enable and disable states are needed when using `reload slot N` @@ -163,6 +205,14 @@ def send_boot_cmd(spawn, context): stack_reload_stmt_list.insert(0, reload_shelf) stack_reload_stmt_list.insert(0, reload_fast) +# Both generic statements match "Press RETURN". Keep one stack handler active +# so authentication prompts that arrive while settling stay in this dialog. +stack_reload_auth_stmt_list = [ + stack_press_return if statement is press_enter else statement + for statement in stack_reload_stmt_list + if statement is not press_return +] + stack_factory_reset_stmt_list = [factory_reset_confirm, are_you_sure_confirm] diff --git a/src/unicon/plugins/iosxe/stack/settings.py b/src/unicon/plugins/iosxe/stack/settings.py index 403bda71..3dce238c 100644 --- a/src/unicon/plugins/iosxe/stack/settings.py +++ b/src/unicon/plugins/iosxe/stack/settings.py @@ -27,4 +27,32 @@ def __init__(self): self.CONFIGURE_ALLOW_STATE_CHANGE = True # Secs to sleep after booting the device - self.STACK_ENABLE_SLEEP = 100 \ No newline at end of file + self.STACK_ENABLE_SLEEP = 100 + + # Some stack topologies expose only the active RP's console (all + # terminal-server lines land on the active). In that case the + # designated standby handle is really another session on the active, + # so device.standby / target='standby' / swap_roles() cannot reach the + # standby RP. + # + # When enabled, init_standby() detects that situation and descends into + # the standby RP console using: + # request platform software console attach switch standby + # After that the standby handle is a genuine standby session and all + # existing services behave normally. + # + # Off by default: it costs a 'show switch' on every console and an + # attach attempt on any stack where no console is on the standby, so it + # is opted into per testbed rather than changing every stack connect. + self.STACK_STANDBY_CONSOLE_ATTACH = False + # Route-processor plane used by the console attach command + self.STACK_STANDBY_CONSOLE_PLANE = 'R0' + # Timeout for the console attach + login sequence + self.STACK_STANDBY_CONSOLE_ATTACH_TIMEOUT = 120 + # How often to poke the attached console while waiting for its prompt + self.STACK_STANDBY_CONSOLE_POLL_INTERVAL = 20 + # The standby RP console is heavily rate limited, so large outputs + # stall part way through and time out the connect. These init commands + # are skipped on an attached standby console; the active handle has + # already supplied the same information. + self.STACK_STANDBY_CONSOLE_SKIP_INIT_COMMANDS = ['show version'] \ No newline at end of file diff --git a/src/unicon/plugins/iosxe/statemachine.py b/src/unicon/plugins/iosxe/statemachine.py index 3d119839..3f5bcb61 100644 --- a/src/unicon/plugins/iosxe/statemachine.py +++ b/src/unicon/plugins/iosxe/statemachine.py @@ -20,6 +20,18 @@ patterns = IosXEPatterns() statements = GenericStatements() + +def _config_to_enable_dialog(config_command): + statement_list = [statements.syslog_msg_stmt] + if config_command == 'config term': + statement_list.append(Statement( + pattern=patterns.confirm_uncommited_changes, + action='sendline(no)', + loop_continue=True, + )) + return Dialog(statement_list) + + def enable_bash_console_transition(statemachine, spawn, context): ''' Transition from enable mode to bash_console @@ -68,8 +80,7 @@ def enable_to_acm_transition(state_machine, spawn, context): spawn.sendline(f'acm configlet create {configlet_name}') def enable_to_syntax_transition(state_machine, spawn, context): - configlet_name = context.get('syntax_configlet', '') - spawn.sendline(f'syntax configlet create {configlet_name}') + spawn.sendline('config check syntax') def maintenance_to_enable_transition(statemachine, spawn, context): @@ -145,7 +156,9 @@ def create(self): enable_to_disable = Path(enable, disable, 'disable', None) enable_to_config = Path(enable, config, config_transition, Dialog([statements.syslog_msg_stmt])) - config_to_enable = Path(config, enable, 'end', Dialog([statements.syslog_msg_stmt])) + config_to_enable = Path( + config, enable, 'end', + _config_to_enable_dialog(self.config_command)) enable_to_guestshell = Path(enable, guestshell, 'guestshell run bash', None) guestshell_to_enable = Path(guestshell, enable, 'exit', None) @@ -156,7 +169,7 @@ def create(self): enable_to_acm = Path(enable, acm, enable_to_acm_transition, None) acm_to_enable = Path(acm, enable, 'end', None) - enable_to_syntax = Path(enable, syntax, 'config check syntax', None) + enable_to_syntax = Path(enable, syntax, enable_to_syntax_transition, None) syntax_to_enable = Path(syntax, enable, 'end', None) enable_to_rules = Path(enable, rules, 'acm rules', None) rules_to_enable = Path(rules, enable, 'end', None) @@ -260,7 +273,9 @@ def update_cur_state(sm, state): enable_to_config = Path(enable, config, config_transition, Dialog([statements.syslog_msg_stmt])) - config_to_enable = Path(config, enable, 'end', Dialog([statements.syslog_msg_stmt])) + config_to_enable = Path( + config, enable, 'end', + _config_to_enable_dialog(self.config_command)) enable_to_rommon = Path(enable, rommon, 'reload', Dialog( connection_statement_list + reload_statement_list)) diff --git a/src/unicon/plugins/iosxe/statements.py b/src/unicon/plugins/iosxe/statements.py index 099d9fab..41760fea 100644 --- a/src/unicon/plugins/iosxe/statements.py +++ b/src/unicon/plugins/iosxe/statements.py @@ -1,7 +1,9 @@ import re import time import logging +from copy import copy from functools import wraps +from inspect import Parameter, signature from datetime import datetime, timedelta from unicon.eal.dialogs import Statement @@ -324,12 +326,48 @@ def boot_finished_deco(func): Used with boot_from_rommon_statement_list (see below) ''' + parameters = signature(func).parameters + accepts_extra_keywords = any( + parameter.kind is Parameter.VAR_KEYWORD + for parameter in parameters.values() + ) + @wraps(func) def wrapper(spawn, session, context, **kwargs): - args = [a for a in [spawn, session, context] if a] - if context: + if context is not None: context.pop('boot_start_time', None) - return func(*args, **kwargs) + + available_arguments = { + 'spawn': spawn, + 'session': session, + 'context': context, + } + positional_arguments = [] + keyword_arguments = {} + for name, parameter in parameters.items(): + if name not in available_arguments: + continue + if parameter.kind is Parameter.POSITIONAL_ONLY: + positional_arguments.append(available_arguments[name]) + elif parameter.kind in ( + Parameter.POSITIONAL_OR_KEYWORD, + Parameter.KEYWORD_ONLY, + ): + keyword_arguments[name] = available_arguments[name] + + if accepts_extra_keywords: + keyword_arguments.update(kwargs) + else: + keyword_arguments.update( + (name, value) for name, value in kwargs.items() + if name in parameters + and parameters[name].kind in ( + Parameter.POSITIONAL_OR_KEYWORD, + Parameter.KEYWORD_ONLY, + ) + ) + + return func(*positional_arguments, **keyword_arguments) return wrapper @@ -337,7 +375,10 @@ def wrapper(spawn, session, context, **kwargs): # If the boot is completed because we hit a statement with # loop_continue = False, use the wrapper to pop the start time # from the context dict. -boot_from_rommon_statement_list += connection_statement_list.copy() +boot_from_rommon_statement_list += [ + copy(statement) for statement in connection_statement_list +] for stmt in boot_from_rommon_statement_list: - if stmt.pattern in [reload_patterns.press_return] or stmt.loop_continue is False: + if (stmt.pattern in [reload_patterns.press_return] or + stmt.loop_continue is False): stmt.action = boot_finished_deco(stmt.action) diff --git a/src/unicon/plugins/pid_tokens.csv b/src/unicon/plugins/pid_tokens.csv index a4b0a925..7f5e47c1 100644 --- a/src/unicon/plugins/pid_tokens.csv +++ b/src/unicon/plugins/pid_tokens.csv @@ -279,6 +279,7 @@ C6880-X-LE,iosxe,cat6k,c6800, C8000V,iosxe,c8kv,c8000v, C8200-1N-4T,iosxe,cat8k,c8200, C8200-UCPE-1N8,iosxe,cat8k,c8200, +C8300-2N2S-4T2X,iosxe,cat8k,c8300, C8500-12X,iosxe,cat8k,c8500, C8500-12X4QC,iosxe,cat8k,c8500, C8500L-8S4X,iosxe,cat8k,c8500, 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 9217c6a2..17124858 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 @@ -474,6 +474,8 @@ general_config: % The key modulus size is 2048 bits % Generating 2048 bit RSA keys, keys will be non-exportable... [OK] (elapsed time was 2 seconds) + "crypto key generate rsa general-keys label 9800-keys exportable": + new_state: general_config_rsa_modulus_prompt "crypto pki server ca": new_state: config_general_server "crypto gkm group g1": @@ -516,6 +518,26 @@ general_config: "end": new_state: general_enable +controller_transaction_enable: + prompt: "%N#" + commands: + <<: *gen_enable_cmds + "config term": + new_state: controller_transaction_config + +controller_transaction_config: + prompt: "%N(config)#" + commands: + "no logging console": "" + "end": + new_state: controller_transaction_confirm + +controller_transaction_confirm: + prompt: "Uncommitted changes found, commit them? [yes/no/CANCEL] " + commands: + "no": + new_state: controller_transaction_enable + general_config_nat_scale_confirm: prompt: "% Are you sure you want to continue? [yes]: " commands: @@ -603,6 +625,30 @@ config_general_redundancy_main_cpu: "end": new_state: general_enable +general_config_rsa_modulus_prompt: + prompt: "How many bits in the modulus [1024]: " + commands: + "2048": + new_state: general_config_rsa_keygen_done + "3072": + new_state: general_config_rsa_keygen_custom_done + +general_config_rsa_keygen_done: + prompt: "%N(conf)#" + result: | + % Generating 2048 bit RSA keys, keys will be exportable... + [OK] (elapsed time was 1 seconds) + commands: + <<: *general_config_cmds + +general_config_rsa_keygen_custom_done: + prompt: "%N(conf)#" + result: | + % Generating 3072 bit RSA keys, keys will be exportable... + [OK] (elapsed time was 1 seconds) + commands: + <<: *general_config_cmds + general_config_ca_profile: prompt: "%N(ca-profile-enroll)#" commands: diff --git a/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_asr1k_ha.yaml b/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_asr1k_ha.yaml index 8382f454..d8dd54ae 100644 --- a/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_asr1k_ha.yaml +++ b/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_asr1k_ha.yaml @@ -46,6 +46,8 @@ ha_asr1k_enable: "term length 0": "" "term width 0": "" "show version | include operating mode" : "" + "config check syntax": + new_state: ha_asr1k_syntax "show version": | Cisco IOS XE Software, Version BLD_V177_THROTTLE_LATEST_20210903_031009_V17_7_0_94 Cisco IOS Software [Bengaluru], ASR1000 Software (X86_64_LINUX_IOSD-UNIVERSALK9-M), Experimental Version 17.7.20210903:032925 [S2C-build-v177_throttle-475-/nobackup/mcpre/BLD-BLD_V177_THROTTLE_LATEST_20210903_031009 163] @@ -164,6 +166,15 @@ ha_asr1k_enable_reload_to_rommon_config: new_state: ha_asr1k_enable_reload_to_rommon +ha_asr1k_syntax: + prompt: "%N(syntax)#" + commands: + "interface loopback 1": "" + "description syntax test": "" + "end": + new_state: ha_asr1k_enable + + ha_asr1k_boot_to_rommon_stdby: preface: |2 Sep 20 19:20:19.891: %PMAN-5-EXITACTION: R0/0: pvp: Process manager is exiting: process exit with reload fru code diff --git a/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_cat9k_ha_reload.yaml b/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_cat9k_ha_reload.yaml index 5b74086c..c42cf2c9 100644 --- a/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_cat9k_ha_reload.yaml +++ b/src/unicon/plugins/tests/mock_data/iosxe/iosxe_mock_data_cat9k_ha_reload.yaml @@ -146,6 +146,9 @@ cat9k_ha_active_enable: "config term": new_state: cat9k_ha_active_config + "config check syntax": + new_state: cat9k_ha_active_syntax + "show boot": | --------------------------- Switch 1 @@ -161,6 +164,10 @@ cat9k_ha_active_enable: iPXE Timeout = 0 +cat9k_ha_active_syntax: + prompt: "%N(syntax)#" + + cat9k_ha_active_enable_reload_confirm: prompt: "System configuration has been modified. Save? [yes/no]: " commands: diff --git a/src/unicon/plugins/tests/test_plugin_generic.py b/src/unicon/plugins/tests/test_plugin_generic.py index c27f4bf6..93606ced 100644 --- a/src/unicon/plugins/tests/test_plugin_generic.py +++ b/src/unicon/plugins/tests/test_plugin_generic.py @@ -1505,17 +1505,20 @@ def test_reload_with_error_pattern(self): class TestGenericConnectionRefused(unittest.TestCase): - def test_connection_refused_handler_logs_max_count(self): + def test_connection_refused_handler_raises_typed_error(self): spawn = Mock() - spawn.settings.CONNECTION_REFUSED_MAX_COUNT = 3 - context = {'connection_refused_count': 2} + context = {} - with self.assertRaisesRegex(Exception, 'Connection refused to device'): + with self.assertRaisesRegex( + ConnectionRefusedError, 'Connection refused to device') as cm: connection_refused_handler(spawn, context) - spawn.log.error.assert_called_once_with( - 'Too many connection refused events: 3, update setting ' - 'CONNECTION_REFUSED_MAX_COUNT as needed.') + self.assertIs(cm.exception.spawn, spawn) + self.assertIs(cm.exception.context, context) + self.assertIs(cm.exception.device, spawn.device) + self.assertNotIn('connection_refused_count', context) + spawn.device.api.execute_clear_line.assert_not_called() + spawn.device.connect.assert_not_called() def test_connection_refused_handler_with_peripheral(self): md = MockDeviceTcpWrapper(device_os='iosxe', hostname='R1', diff --git a/src/unicon/plugins/tests/test_plugin_iosxe.py b/src/unicon/plugins/tests/test_plugin_iosxe.py index cafcf7e8..820876a1 100644 --- a/src/unicon/plugins/tests/test_plugin_iosxe.py +++ b/src/unicon/plugins/tests/test_plugin_iosxe.py @@ -16,13 +16,37 @@ from pyats.topology import loader import unicon -from unicon.plugins.generic.statements import terminal_position_handler +from unicon.plugins.generic.statements import ( + connection_statement_list, + generic_statements, + terminal_position_handler, +) from unicon import Connection 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 +from unicon.core.errors import ( + ConnectionError as UniconConnectionError, + StateMachineError, + SubCommandFailure, + UniconAuthenticationError, +) +from unicon.plugins.iosxe.patterns import IosXEPatterns +from unicon.plugins.iosxe.sdwan.statemachine import ( + SDWANDualRpStateMachine, + SDWANSingleRpStateMachine, +) +from unicon.plugins.iosxe.statemachine import ( + IosXEDualRpStateMachine, + IosXESingleRpStateMachine, +) +from unicon.plugins.iosxe.statements import ( + boot_finished_deco, + boot_from_rommon_statement_list, + 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 @@ -30,6 +54,125 @@ class TestIosXEStatements(unittest.TestCase): + class RommonDialogSpawn: + + def __init__(self): + self.timeout = 1 + self.hostname = None + self.match = ExpectMatch() + self.log = Mock() + + def read_update_buffer(self, size=None): + return 'Connection closed by foreign host' + + def match_buffer(self, patterns): + self.match.match_output = 'Connection closed by foreign host' + return 0 + + def trim_buffer(self): + pass + + def __str__(self): + return 'Router' + + def test_boot_finished_decorator_maps_arguments_by_name(self): + spawn = Mock() + session = {'booting': True} + context = {'boot_start_time': object()} + received = {} + + def callback(spawn, /, context, *, wait=None): + received.update(spawn=spawn, context=context, wait=wait) + + decorated_callback = boot_finished_deco(callback) + decorated_callback( + spawn, session, context, wait=5, unused_argument=True + ) + + self.assertEqual( + received, + {'spawn': spawn, 'context': context, 'wait': 5}, + ) + self.assertNotIn('boot_start_time', context) + + def test_boot_finished_decorator_forwards_statement_arguments(self): + spawn = Mock() + received = {} + + def callback(spawn, **kwargs): + received.update(spawn=spawn, kwargs=kwargs) + + boot_finished_deco(callback)( + spawn, {'booting': True}, {}, wait=5 + ) + + self.assertEqual(received, {'spawn': spawn, 'kwargs': {'wait': 5}}) + + def test_boot_finished_decorator_propagates_callback_type_error_once(self): + spawn = Mock() + callback_calls = [] + + def callback(spawn): + callback_calls.append(spawn) + raise TypeError('callback failure') + + with self.assertRaisesRegex(TypeError, 'callback failure'): + boot_finished_deco(callback)( + spawn, {}, {'boot_start_time': object()} + ) + + self.assertEqual(callback_calls, [spawn]) + + def test_boot_from_rommon_dialog_supports_spawn_only_callback(self): + disconnect_statement = next( + statement for statement in boot_from_rommon_statement_list + if statement.pattern == + generic_statements.disconnect_error_stmt.pattern + ) + context = {'boot_start_time': object()} + + with self.assertRaisesRegex( + Exception, 'received disconnect from router Router'): + Dialog([disconnect_statement]).process( + self.RommonDialogSpawn(), timeout=1, context=context + ) + + self.assertNotIn('boot_start_time', context) + + def test_boot_from_rommon_dialog_supports_spawn_context_callback(self): + connection_refused_statement = next( + statement for statement in boot_from_rommon_statement_list + if statement.pattern == + generic_statements.connection_refused_stmt.pattern + ) + spawn = self.RommonDialogSpawn() + spawn.device = None + context = {'boot_start_time': object()} + + with self.assertRaisesRegex( + ConnectionRefusedError, + 'Connection refused to device Router') as cm: + Dialog([connection_refused_statement]).process( + spawn, timeout=1, context=context + ) + + self.assertEqual(context, {}) + self.assertIs(cm.exception.spawn, spawn) + self.assertIs(cm.exception.context, context) + + def test_boot_from_rommon_owns_generic_connection_statements(self): + iosxe_connection_statements = boot_from_rommon_statement_list[ + -len(connection_statement_list): + ] + + self.assertEqual( + len(iosxe_connection_statements), + len(connection_statement_list), + ) + for iosxe_statement, generic_statement in zip( + iosxe_connection_statements, connection_statement_list): + self.assertIsNot(iosxe_statement, generic_statement) + def test_boot_image_finds_images_from_multiple_filesystems(self): spawn = Mock() spawn.settings = SimpleNamespace( @@ -252,6 +395,37 @@ def test_boot_image_passes_rommon_dir_timeout_to_expect(self): ) +class TestIosXEStateMachine(unittest.TestCase): + + def assert_config_to_enable_prompt_handler(self, state_machine_class): + state_machine = state_machine_class() + dialog = state_machine.get_path('config', 'enable').dialog + prompt_statements = [ + statement for statement in dialog.statements + if statement.pattern == IosXEPatterns().confirm_uncommited_changes + ] + + self.assertEqual(len(prompt_statements), 1) + statement = prompt_statements[0] + self.assertEqual(statement.action.__name__, 'sendline') + self.assertEqual(statement.args, {'key': 'no'}) + self.assertTrue(statement.loop_continue) + + def test_generic_config_to_enable_prompt_handler(self): + for state_machine_class in ( + IosXESingleRpStateMachine, IosXEDualRpStateMachine): + with self.subTest(state_machine=state_machine_class.__name__): + self.assert_config_to_enable_prompt_handler( + state_machine_class) + + def test_sdwan_config_to_enable_prompt_handler_is_not_duplicated(self): + for state_machine_class in ( + SDWANSingleRpStateMachine, SDWANDualRpStateMachine): + with self.subTest(state_machine=state_machine_class.__name__): + self.assert_config_to_enable_prompt_handler( + state_machine_class) + + class TestIosXEPluginConnect(unittest.TestCase): def test_asr_login_connect(self): @@ -399,6 +573,47 @@ def test_general_configure(self): c.configure(cmd, timeout=60, error_pattern=[], service_dialogue=None) self.assertEqual(c.spawn.match.match_output, 'end\r\nRouter#') + def test_configure_rsa_key_modulus_prompt_auto_answered(self): + # The default applies when the caller does not provide a reply. + c = Connection(hostname='Router', + start=['mock_device_cli --os iosxe ' + '--state general_login --hostname Router'], + os='iosxe', + credentials=dict( + default=dict(username='cisco', password='cisco'))) + try: + c.connect() + c.configure( + 'crypto key generate rsa general-keys ' + 'label 9800-keys exportable', timeout=60) + self.assertEqual(c.spawn.match.match_output, 'end\r\nRouter#') + finally: + c.disconnect() + + def test_configure_rsa_key_modulus_prompt_caller_override(self): + c = Connection(hostname='Router', + start=['mock_device_cli --os iosxe ' + '--state general_login --hostname Router'], + os='iosxe', + credentials=dict( + default=dict(username='cisco', password='cisco'))) + reply = Dialog([ + Statement(pattern=r'^.*How many bits in the modulus.*:\s*$', + action='sendline(3072)', + loop_continue=True, + continue_timer=False) + ]) + try: + c.connect() + output = c.configure( + 'crypto key generate rsa general-keys ' + 'label 9800-keys exportable', reply=reply, timeout=60) + self.assertIn( + 'How many bits in the modulus [1024]: 3072', output) + self.assertEqual(c.spawn.match.match_output, 'end\r\nRouter#') + finally: + c.disconnect() + def test_general_config_ca_profile(self): c = Connection(hostname='Router', start=['mock_device_cli --os iosxe --state general_login --hostname Router'], @@ -1101,6 +1316,32 @@ def test_config_transaction_sdwan_iosxe_confirm(self): d.configure('no logging console') d.disconnect() + +class TestIosXEControllerTransactionConfigure(unittest.TestCase): + + def test_config_to_enable_discards_uncommitted_changes(self): + connection = Connection( + hostname='Router', + start=[ + 'mock_device_cli --os iosxe ' + '--state controller_transaction_enable --hostname Router' + ], + os='iosxe', + credentials=dict( + default=dict(username='cisco', password='cisco')), + log_buffer=True, + mit=True, + ) + + try: + connection.connect() + connection.configure('no logging console') + self.assertEqual( + connection.state_machine.current_state, 'enable') + finally: + connection.disconnect() + + class TestIosXECat9kPluginReload(unittest.TestCase): @classmethod @@ -2121,7 +2362,16 @@ def test_syntax_configure(self): 'interface loopback 1', 'description test' ] - c.configure(config_txt, config_syntax_check='my_config') + with patch.object( + c.spawn, 'sendline', wraps=c.spawn.sendline) as sendline: + c.configure(config_txt, config_syntax_check=True) + self.assertEqual(c.state_machine.current_state, 'enable') + self.assertEqual(sendline.call_args_list, [ + unittest.mock.call('config check syntax'), + unittest.mock.call('interface loopback 1'), + unittest.mock.call('description test'), + unittest.mock.call('end'), + ]) c.disconnect() @@ -2151,6 +2401,130 @@ def test_multiple_banner_configure(self): c.configure(banner_config) c.disconnect() + +class TestIosxeConfigSyntaxService(unittest.TestCase): + + def test_single_rp_config_syntax_service(self): + c = Connection( + hostname='PE1', + start=[ + 'mock_device_cli --os iosxe --state general_enable ' + '--hostname PE1' + ], + os='iosxe', + mit=True + ) + c.connect() + try: + with patch.object( + c.spawn, 'sendline', wraps=c.spawn.sendline) as sendline: + c.config_syntax([ + 'interface loopback 1', + 'description test', + ]) + self.assertEqual(c.state_machine.current_state, 'enable') + self.assertEqual(sendline.call_args_list, [ + unittest.mock.call('config check syntax'), + unittest.mock.call('interface loopback 1'), + unittest.mock.call('description test'), + unittest.mock.call('end'), + ]) + finally: + c.disconnect() + + def test_dual_rp_config_syntax_service(self): + md = MockDeviceTcpWrapperIOSXE( + port=0, + state='ha_asr1k_exec,ha_asr1k_stby_exec', + hostname='R1') + md.start() + c = Connection( + hostname='R1', + start=[ + 'telnet 127.0.0.1 {}'.format(md.ports[0]), + 'telnet 127.0.0.1 {}'.format(md.ports[1]), + ], + os='iosxe', + connection_timeout=10, + credentials=dict(default=dict(password='lab')), + ) + try: + c.connect() + with patch.object( + c.active.spawn, + 'sendline', + wraps=c.active.spawn.sendline) as sendline: + c.config_syntax([ + 'interface loopback 1', + 'description syntax test', + ]) + self.assertEqual(c.active.state_machine.current_state, 'enable') + self.assertEqual(sendline.call_args_list, [ + unittest.mock.call('config check syntax'), + unittest.mock.call('interface loopback 1'), + unittest.mock.call('description syntax test'), + unittest.mock.call('end'), + ]) + finally: + c.disconnect() + md.stop() + + def test_syntax_configlet_rejected_single_rp(self): + c = Connection( + hostname='PE1', + start=[ + 'mock_device_cli --os iosxe --state general_enable ' + '--hostname PE1' + ], + os='iosxe', + mit=True + ) + c.connect() + try: + with patch.object( + c.spawn, 'sendline', wraps=c.spawn.sendline) as sendline: + with self.assertRaises(SubCommandFailure): + c.configure('interface loopback 1', + syntax_configlet='my_config') + sent = [call.args[0] for call in sendline.call_args_list + if call.args] + self.assertNotIn('config term', sent) + finally: + c.disconnect() + + def test_syntax_configlet_rejected_dual_rp(self): + md = MockDeviceTcpWrapperIOSXE( + port=0, + state='ha_asr1k_exec,ha_asr1k_stby_exec', + hostname='R1') + md.start() + c = Connection( + hostname='R1', + start=[ + 'telnet 127.0.0.1 {}'.format(md.ports[0]), + 'telnet 127.0.0.1 {}'.format(md.ports[1]), + ], + os='iosxe', + connection_timeout=10, + credentials=dict(default=dict(password='lab')), + ) + try: + c.connect() + with patch.object( + c.active.spawn, + 'sendline', + wraps=c.active.spawn.sendline) as sendline: + with self.assertRaises(SubCommandFailure): + c.configure('interface loopback 1', + syntax_configlet='my_config') + sent = [call.args[0] for call in sendline.call_args_list + if call.args] + self.assertNotIn('config term', sent) + finally: + c.disconnect() + md.stop() + + class TestIosxeBannerACM(unittest.TestCase): ''' Testcase to test banner using acm terminal input. diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_cat8k.py b/src/unicon/plugins/tests/test_plugin_iosxe_cat8k.py index cde1b4d1..1951cfe0 100644 --- a/src/unicon/plugins/tests/test_plugin_iosxe_cat8k.py +++ b/src/unicon/plugins/tests/test_plugin_iosxe_cat8k.py @@ -2,18 +2,41 @@ Unittests for iosxe/cat8k plugin """ +import re import unittest import unicon from unicon import Connection from unicon.plugins.tests.mock.mock_device_iosxe_cat8k import MockDeviceTcpWrapperIOSXECat8k from unicon.core.errors import SubCommandFailure +from unicon.plugins.iosxe.cat8k.service_statements import ( + boot_interrupt_stmt, + reload_to_rommon_statement_list, + save_env_to_rommon, +) unicon.settings.Settings.POST_DISCONNECT_WAIT_SEC = 0 unicon.settings.Settings.GRACEFUL_DISCONNECT_WAIT_SEC = 0.2 +class TestIosXeCat8kRommonStatements(unittest.TestCase): + + def test_system_bootstrap_interrupts_boot(self): + output = 'System Bootstrap, Version 17.11(1r), RELEASE SOFTWARE' + + self.assertIsNotNone(re.search(boot_interrupt_stmt.pattern, output)) + + def test_existing_autoboot_prompt_still_interrupts_boot(self): + output = 'Preparing to autoboot. [Press Ctrl-C to interrupt]' + + self.assertIsNotNone(re.search(boot_interrupt_stmt.pattern, output)) + + def test_rommon_transition_saves_config_register_change(self): + self.assertIs(reload_to_rommon_statement_list[0], save_env_to_rommon) + self.assertEqual(save_env_to_rommon.args, {'key': 'yes'}) + + class TestIosXeCat8kPlugin(unittest.TestCase): def test_connect(self): diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_cat9k.py b/src/unicon/plugins/tests/test_plugin_iosxe_cat9k.py index 248cd12e..fcc7faf5 100644 --- a/src/unicon/plugins/tests/test_plugin_iosxe_cat9k.py +++ b/src/unicon/plugins/tests/test_plugin_iosxe_cat9k.py @@ -11,6 +11,7 @@ from unicon.eal.dialogs import Statement, Dialog from unicon.plugins.tests.mock.mock_device_iosxe import MockDeviceTcpWrapperIOSXE from unicon.plugins.tests.mock.mock_device_iosxe_cat9k import MockDeviceTcpWrapperIOSXECat9k +from unicon.plugins.iosxe.cat9k.statemachine import IosXECat9kDualRpStateMachine from unicon.core.errors import SubCommandFailure from pyats.topology import loader @@ -19,6 +20,37 @@ unicon.settings.Settings.GRACEFUL_DISCONNECT_WAIT_SEC = 0.2 +class TestIosXeCat9kStateMachine(unittest.TestCase): + + def test_dual_rp_syntax_transition(self): + md = MockDeviceTcpWrapperIOSXECat9k( + port=0, + state='cat9k_ha_active_enable,cat9k_ha_standby_enable') + md.start() + + c = Connection( + hostname='switch', + start=[ + 'telnet 127.0.0.1 {}'.format(md.ports[0]), + 'telnet 127.0.0.1 {}'.format(md.ports[1]), + ], + os='iosxe', + platform='cat9k', + credentials=dict( + default=dict(username='cisco', password='cisco'), + alt=dict(username='admin', password='lab')), + ) + try: + c.connect() + sm = IosXECat9kDualRpStateMachine() + sm.update_cur_state('enable') + sm.go_to('syntax', c.spawn) + self.assertEqual(sm.current_state, 'syntax') + finally: + c.disconnect() + md.stop() + + class TestIosXeCat9kPlugin(unittest.TestCase): def test_connect(self): diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_stack.py b/src/unicon/plugins/tests/test_plugin_iosxe_stack.py index 20bc9740..a4549c34 100644 --- a/src/unicon/plugins/tests/test_plugin_iosxe_stack.py +++ b/src/unicon/plugins/tests/test_plugin_iosxe_stack.py @@ -3,7 +3,6 @@ """ -import re import unittest from unittest.mock import MagicMock, Mock, call, patch @@ -15,10 +14,20 @@ from unicon.eal.utils import ExpectMatch from unicon.core.errors import SubCommandFailure from unicon.plugins.iosxe.stack.service_implementation import StackReload +from unicon.plugins.iosxe.stack.service_statements import ( + stack_press_return, + stack_reload_auth_stmt_list, + stack_wait_and_enter, +) +from unicon.plugins.generic.statements import wait_and_enter as generic_wait_and_enter from unicon.plugins.tests.mock.mock_device_iosxe import MockDeviceTcpWrapperIOSXE from unicon.plugins.iosxe.stack.utils import StackUtils from unicon.plugins.generic.service_patterns import reload_patterns -from unicon.plugins.generic.service_statements import reload_statement_list +from unicon.plugins.generic.service_statements import ( + press_enter, + press_return, + reload_statement_list, +) unicon.settings.Settings.POST_DISCONNECT_WAIT_SEC = 0 @@ -239,16 +248,23 @@ def __init__(self, match_index): self.timeout = 1 self.hostname = None self.match_index = match_index + self.buffer = 'Press RETURN to get started!' self.match = ExpectMatch() self.log = Mock() self.sendline = Mock() + self.settings = MagicMock( + LOGIN_PROMPT=None, + PASSWORD_PROMPT=None, + ) self.match_buffer = Mock(side_effect=self._match_buffer) def read_update_buffer(self, size=None): - return 'Press RETURN to get started!\nUsername:' + return self.buffer def _match_buffer(self, pat_list): - self.match.match_output = 'Press RETURN to get started!' + pattern = pat_list[self.match_index] + self.match.last_match = pattern.search(self.buffer) + self.match.match_output = self.match.last_match.group() return self.match_index def trim_buffer(self): @@ -271,6 +287,84 @@ def test_reload_dialog_matched_retries_reset_for_stack_subconnections(self): self.assertEqual(press_return_stmt.matched_retries, 1) + def test_stack_reload_dialog_uses_stack_press_return(self): + connection = MagicMock( + settings=MagicMock(STACK_RELOAD_TIMEOUT=10)) + service = StackReload(connection, {}) + statements = service.dialog.statements + + self.assertEqual( + sum(statement.action is stack_wait_and_enter + for statement in statements), 1) + self.assertFalse(any( + statement.action is generic_wait_and_enter and + statement.pattern in (press_enter.pattern, press_return.pattern) + for statement in statements)) + + def test_stack_press_return_sends_return_when_prompt_is_unchanged(self): + spawn = self.StackReloadPromptSpawn(match_index=0) + + stack_wait_and_enter(spawn, wait=0.01) + + spawn.sendline.assert_called_once_with() + + def test_stack_reload_authenticates_after_press_return(self): + class StackReloadAuthSpawn: + def __init__(self): + self.timeout = 1 + self.hostname = 'Router' + self.buffer = 'Press RETURN to get started!' + self.match = ExpectMatch() + self.log = Mock() + self.settings = MagicMock( + LOGIN_PROMPT=None, + PASSWORD_PROMPT=None, + PASSWORD_ATTEMPTS=3, + ) + self.sent_lines = [] + self.last_sent = '' + self.read_count = 0 + + def read_update_buffer(self, size=None): + self.read_count += 1 + if self.read_count == 2: + self.buffer += '\nUsername:' + return self.buffer + + def match_buffer(self, patterns): + for index, pattern in enumerate(patterns): + match = pattern.search(self.buffer) + if match: + self.match.last_match = match + self.match.match_output = match.group() + return index + raise AssertionError( + 'No reload dialog pattern matched {!r}'.format( + self.buffer)) + + def trim_buffer(self): + self.buffer = self.buffer[self.match.last_match.end():] + + def sendline(self, value=''): + self.sent_lines.append(value) + self.last_sent = value + if value == 'admin': + self.buffer += '\nPassword:' + elif value == 'lab': + self.buffer += '\nRouter#' + + spawn = StackReloadAuthSpawn() + Dialog(stack_reload_auth_stmt_list).process( + spawn, + timeout=1, + context={'username': 'admin', 'tacacs_password': 'lab'}) + + self.assertTrue(stack_press_return.loop_continue) + self.assertEqual(spawn.sent_lines, ['admin', 'lab']) + self.assertEqual(spawn.buffer, '\nRouter#') + spawn.log.debug.assert_any_call( + 'Authentication prompt appeared while waiting; skipping RETURN') + @patch('unicon.plugins.iosxe.stack.service_implementation.sleep') @patch('unicon.plugins.iosxe.stack.service_implementation.utils') @patch('unicon.plugins.iosxe.stack.service_implementation.custom_auth_statements', diff --git a/src/unicon/plugins/tests/test_plugin_iosxe_stack_standby_console.py b/src/unicon/plugins/tests/test_plugin_iosxe_stack_standby_console.py new file mode 100644 index 00000000..758adef4 --- /dev/null +++ b/src/unicon/plugins/tests/test_plugin_iosxe_stack_standby_console.py @@ -0,0 +1,374 @@ +""" +Unittests for the IOSXE/Stack standby console attach. + +On a C9300 stack every member console presents the active RP's CLI, so the +handle unicon designates as standby is really a second session on the active. +These tests cover the detection and the attach that corrects it. + +The existing stack tests start several *identical* mock devices, so they cannot +express "console A is active and console B is standby". The logic is therefore +driven directly here. +""" + +import unittest +from unittest.mock import MagicMock, patch + +from unicon.plugins.iosxe.stack.connection_provider import ( + CONSOLE_BANNER_PATTERN, + CONSOLE_BUSY_PATTERN, + SHOW_SWITCH_ROW_RE, + StackRpConnectionProvider, + _console_banner_handler, + _console_busy_handler, +) + +import re + + +SHOW_SWITCH_ACTIVE = """show switch +Switch/Stack Mac Address : 74e2.e72d.8180 - Local Mac Address +Mac persistency wait time: Indefinite + H/W Current +Switch# Role Mac Address Priority Version State +------------------------------------------------------------------------------- +*1 Active 74e2.e72d.8180 1 V07 Ready + 2 Member c4ab.4d73.eb80 1 V07 Ready + 3 Standby 74e2.e72d.9080 1 V07 Ready +Switch#""" + +SHOW_SWITCH_STANDBY = SHOW_SWITCH_ACTIVE.replace( + '*1 Active', ' 1 Active').replace( + ' 3 Standby', '*3 Standby') + +ATTACH_BANNER = ( + 'request platform software console attach switch standby R0\r\n' + '#\r\n' + '# Connecting to the IOS console on the route-processor in slot 0.\r\n' + '# Enter Control-C to exit.\r\n' + '#\r\n' +) + +BUSY_OUTPUT = ('request platform software console attach switch standby R0\r\n' + '#ioucon:TTY open (remote): Address already in use\r\n') + + +def make_provider(connection): + """Build the provider without running the real __init__.""" + provider = StackRpConnectionProvider.__new__(StackRpConnectionProvider) + provider.connection = connection + provider.prompt_recovery = False + return provider + + +def make_subcon(output, alias='peer_1'): + subcon = MagicMock() + subcon.alias = alias + subcon.execute.return_value = output + subcon.context = {} + return subcon + + +class TestShowSwitchParsing(unittest.TestCase): + """The '*' row identifies the console; the standby need not be switch 2.""" + + def _local(self, output): + provider = make_provider(MagicMock()) + return provider._get_local_switch(make_subcon(output)) + + def test_active_console(self): + self.assertEqual(self._local(SHOW_SWITCH_ACTIVE), ('1', 'Active')) + + def test_standby_console_is_switch_3(self): + self.assertEqual(self._local(SHOW_SWITCH_STANDBY), ('3', 'Standby')) + + def test_ignores_non_local_rows(self): + """Switch 3 is Standby but is not this console.""" + self.assertEqual(self._local(SHOW_SWITCH_ACTIVE)[1], 'Active') + + def test_no_star_row(self): + self.assertEqual(self._local('show switch\nSwitch#'), (None, None)) + + def test_multi_digit_switch_number(self): + output = '*12 Standby 74e2.e72d.9080 1 V07 Ready' + self.assertEqual(self._local(output), ('12', 'Standby')) + + def test_execute_failure(self): + provider = make_provider(MagicMock()) + subcon = MagicMock() + subcon.execute.side_effect = Exception('boom') + self.assertEqual(provider._get_local_switch(subcon), (None, None)) + + +class TestConsolePatterns(unittest.TestCase): + + def test_busy_pattern_matches_device_message(self): + self.assertTrue(re.match(CONSOLE_BUSY_PATTERN, + '#ioucon:TTY open (remote): Address already in use')) + + def test_busy_pattern_ignores_normal_banner(self): + self.assertIsNone(re.match(CONSOLE_BUSY_PATTERN, + '# Enter Control-C to exit.')) + + def test_banner_pattern_matches(self): + self.assertTrue(re.match(CONSOLE_BANNER_PATTERN, + '# Enter Control-C to exit.')) + + def test_busy_handler_flags_context(self): + context = {} + _console_busy_handler(MagicMock(), context) + self.assertTrue(context['_standby_console_busy']) + + def test_banner_handler_pokes_console(self): + """The attached console prints nothing until it receives a newline.""" + spawn = MagicMock() + _console_banner_handler(spawn, {}) + spawn.sendline.assert_called_once_with() + + +class TestAttachPokesUntilPrompt(unittest.TestCase): + """A poke sent before the console is ready is swallowed, so it must retry.""" + + def _provider(self): + con = MagicMock() + con.settings.STACK_STANDBY_CONSOLE_PLANE = 'R0' + con.settings.STACK_STANDBY_CONSOLE_ATTACH_TIMEOUT = 10 + con.settings.STACK_STANDBY_CONSOLE_POLL_INTERVAL = 1 + # Real Statement objects are built from these, so they must be strings. + con.settings.LOGIN_PROMPT = r'^.*[Uu]sername: ?$' + con.settings.PASSWORD_PROMPT = r'^.*[Pp]assword: ?$' + provider = make_provider(con) + return provider + + def _subcon(self): + subcon = MagicMock() + subcon.context = {} + subcon.hostname = 'CTS-REG-UUT1' + # MagicMock auto-creates attributes, so start from a real False. + subcon._standby_console_attached = False + return subcon + + def test_retries_poke_when_console_stays_silent(self): + provider = self._provider() + subcon = self._subcon() + + # Silent twice, then the prompt arrives. + attempts = [TimeoutError('timeout'), TimeoutError('timeout'), None] + + def process(*args, **kwargs): + outcome = attempts.pop(0) + if outcome: + raise outcome + + with patch('unicon.plugins.iosxe.stack.connection_provider.Dialog') as dialog: + dialog.return_value.process.side_effect = process + provider._attach_standby_console(subcon) + + self.assertEqual(dialog.return_value.process.call_count, 3) + self.assertGreaterEqual(subcon.spawn.sendline.call_count, 2) + + def test_prompt_recovery_disabled_during_attach(self): + """Prompt recovery sends Ctrl-C, which would exit the attach.""" + provider = self._provider() + # Enabled on the provider, so passing it through would be visible here. + provider.prompt_recovery = True + subcon = self._subcon() + + with patch('unicon.plugins.iosxe.stack.connection_provider.Dialog') as dialog: + provider._attach_standby_console(subcon) + + _, kwargs = dialog.return_value.process.call_args + self.assertFalse(kwargs['prompt_recovery']) + + def test_busy_stops_retrying_immediately(self): + provider = self._provider() + subcon = self._subcon() + + def process(*args, **kwargs): + subcon.context['_standby_console_busy'] = True + raise TimeoutError('timeout') + + with patch('unicon.plugins.iosxe.stack.connection_provider.Dialog') as dialog: + dialog.return_value.process.side_effect = process + provider._attach_standby_console(subcon) + + self.assertEqual(dialog.return_value.process.call_count, 1) + self.assertFalse(getattr(subcon, '_standby_console_attached', False)) + + +class TestFindStandbyConsole(unittest.TestCase): + """A console attach is only needed when *no* console is on the standby.""" + + def _provider(self, outputs): + con = MagicMock() + con._subconnections = { + alias: make_subcon(out, alias) for alias, out in outputs.items() + } + provider = make_provider(con) + provider._release_stale_console_attach = MagicMock() + return provider + + def test_all_consoles_active_returns_none(self): + provider = self._provider({'peer_1': SHOW_SWITCH_ACTIVE, + 'peer_2': SHOW_SWITCH_ACTIVE, + 'peer_3': SHOW_SWITCH_ACTIVE}) + self.assertIsNone(provider._find_standby_console()) + + def test_returns_alias_of_standby_console(self): + provider = self._provider({'peer_1': SHOW_SWITCH_ACTIVE, + 'peer_2': SHOW_SWITCH_STANDBY, + 'peer_3': SHOW_SWITCH_ACTIVE}) + self.assertEqual(provider._find_standby_console(), 'peer_2') + + def test_finds_standby_that_is_not_the_designated_handle(self): + """The designation is positional, so the standby can be any console.""" + provider = self._provider({'peer_1': SHOW_SWITCH_ACTIVE, + 'peer_2': SHOW_SWITCH_ACTIVE, + 'peer_3': SHOW_SWITCH_STANDBY}) + self.assertEqual(provider._find_standby_console(), 'peer_3') + + def test_unreadable_console_does_not_claim_standby(self): + provider = self._provider({'peer_1': 'show switch\nSwitch#', + 'peer_2': 'show switch\nSwitch#'}) + self.assertIsNone(provider._find_standby_console()) + + def test_stale_attach_is_cleared_before_probing(self): + """Otherwise a leftover '-stby#' is mistaken for a real standby.""" + provider = self._provider({'peer_1': SHOW_SWITCH_ACTIVE, + 'peer_2': SHOW_SWITCH_ACTIVE}) + provider._find_standby_console() + self.assertEqual(provider._release_stale_console_attach.call_count, 2) + + +class TestInitStandby(unittest.TestCase): + + def _connection(self, attach_enabled=True): + con = MagicMock() + con.settings.STACK_STANDBY_CONSOLE_ATTACH = attach_enabled + con.standby.alias = 'peer_2' + return con + + def test_attaches_when_no_console_reaches_standby(self): + con = self._connection() + provider = make_provider(con) + provider._find_standby_console = MagicMock(return_value=None) + provider._attach_standby_console = MagicMock() + provider._get_local_switch = MagicMock(return_value=('3', 'Standby')) + + with patch('unicon.bases.routers.connection_provider.' + 'BaseStackRpConnectionProvider.init_standby'): + provider.init_standby() + + provider._attach_standby_console.assert_called_once_with(con.standby) + + def test_does_not_attach_when_a_console_is_on_standby(self): + con = self._connection() + provider = make_provider(con) + provider._find_standby_console = MagicMock(return_value='peer_2') + provider._attach_standby_console = MagicMock() + + with patch('unicon.bases.routers.connection_provider.' + 'BaseStackRpConnectionProvider.init_standby'): + provider.init_standby() + + provider._attach_standby_console.assert_not_called() + + def test_warns_when_standby_console_is_not_the_standby_handle(self): + con = self._connection() + provider = make_provider(con) + provider._find_standby_console = MagicMock(return_value='peer_3') + provider._attach_standby_console = MagicMock() + + with patch('unicon.bases.routers.connection_provider.' + 'BaseStackRpConnectionProvider.init_standby'): + provider.init_standby() + + provider._attach_standby_console.assert_not_called() + self.assertTrue(con.log.warning.called) + + def test_disabled_setting_is_a_no_op(self): + """Must not change behaviour on platforms with a real standby console.""" + con = self._connection(attach_enabled=False) + provider = make_provider(con) + provider._find_standby_console = MagicMock() + provider._attach_standby_console = MagicMock() + + with patch('unicon.bases.routers.connection_provider.' + 'BaseStackRpConnectionProvider.init_standby'): + provider.init_standby() + + provider._find_standby_console.assert_not_called() + provider._attach_standby_console.assert_not_called() + + +class TestSkipHeavyInitCommands(unittest.TestCase): + """'show version' stalls on the rate-limited standby RP console.""" + + def _provider(self, attached): + con = MagicMock() + con.settings.STACK_STANDBY_CONSOLE_SKIP_INIT_COMMANDS = ['show version'] + con.standby._standby_console_attached = attached + provider = make_provider(con) + return provider + + def _run(self, provider): + base = ('unicon.bases.routers.connection_provider.' + 'BaseStackRpConnectionProvider.set_init_commands') + + def fake(self_): + self_.init_exec_commands = ['term length 0', 'term width 0', + 'show version'] + + with patch(base, new=fake): + provider.set_init_commands() + return provider.init_exec_commands + + def test_show_version_dropped_when_console_attached(self): + provider = self._provider(attached=True) + self.assertEqual(self._run(provider), + ['term length 0', 'term width 0']) + + def test_full_command_set_kept_when_not_attached(self): + provider = self._provider(attached=False) + self.assertEqual(self._run(provider), + ['term length 0', 'term width 0', 'show version']) + + +class TestDetachOnDisconnect(unittest.TestCase): + + def test_detaches_only_attached_subconnections(self): + attached = MagicMock() + attached._standby_console_attached = True + plain = MagicMock() + plain._standby_console_attached = False + + con = MagicMock() + con.subconnections = [attached, plain] + provider = make_provider(con) + provider._detach_standby_console = MagicMock() + + with patch('unicon.bases.routers.connection_provider.' + 'BaseStackRpConnectionProvider.disconnect'): + provider.disconnect() + + provider._detach_standby_console.assert_called_once_with(attached) + + def test_detach_sends_ctrl_c(self): + """Ctrl-C exits the attached console; Ctrl-D does not.""" + con = MagicMock() + provider = make_provider(MagicMock()) + with patch('unicon.plugins.iosxe.stack.connection_provider.time.sleep'): + provider._detach_standby_console(con) + con.spawn.send.assert_any_call('\x03') + self.assertFalse(con._standby_console_attached) + + def test_release_stale_attach_survives_errors(self): + con = MagicMock() + con.spawn.send.side_effect = Exception('closed') + provider = make_provider(MagicMock()) + with patch('unicon.plugins.iosxe.stack.connection_provider.time.sleep'): + provider._release_stale_console_attach(con) + self.assertFalse(con._standby_console_attached) + + +if __name__ == '__main__': + unittest.main()