tessl/pypi-junos-eznc

Junos 'EZ' automation library for remotely managing and automating Juniper Networks Junos devices through NETCONF protocol

—

Pending

Overview

Eval results

Files

Software Management

Name: tessl/pypi-junos-eznc
Author: tessl

Software installation and management utilities for updating Junos software, managing software packages, performing system operations like reboots, and handling software validation and rollback operations.

Capabilities

Software Utility Class

The SW utility provides comprehensive software management capabilities including package installation, validation, system reboot operations, and software rollback functionality.

class SW:
    def __init__(self, dev):
        """
        Initialize SW utility bound to a device.
        
        Parameters:
        - dev (Device): Device object to bind to
        """

    def install(
        self,
        package=None,
        remote_path='/var/tmp',
        progress=None,
        validate=False,
        no_copy=False,
        timeout=1800,
        cleanfs=True,
        no_cleanfs_check=False
    ):
        """
        Install software package on the device.
        
        Parameters:
        - package (str): Local path to software package file
        - remote_path (str): Remote directory for package upload
        - progress (callable): Progress callback function
        - validate (bool): Validate package before installation
        - no_copy (bool): Skip copying package (package already on device)
        - timeout (int): Installation timeout in seconds
        - cleanfs (bool): Clean filesystem before installation
        - no_cleanfs_check (bool): Skip filesystem space check
        
        Returns:
        - bool: True if installation successful
        
        Raises:
        - SwRollbackError: Installation failed and rollback occurred
        - RpcTimeoutError: Installation timed out
        - RpcError: Installation RPC error
        """

    def reboot(self, in_min=0, at=None):
        """
        Reboot the device.
        
        Parameters:
        - in_min (int): Delay before reboot in minutes
        - at (str): Specific time for reboot (HH:MM format)
        
        Returns:
        - bool: True if reboot command accepted
        
        Raises:
        - RpcError: Reboot command failed
        """

    def poweroff(self, in_min=0, at=None):
        """
        Power off the device.
        
        Parameters:
        - in_min (int): Delay before poweroff in minutes  
        - at (str): Specific time for poweroff (HH:MM format)
        
        Returns:
        - bool: True if poweroff command accepted
        
        Raises:
        - RpcError: Poweroff command failed
        """

Software Package Operations

Operations for uploading, validating, and managing software packages on Junos devices.

def put(self, local_file, remote_path='/var/tmp', progress=None):
    """
    Upload file to device using SCP.
    
    Parameters:
    - local_file (str): Local file path to upload
    - remote_path (str): Remote directory path
    - progress (callable): Progress callback function
    
    Returns:
    - bool: True if upload successful
    
    Raises:
    - ScpError: SCP transfer failed
    """

def pkgadd(self, local_file, remote_path='/var/tmp', progress=None):
    """
    Add software package to device.
    
    Parameters:
    - local_file (str): Local package file path
    - remote_path (str): Remote directory for package
    - progress (callable): Progress callback function
    
    Returns:
    - bool: True if package add successful
    
    Raises:
    - RpcError: Package add operation failed
    """

def validate(self, remote_package, timeout=300):
    """
    Validate software package on device.
    
    Parameters:
    - remote_package (str): Remote package file path
    - timeout (int): Validation timeout in seconds
    
    Returns:
    - bool: True if package is valid
    
    Raises:
    - RpcError: Package validation failed
    - RpcTimeoutError: Validation timed out
    """

def rollback(self, timeout=1800):
    """
    Rollback software installation to previous version.
    
    Parameters:
    - timeout (int): Rollback timeout in seconds
    
    Returns:
    - bool: True if rollback successful
    
    Raises:
    - SwRollbackError: Rollback operation failed
    - RpcTimeoutError: Rollback timed out
    """

Software Binding

Bind the SW utility to a Device instance to enable software management operations.

# Binding SW utility to device
dev.bind(sw=SW)  # 'sw' is conventional name for SW utility

# Access bound utility
dev.sw.install('/path/to/package.tgz')
dev.sw.reboot()

Usage Examples

Basic Software Installation

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()

# Bind SW utility
dev.bind(sw=SW)

# Install software package
package_path = '/home/user/junos-install-ex-4300-21.4R3.15.tgz'

try:
    print("Starting software installation...")
    result = dev.sw.install(
        package=package_path,
        validate=True,  # Validate package before installation
        timeout=3600    # 1 hour timeout
    )
    
    if result:
        print("Software installation completed successfully")
        
        # Reboot device to activate new software
        print("Rebooting device to activate new software...")
        dev.sw.reboot()
    else:
        print("Software installation failed")
        
except Exception as e:
    print(f"Installation error: {e}")

dev.close()

Software Installation with Progress Tracking

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW

def progress_callback(dev, report):
    """Progress callback function for installation tracking."""
    print(f"Progress: {report}")

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(sw=SW)

# Install with progress tracking
package_path = '/home/user/junos-install-package.tgz'

dev.sw.install(
    package=package_path,
    progress=progress_callback,
    validate=True,
    cleanfs=True  # Clean filesystem before installation
)

dev.close()

Package Validation

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(sw=SW)

# Upload package first
local_package = '/home/user/software-package.tgz'
remote_package = '/var/tmp/software-package.tgz'

# Upload package
dev.sw.put(local_package, '/var/tmp/')

# Validate package before installation
try:
    is_valid = dev.sw.validate(remote_package)
    if is_valid:
        print("Package validation successful")
        
        # Install using no_copy since package is already on device
        dev.sw.install(
            package=remote_package,
            no_copy=True,
            timeout=2400  # 40 minutes
        )
    else:
        print("Package validation failed")
        
except Exception as e:
    print(f"Validation error: {e}")

dev.close()

Scheduled Reboot

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(sw=SW)

# Schedule reboot in 10 minutes
dev.sw.reboot(in_min=10)
print("Device scheduled to reboot in 10 minutes")

# Or schedule reboot at specific time
dev.sw.reboot(at='02:00')  # Reboot at 2:00 AM
print("Device scheduled to reboot at 2:00 AM")

dev.close()

Software Rollback

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW
from jnpr.junos.exception import SwRollbackError

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(sw=SW)

# Check current software version
print(f"Current version: {dev.facts['version']}")

try:
    # Perform software rollback
    print("Starting software rollback...")
    result = dev.sw.rollback(timeout=3600)
    
    if result:
        print("Software rollback completed successfully")
        
        # Reboot to activate previous software version
        dev.sw.reboot()
    else:
        print("Software rollback failed")
        
except SwRollbackError as e:
    print(f"Rollback error: {e}")

dev.close()

Error Handling

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW
from jnpr.junos.exception import SwRollbackError, RpcTimeoutError, RpcError

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(sw=SW)

package_path = '/home/user/software-package.tgz'

try:
    # Attempt software installation
    dev.sw.install(
        package=package_path,
        validate=True,
        timeout=3600
    )
    print("Installation successful")
    
except SwRollbackError as e:
    print(f"Installation failed and rollback occurred: {e}")
    
except RpcTimeoutError:
    print("Installation timed out")
    
except RpcError as e:
    print(f"Installation RPC error: {e}")
    
except FileNotFoundError:
    print(f"Package file not found: {package_path}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

dev.close()

Multiple Device Software Update

from jnpr.junos import Device
from jnpr.junos.utils.sw import SW
import concurrent.futures

def update_device(hostname, package_path):
    """Update software on a single device."""
    try:
        dev = Device(host=hostname, user='admin', passwd='secret')
        dev.open()
        dev.bind(sw=SW)
        
        print(f"Updating {hostname}...")
        
        # Install software
        result = dev.sw.install(
            package=package_path,
            validate=True,
            timeout=3600
        )
        
        if result:
            print(f"{hostname}: Installation successful")
            dev.sw.reboot(in_min=2)  # Reboot in 2 minutes
            return f"{hostname}: Success"
        else:
            return f"{hostname}: Installation failed"
            
    except Exception as e:
        return f"{hostname}: Error - {e}"
    finally:
        if 'dev' in locals():
            dev.close()

# List of devices to update
devices = ['router1.example.com', 'router2.example.com', 'router3.example.com']
package_path = '/home/user/junos-software.tgz'

# Update devices in parallel
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(update_device, hostname, package_path) 
               for hostname in devices]
    
    for future in concurrent.futures.as_completed(futures):
        result = future.result()
        print(result)

Types

# Software package types
PackagePath = str  # Local or remote path to software package
RemotePath = str   # Remote directory path on device

# Progress callback type
ProgressCallback = callable  # Function(device, report) for progress updates

# Installation options
InstallOptions = dict[str, any]  # Dictionary of installation parameters

# Time format for scheduled operations
TimeFormat = str  # HH:MM format for scheduled operations

# Installation result
InstallResult = bool  # True if installation successful

# Software validation result
ValidationResult = bool  # True if package is valid

Install with Tessl CLI