Release 25.11

Makefile

@@ -48,7 +48,7 @@ RELATED_PKGS += genie.libs.filetransferutils
 # Adding pyasyncore pkg to fix pysnmp scripts for python 3.12
 DEPENDENCIES = restview psutil Sphinx wheel asynctest 'pysnmp>=6.1.4,<6.2' pyasn1==0.6.0
 DEPENDENCIES += sphinx-rtd-theme==1.1.0 pyftpdlib tftpy\<0.8.1 robotframework
-DEPENDENCIES += Cython requests ruamel.yaml grpcio protobuf jinja2 pyVmomi
+DEPENDENCIES += Cython requests "ruamel.yaml.clib<0.2.15" ruamel.yaml grpcio protobuf jinja2 pyVmomi asyncssh
 # Internal variables.
 # (note - build examples & templates last because it will fail uploading to pypi
 #  due to duplicates, and we'll for now accept that error)

pkgs/clean-pkg/changelog/2025/december.rst

@@ -0,0 +1,31 @@
+--------------------------------------------------------------------------------
+                                      Fix                                       
+--------------------------------------------------------------------------------
+
+* clean-pkg
+    * Added support to block copy operations if the target file size does not match.
+
+* rommonboot stage
+    * iosxe
+        * Removed duplicate task function.
+    * cat9k
+        * Removed device.destroy() call and added device.sendline() in the rommon boot stage so that the device reaches the rommon prompt.
+
+* clean-pkg/stages
+    * Added the reset of the rollup flag when recovery is enabled
+    * Updated the api configure_management to able to skip for missing attribute instead of failing complete stages.
+
+* iosxe
+    * clean-pkg/utils
+        * Fixed issue with updating protected file for image
+
+
+--------------------------------------------------------------------------------
+                                      New                                       
+--------------------------------------------------------------------------------
+
+* clean/recover
+    * Added power cycle retry mechanism to enhance reliability during device recovery.
+    * Updated the console speed configuration in case of failiure connection to device
+
+

pkgs/clean-pkg/sdk_generator/github/clean_datafile.yaml

@@ -3,6 +3,6 @@ root_directories:
      root: 'genie.libs.clean'
      mod_name: 'clean'
      url:
-         link: 'https://github.com/CiscoTestAutomation/genielibs/tree/{branch}/'
-         branch: 'master'
+         link: 'https://github.com/CiscoTestAutomation/genielibs/tree/{branch}/pkgs/clean-pkg/'
+         branch: 'main'
          style: 'github'

pkgs/clean-pkg/src/genie/libs/clean/__init__.py

@@ -8,7 +8,7 @@
 '''
 
 # metadata
-__version__ = "25.10"
+__version__ = "25.11"
 __author__ = 'Cisco Systems Inc.'
 __contact__ = ['[email protected]', '[email protected]']
 __copyright__ = 'Copyright (c) 2019, Cisco Systems Inc.'

pkgs/clean-pkg/src/genie/libs/clean/recovery/recovery.py

@@ -59,7 +59,8 @@ def _disconnect_reconnect(device):
 
 
 def _recovery_steps(device, clear_line=True, powercycler=True,
-                    powercycler_delay=30, reconnect_delay=60, **kwargs):
+                    powercycler_delay=30, reconnect_delay=60, configure_console_speed=False,
+                    **kwargs):
 
     '''Steps to recover device
     1. First step clear line
@@ -99,29 +100,40 @@ def _recovery_steps(device, clear_line=True, powercycler=True,
     else:
         log.info('Clear line is not provided!')
 
-    # Step-3: Powercycle device
+    # Step-3&4: Powercycle device
     if powercycler:
-        log.info(banner("Powercycling device '{}'".format(device.name)))
-
-        try:
+        power_cycle_retry = 1
+        if configure_console_speed:
+            power_cycle_retry = 2
+        for powercycler_attempt in range(power_cycle_retry):
+            log.info(banner(f" Powercycling device '{device.name}'"))
             device.api.execute_power_cycle_device(delay=powercycler_delay)
+            log.info(f"Successfully powercycled device '{device.name}' during recovery")
+            try:
+                device.api.device_recovery_boot()
+            except Exception as e:
+                log.error(str(e))
+                if configure_console_speed and powercycler_attempt + 1 < power_cycle_retry:
+                    log.info(f"Device '{device.name}' failed to boot. Updating the console speed.")
+                    log.info(banner(f"Attempting to configure management console speed on device '{device.name}'"))
+                    device.destroy()
+                    device.api.configure_management_console()
+                    log.info(f"Successfully configured management console speed on device '{device.name}'")
+                    device.disconnect()
+                    continue
+                else:
+                    raise Exception(f"Failed to boot device '{device.name}' after powercycle")
+            else:
+                log.info(f"Successfully booted device '{device.name}' after powercycle")
+                break
+    else:
+        # Powercycler not provided do only step-4:
+        log.info('Powercycler is not provided!')
+        try:
+            device.api.device_recovery_boot()
         except Exception as e:
             log.error(str(e))
-            raise Exception("Failed to powercycle device '{}'".format(device.name))
-        else:
-            log.info("Successfully powercycled device '{}' during recovery".\
-                     format(device.name))
-    else:
-        log.info("powercycle is not provided!")
-
-    # Step-4: Boot device with given golden image or by tftp boot
-    try:
-        device.api.device_recovery_boot()
-    except Exception as e:
-        log.error(e)
-        raise Exception(f"Failed to boot the device {device.name}")
-    else:
-        log.info(f"successfully booted the device {device.name}")
+            raise Exception(f"Failed to boot device '{device.name}' after powercycle")
 
     log.info('Sleeping for {r} before reconnection'.format(r=reconnect_delay))
     time.sleep(reconnect_delay)
@@ -158,6 +170,7 @@ def _recovery_steps(device, clear_line=True, powercycler=True,
     },
     Optional('recovery_password'): str,
     Optional('clear_line'): bool,
+    Optional('configure_console_speed'): bool,
     Optional('powercycler'): bool,
     Optional('powercycler_delay'): int,
     Optional('reconnect_delay'): int,
@@ -181,7 +194,8 @@ def recovery_processor(
         powercycler_delay=30,
         reconnect_delay=60,
         post_recovery_configuration=None,
-        connection_timeout=45
+        connection_timeout=45,
+        configure_console_speed=True,
         ):
 
     '''
@@ -195,6 +209,7 @@ def recovery_processor(
           console_breakboot_char: <Character to send when console_activity_pattern is matched, 'str'>
           console_breakboot_telnet_break: Use telnet `send break` to interrupt device boot
           grub_activity_pattern: <Break pattern on the device for grub boot mode, 'str'>
+          configure_console_speed: <Should configure console speed during recovery, 'bool'> (Default: False)
           grub_breakboot_char: <Character to send when grub_activity_pattern is matched, 'str'>
           timeout: <Timeout in seconds to recover the device, 'int'>
           recovery_password: <Device password after coming up, 'str'>
@@ -364,7 +379,7 @@ def recovery_processor(
 
         try:
             _recovery_steps(device, clear_line, powercycler,
-                          powercycler_delay, reconnect_delay)
+                          powercycler_delay, reconnect_delay, configure_console_speed)
         except Exception as e:
             # Could not recover the device!
             log.error(banner("*** Terminating Genie Clean ***"))

pkgs/clean-pkg/src/genie/libs/clean/stages/iosxe/cat9k/stages.py

@@ -385,7 +385,7 @@ def reload_check(device, target):
                 device.expect(['(.*Initializing Hardware.*|^(.*)((rommon(.*))+>|switch *:).*$)'], timeout=60)
 
             log.info("Device is reloading")
-            device.destroy_all()
+            device.sendline()
 
     def rommon_boot(self, steps, device, image, tftp=None, timeout=TIMEOUT, recovery_password=RECOVERY_PASSWORD,
                   recovery_username=RECOVERY_USERNAME, recovery_enable_password=RECOVERY_ENABLE_PASSWORD, ether_port=ETHER_PORT):

pkgs/clean-pkg/src/genie/libs/clean/stages/iosxe/cat9k/tests/test_stages.py

@@ -118,7 +118,6 @@ def test_go_to_rommon_pass(self, dialog, statement):
         self.device.is_ha = False
         self.device.subconnections = mock.MagicMock()
         self.device.expect = mock.Mock()
-        self.device.destroy_all = mock.Mock()
 
         reload_dialog = mock.Mock()
         dialog.return_value = reload_dialog
@@ -140,13 +139,15 @@ def test_go_to_rommon_pass(self, dialog, statement):
                       continue_timer=False)
         ])
 
-        self.device.sendline.assert_called_with("reload")
+        # Check both sendline calls
+        self.device.sendline.assert_has_calls([
+            mock.call("reload"),
+            mock.call()
+        ])
         reload_dialog.process.assert_called_with(self.device.spawn)
         self.device.expect.assert_called_with(
             ['(.*Initializing Hardware.*|^(.*)((rommon(.*))+>|switch *:).*$)'], timeout=60)
 
-        self.device.destroy_all.assert_called_once()
-
         #  step_context comes from the following snippet
         #  with steps.start('...') as step_context:
         step_context = steps.start.return_value.__enter__.return_value

pkgs/clean-pkg/src/genie/libs/clean/stages/iosxe/stages.py

@@ -2591,61 +2591,6 @@ def _grub_boot_device(spawn, session, context):
             con.context['grub_boot_image'] = image_to_boot
         con.context['image_to_boot'] = image_to_boot
 
-    def task(con, timeout, image, grub_activity_pattern):
-        """
-        A method for processing a dialog that loads a local image onto a device
-
-        Args:
-            con (obj): connection object
-            timeout (int, optional): Recovery process timeout. Defaults to 600.
-            image (dict): Information to load golden image on the device
-            grub_activity_pattern (str): Grub activity pattern
-        Returns:
-            None
-        """
-        # if we have grup activity pattern then we need to update the context with grub boot image
-
-        if grub_activity_pattern:
-            image_to_boot = con.context.get('grub_boot_image')
-            con.context['grub_boot_image'] = image[0]
-
-        # check for image to boot if we have a value store it and replace it with the golden image
-        image_to_boot = con.context.get('image_to_boot')
-        con.context['image_to_boot'] = image[0]
-
-        # these statments needed for booting from grub menu
-        def _grub_boot_device(spawn, session, context):
-            # '\033' == <ESC>
-            spawn.send('\033')
-            time.sleep(0.8)
-
-        grub_prompt_stmt = \
-            Statement(pattern=r'.*grub *>.*',
-                    action=_grub_boot_device,
-                    args=None,
-                    loop_continue=True,
-                    continue_timer=False)
-
-        grub_boot_stmt = \
-            Statement(pattern=r'.*Use the UP and DOWN arrow keys to select.*',
-                    action=grub_prompt_handler,
-                    args=None,
-                    loop_continue=True,
-                    continue_timer=False)
-
-        dialog = Dialog([grub_boot_stmt, grub_prompt_stmt])
-
-        con.state_machine.go_to('enable',
-                                con.spawn,
-                                timeout=timeout,
-                                context=con.context,
-                                dialog=dialog,
-                                prompt_recovery=True)
-        # we need to set the value of grub_boot_image and image_to_boot to original value
-        if grub_activity_pattern:
-            con.context['grub_boot_image'] = image_to_boot
-        con.context['image_to_boot'] = image_to_boot
-
     def reconnect(self, steps, device, reconnect_timeout=RECONNECT_TIMEOUT):
         with steps.start("Reconnect to device") as step:
 

pkgs/clean-pkg/src/genie/libs/clean/stages/iosxe/tests/test_install_image.py

@@ -540,11 +540,11 @@ def test_iosxe_install_image_pass_retries_not_enough_space(self, dialog):
         device.api.get_running_image = Mock()
         device.api.collect_install_log = Mock()
         device.api.free_up_disk_space = Mock(return_value=True)
-        cls.install_image(steps=steps, device=device, images=['sftp://server/image.bin'])
+        cls.install_image(steps=steps, device=device, images=['bootflash:/image.bin'])
 
-        expected_execute_call = [call('install add file sftp://server/image.bin activate commit prompt-level none', reply=reload_dialog, error_pattern=['FAILED:'], timeout=500),
+        expected_execute_call = [call('install add file bootflash:/image.bin activate commit prompt-level none', reply=reload_dialog, error_pattern=['FAILED:'], timeout=500),
                                 call('more bootflash:packages.conf'),
-                                call('install add file sftp://server/image.bin activate commit prompt-level none', reply=reload_dialog, error_pattern=['FAILED:'], timeout=500),
+                                call('install add file bootflash:/image.bin activate commit prompt-level none', reply=reload_dialog, error_pattern=['FAILED:'], timeout=500),
                                 call('install commit')]
 
         device.execute.assert_has_calls(expected_execute_call)
@@ -559,7 +559,7 @@ def test_iosxe_install_image_pass_retries_not_enough_space(self, dialog):
             )
         device.reload.assert_has_calls([expected_reload_call])
         device.api.free_up_disk_space.assert_called_with(destination='', required_size=5000,
-                                                         protected_files=['//server/image.bin'], allow_deletion_failure=True, skip_deletion=False)
+                                                         protected_files=['image.bin'], allow_deletion_failure=True, skip_deletion=False)
         device.api.get_running_image.assert_called_once()
         self.assertEqual(Passed, steps.details[0].result)
 

pkgs/clean-pkg/src/genie/libs/clean/stages/stages.py

@@ -102,6 +102,14 @@ def connect(self,
 
             log.info('Checking connection to device: %s' % device.name)
 
+            # If recovery is enabled, ignore rollup
+            section = self.parameters.get('section')
+
+            # Check if 'section' exists and has a parent with 'device_recovery_processor'
+            if section and getattr(section.parent, 'device_recovery_processor',
+                                   None):
+                step.result_rollup = False
+
             # Create a timeout that will loop
             retry_timeout = Timeout(float(retry_timeout),
                                     float(retry_interval))
@@ -1170,8 +1178,36 @@ def copy_to_device(self,
                                 "Unable to check if image '{}' exists on device {} {}."
                                 "Error: {}".format(dest_file_path, device.name,
                                                    dest, str(e)))
-
-                        if (not exist) or (exist and overwrite) or (
+                        name_exists = False
+                        try:
+                            if os.path.basename(dest_file_path) in dir_before:
+                                name_exists = True
+                        except Exception:
+                            name_exists = False
+                        if name_exists and not exist and not (unique_file_name or unique_number or rename_images):
+                            if not overwrite:
+                                step.failed(
+                                    "Destination file '{}' already exists on device {} {} with a different size. "
+                                    "overwrite is set to False will not overwrite. Remove the file manually or set overwrite=True to replace.".format(
+                                        dest_file_path, device.name, dest
+                                    )
+                                )
+                            else:
+                                file_copy_info = {
+                                    file: {
+                                        'size': file_size,
+                                        'dest_path': dest_file_path,
+                                        'exist': True
+                                    }
+                                }
+                                files_to_copy.update(file_copy_info)
+                                step.passed(
+                                    "Destination file '{}' exists with different size on device {} {}. "
+                                    "overwrite=True, will attempt copy and replace.".format(
+                                        dest_file_path, device.name, dest
+                                    )
+                                )
+                        elif (not exist) or (exist and overwrite) or (
                                 exist and (unique_file_name or unique_number
                                            or rename_images)):
                             # Update list of files to copy
@@ -3266,26 +3302,27 @@ def ping_gateway(self,
                      ping_sleep=PING_SLEEP,
                      **kwargs):
         """Ping the gateway to ensure connectivity."""
-        management = getattr(device, "management", {})
-        config_kwargs = {
-            k: v
-            for k, v in kwargs.items()
-            if k in [k.schema for k in self.schema.keys()]
-        }
-        ip4_gateway = config_kwargs.get("gateway",
-                                        {}).get("ipv4") or management.get(
-                                            "gateway", {}).get("ipv4")
-        ip6_gateway = config_kwargs.get("gateway",
-                                        {}).get("ipv6") or management.get(
-                                            "gateway", {}).get("ipv6")
-        interface = config_kwargs.get("interface") or management.get(
-            "interface")
-
-        vrf = config_kwargs.get("vrf") or management.get("vrf")
-
-        if not ip4_gateway and not ip6_gateway:
-            steps.failed("No gateway configured for management interface")
-            return
+        with steps.start("Verify Gateway Configuration") as step:
+            management = getattr(device, "management", {})
+            config_kwargs = {
+                k: v
+                for k, v in kwargs.items()
+                if k in [k.schema for k in self.schema.keys()]
+            }
+            ip4_gateway = config_kwargs.get("gateway",
+                                            {}).get("ipv4") or management.get(
+                                                "gateway", {}).get("ipv4")
+            ip6_gateway = config_kwargs.get("gateway",
+                                            {}).get("ipv6") or management.get(
+                                                "gateway", {}).get("ipv6")
+            interface = config_kwargs.get("interface") or management.get(
+                "interface")
+
+            vrf = config_kwargs.get("vrf") or management.get("vrf")
+
+            if not ip4_gateway and not ip6_gateway:
+                step.failed("No gateway configured for management interface")
+                return
 
         for gateway in [ip4_gateway, ip6_gateway]:
             if gateway: