Release 25.9

docs/changelog/2025/september.rst

@@ -0,0 +1,39 @@
+September 2025
+==========
+
+September 30 - Unicon v25.9 
+------------------------
+
+
+
+.. csv-table:: Module Versions
+    :header: "Modules", "Versions"
+
+        ``unicon.plugins``, v25.9 
+        ``unicon``, v25.9 
+
+
+
+
+Changelogs
+^^^^^^^^^^
+--------------------------------------------------------------------------------
+                                      Fix                                       
+--------------------------------------------------------------------------------
+
+* backend/spawn
+    * Modified backend spawn implementation, updated logic to optimize buffer matching
+
+
+--------------------------------------------------------------------------------
+                                      Fix                                       
+--------------------------------------------------------------------------------
+
+* generic/service_statements
+    * Fixed the copy_overwrite_handler prompt
+
+* iosxe/stack
+    * Added pattern for install image to match
+    * Fixed the stack reload for the return_output true condition
+
+

docs/changelog/index.rst

@@ -4,6 +4,7 @@ Changelog
 .. toctree::
    :maxdepth: 2
 
+   2025/september
    2025/august
    2025/july
    2025/june

docs/changelog_plugins/2025/september.rst

@@ -0,0 +1,41 @@
+September 2025
+==========
+
+September 30 - Unicon v25.9
+------------------------
+
+
+
+.. csv-table:: Module Versions
+    :header: "Modules", "Versions"
+
+        ``unicon.plugins``, v25.9 
+        ``unicon``, v25.9 
+
+
+
+
+Changelogs
+^^^^^^^^^^
+--------------------------------------------------------------------------------
+                                      Fix                                       
+--------------------------------------------------------------------------------
+
+* generic
+    * Update config error pattern
+
+* iosxe
+    * Updated disable_prompt to ensure no false positives
+    * Added support to syntax prompt
+
+* pid tokens
+    * Updated PID tokens to support C9500X-60L4D devices
+
+
+--------------------------------------------------------------------------------
+                                      New                                       
+--------------------------------------------------------------------------------
+
+* iosxe/cat9k add rommon variable support in reload service
+
+

docs/changelog_plugins/index.rst

@@ -4,6 +4,7 @@ Plugins Changelog
 .. toctree::
    :maxdepth: 2
 
+   2025/september
    2025/august
    2025/july
    2025/june

src/unicon/plugins/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "25.8"
+__version__ = "25.9"
 
 supported_chassis = [
     'single_rp',

src/unicon/plugins/generic/service_statements.py

@@ -170,9 +170,9 @@ def copy_handler_1(spawn, context, send_key):
 
 def copy_overwrite_handler(spawn, context):
     if context['overwrite'] == 'False':
-        spawn.sendline('n')
+        spawn.send('n')
     else:
-        spawn.sendline('y')
+        spawn.send('y')
 
 def copy_error_handler(context, retry=False):
     if retry:

src/unicon/plugins/generic/settings.py

@@ -149,7 +149,8 @@ def __init__(self):
                                         r'subinterface is already configured as part of an IEEE 802.10, IEEE 802.1Q, '
                                         r'or ISL vLAN.',
                                         r'% OSPF: Please enable segment-routing globally',
-                                        r"% Invalid input detected at '^' marker"
+                                        r"% Invalid input detected at '^' marker",
+                                        r"%ERROR:"
                                         ]
 
         # Number of times to retry for config mode by configure service.

src/unicon/plugins/iosxe/__init__.py

@@ -30,6 +30,7 @@ def __init__(self):
         self.rommon = svc.Rommon
         self.tclsh = svc.Tclsh
         self.maintenance_mode = svc.MaintenanceMode
+        self.config_syntax = svc.ConfigSyntax
 
 
 class HAIosXEServiceList(HAServiceList):
@@ -48,6 +49,7 @@ def __init__(self):
         self.reset_standby_rp = svc.ResetStandbyRP
         self.rommon = svc.HARommon
         self.tclsh = svc.Tclsh
+        self.config_syntax = svc.ConfigSyntax
 
 
 class IosXESingleRpConnection(BaseSingleRpConnection):

src/unicon/plugins/iosxe/cat9k/c9500x/stackwise_virtual/service_implementation.py

@@ -1,4 +1,6 @@
 """ Stack based IOS-XE/cat9k/c9500X service implementations. """
+import io
+import logging
 from time import sleep
 from collections import namedtuple
 from datetime import timedelta
@@ -7,6 +9,7 @@
 from unicon.eal.dialogs import Dialog
 from unicon.core.errors import SubCommandFailure
 from unicon.bases.routers.services import BaseService
+from unicon.logs import UniconStreamHandler, UNICON_LOG_FORMAT
 
 from unicon.plugins.iosxe.stack.utils import StackUtils
 from unicon.plugins.generic.statements import custom_auth_statements
@@ -44,6 +47,7 @@ def __init__(self, connection, context, *args, **kwargs):
         self.end_state = 'enable'
         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)
 
     def call_service(self,
@@ -82,6 +86,20 @@ def call_service(self,
             if not isinstance(append_error_pattern, list):
                 raise ValueError('append_error_pattern should be a list')
             self.error_pattern += append_error_pattern
+
+        # Connecting to the log handler to capture the buffer output
+        lb = UniconStreamHandler(self.log_buffer)
+        lb.setFormatter(logging.Formatter(fmt=UNICON_LOG_FORMAT))
+        self.connection.log.addHandler(lb)
+
+        # logging the output to subconnections
+        for subcon in self.connection.subconnections:
+            subcon.log.addHandler(lb)
+
+        # Clear log buffer
+        self.log_buffer.seek(0)
+        self.log_buffer.truncate()
+
         # update all subconnection context with image_to_boot
         if image_to_boot:
             for subconn in self.connection.subconnections:
@@ -243,12 +261,23 @@ def boot(con):
         self.connection.connection_provider.init_connection()
 
         self.connection.log.info("+++ Reload Completed Successfully +++")
+
+        # Read the log buffer 
+        self.log_buffer.seek(0)
+        reload_output = self.log_buffer.read()
+        # clear buffer
+        self.log_buffer.truncate()
+
+        # Remove the handler
+        self.connection.log.removeHandler(lb)
+        for subcon in self.connection.subconnections:
+            subcon.log.removeHandler(lb)
+
         self.result = True
 
         if return_output:
             Result = namedtuple('Result', ['result', 'output'])
-            self.result = Result(self.result, reload_cmd_output.match_output.replace(reload_cmd, '', 1))
-
+            self.result = Result(self.result, reload_output.replace(reload_cmd, '', 1))
 
 class SVLStackSwitchover(BaseService):
     """ Get Rp state
@@ -355,3 +384,4 @@ def call_service(self, command=None,
         else:
             self.connection.log.info('Switchover failed')
             self.result = False
+

src/unicon/plugins/iosxe/cat9k/service_implementation.py

@@ -41,6 +41,8 @@ def call_service(self, *args, **kwargs):
         # assume the device is in rommon if image_to_boot is passed
         # update reload command to use rommon boot syntax
         if "image_to_boot" in kwargs:
+            if 'rommon_vars' in kwargs and self.connection.state_machine.current_state == 'rommon':
+                self.connection.execute([f'set {k}={v}' for k, v in kwargs['rommon_vars'].items()])
             self.context["image_to_boot"] = kwargs["image_to_boot"]
             reload_command = "boot {}".format(
                 self.context['image_to_boot']).strip()
@@ -63,6 +65,9 @@ def __init__(self, connection, context, **kwargs):
 
     def pre_service(self, *args, **kwargs):
         if "image_to_boot" in kwargs:
+            if 'rommon_vars' in kwargs and all(con.state_machine.current_state == 'rommon' for con in self.connection._subconnections):
+                for con in self.connection._subconnections:
+                    con.execute([f'set {k}={v}' for k, v in kwargs['rommon_vars'].items()])
             self.start_state = 'rommon'
             if 'image_to_boot' in self.context:
                 self.context['orig_image_to_boot'] = self.context['image_to_boot']

src/unicon/plugins/iosxe/connection_provider.py

@@ -32,7 +32,7 @@ def learn_tokens(self):
                                     context=con.context,
                                     prompt_recovery=con.prompt_recovery)
 
-            if con.state_machine.current_state in ['acm', 'config', 'rules', 'tclsh']:
+            if con.state_machine.current_state in ['acm', 'config', 'rules', 'syntax', 'tclsh']:
                 con.state_machine.go_to('enable',
                                         con.spawn,
                                         context=con.context,

src/unicon/plugins/iosxe/patterns.py

@@ -23,7 +23,7 @@ def __init__(self):
         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 = \
-            r'^(.*?)(\(unlicensed\))?(wlc|WLC|Router|RouterRP|Switch|ios|switch|%N)([0-9])?(\(recovery-mode\))?(\(rp-rec-mode\))?(\(standby\))?(-stby)?(-standby)?(\(boot\))?>\s?$'
+            r'^(.*?)(\(unlicensed\))?(wlc|WLC|Router|RouterRP|Switch|ios|switch|%N)([0-9])?(\(recovery-mode\))?(\(rp-rec-mode\))?(\(standby\))?(-stby)?(-standby)?(\(boot\))?(?<! -)>\s?$'
         self.enable_prompt = \
             r'^(.*?)(\(unlicensed\))?(wlc|WLC|Router|RouterRP|Switch|ios|switch|%N)([0-9])?(\(recovery-mode\))?(\(rp-rec-mode\))?(\(standby\))?(-stby)?(-standby)?(\(boot\))?#[\s\x07]*$'
         self.maintenance_mode_prompt = \
@@ -42,6 +42,7 @@ def __init__(self):
         self.macro_prompt = r'^(.*?)(\{\.\.\}|then.else.fi)\s*>\s*$'
         self.unable_to_create = r'^(.*?)Unable to create.*$'
         self.acm_prompt = r'^(.*?)\(acm.*?\)#[\s\x07]*$'
+        self.syntax_prompt = r'^(.*?)\(syntax.*?\)#[\s\x07]*$'
         self.rules_prompt = r'^(.*?)\(rules.*?\)#[\s\x07]*$'
 
 

src/unicon/plugins/iosxe/service_implementation.py

@@ -5,6 +5,8 @@
 import re
 from unicon.eal.dialogs import Dialog
 from unicon.core.errors import SubCommandFailure
+from unicon.bases.routers.services import BaseService
+
 
 from unicon.plugins.generic.service_implementation import (
     Configure as GenericConfigure,
@@ -58,6 +60,8 @@ def truncate_trailing_prompt(self, con_state,
     def pre_service(self, *args, **kwargs):
 
         self.acm_configlet = kwargs.pop('acm_configlet', None)
+        self.syntax_configlet = kwargs.pop('syntax_configlet', None)
+        self.config_syntax_check = kwargs.pop('config_syntax_check', False)
         self.rules = kwargs.pop('rules', False)
         self.prompt_recovery = kwargs.get('prompt_recovery', True)
 
@@ -72,6 +76,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})
+            self.start_state = 'syntax'
+            self.end_state = 'syntax'
+
         else:
             super().pre_service(*args, **kwargs)
 
@@ -88,6 +99,14 @@ class Config(Configure):
     pass
 
 
+class ConfigSyntax(Configure):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.start_state = 'syntax'
+        self.end_state = 'enable'
+
+
 class Execute(GenericExecute):
 
     def __init__(self, connection, context, **kwargs):
@@ -126,13 +145,19 @@ def __init__(self, connection, context, **kwargs):
 
     def pre_service(self, *args, **kwargs):
         self.acm_configlet = kwargs.pop('acm_configlet', None)
+        self.syntax_configlet = kwargs.pop('syntax_configlet', None)
         self.rules = kwargs.pop('rules', False)
         self.prompt_recovery = kwargs.get('prompt_recovery', True)
 
         if self.acm_configlet:
             self.connection.state_machine.go_to('acm', self.connection.spawn,context={'acm_configlet': self.acm_configlet})
             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})
+            self.start_state = 'syntax'
+            self.end_state = 'syntax'
         elif self.rules:
             if self.connection.connected:
                 self.connection.state_machine.go_to('rules', self.connection.spawn)
@@ -450,6 +475,18 @@ 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):