Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
protocol ('str'): Transfer protocol
image ('str'): Image name
timeout_seconds ('int'): Maximum duration to wait for file copy
wait_time_after_copy ('int'): Wait time after file copy
overwrite ('bool'): Flag to overwrite existing file
Raises:
Exception: Failed copying ISSU image to disk
Returns:
None
"""
from_url = "{protocol}://{address}/{path}/{image}".format(
protocol=protocol, address=address, path=path, image=image
)
filetransfer = FileUtils.from_device(device)
filetransfer.copyfile(
source=from_url, destination=disk, device=device,
timeout_seconds=timeout_seconds, overwrite=overwrite
)
time.sleep(wait_time_after_copy)
output = device.execute(
"dir {disk}{image}".format(disk=disk, image=basename(image))
)
if "Error" not in output:
log.info("Copied ISSU image to '{disk}'".format(disk=disk))
else:
raise Exception(
"Unable to copy ISSU image to '{disk}'".format(disk=disk)
format(f=self.to_url, d=device.name))
log.error(str(diff.diffs))
raise Exception(
"Comparison between running-config and "
"config-replace file '{f}' failed for device"
" {d}".format(
f=self.to_url, d=device.name))
else:
log.info(
"Comparison between running-config and config-replace"
"file '{f}' passed for device {d}". format(
f=self.to_url, d=device.name))
if delete_after_restore:
# Delete location:
self.filetransfer = FileUtils.from_device(device)
self.filename = self.to_url
self.filetransfer.deletefile(target=self.to_url, device=device)
# Verify location: deleted
dir_output = self.filetransfer.dir(
target=self.to_url, device=device)
for file in dir_output:
if self.filename in file:
break
else:
log.info("Successfully deleted '{}'".format(self.to_url))
return
raise Exception("Unable to delete '{}'".format(self.to_url))
else:
# modify the device via callable function
# using Conf object
def save_configuration_to_file(self, device, default_dir, file_name):
''' Save current configuration to file on device
'''
file_path = '{}{}'.format(default_dir, file_name)
try:
# Instantiate a filetransferutils instance for IOSXE device
self.filetransfer = FileUtils.from_device(device)
self.filetransfer.copyconfiguration(source='running-config',
destination=file_path,
device=device)
except Exception as e:
log.error(e)
raise Exception(
"Issue saving config to {c}".format(
c=file_path)) from e
elif method == 'local':
self.run_config = device.execute('show running-config')
elif method == 'config_replace':
# Create unique filename
self.filename = self.__class__.__name__ + \
time.ctime().replace(' ', '_').replace(':', '_')
# Set from/to locations
self.from_url = 'running-config'
self.to_url = '{dir}{filename}'.format(
filename=self.filename, dir=default_dir[device.name])
# Instantiate a filetransferutils instance for IOSXE device
self.filetransfer = FileUtils.from_device(device)
# Execute copy running-config to location:
self.filetransfer.copyconfiguration(source=self.from_url,
destination=self.to_url,
device=device)
if copy_to_standby:
self.stby_url = '{dir}{filename}'.format(
dir='stby-{}' .format(default_dir[device.name]), filename=self.filename)
# copy config to stby-bootflash:
self.filetransfer.copyconfiguration(source=self.from_url,
destination=self.stby_url,
device=device)
# Verify location: exists
# Get args
server = origin['hostname']
image_files = remove_string_from_image(images=origin['files'])
# Set active node destination directory
destination_act = destination['directory']
# Set standby node destination directory
if 'standby_directory' in destination:
destination_stby = destination['standby_directory']
else:
destination_stby = None
# Check remote server info present in testbed YAML
if not FileUtils.from_device(device).get_server_block(server):
log.error(banner("*** Terminating Genie Clean ***"))
section.failed("Server '{}' was provided in the clean yaml file but "
"doesn't exist in the testbed file.\n".format(server),
goto=['exit'])
# Check image files provided
if verify_num_images:
# Verify correct number of images provided
with steps.start("Verify correct number of images provided") as step:
if not verify_num_images_provided(image_list=image_files,
expected_images=expected_num_images):
log.error(banner("*** Terminating Genie Clean ***"))
section.failed("Incorrect number of images provided. Please "
"provide {} expected image(s) under destination"
".path in clean yaml file.\n".format(
expected_num_images), goto=['exit'])
def delete_unprotected_files(device, directory, protected, files_to_delete=None, dir_output=None):
"""delete all files not matching regex in the protected list
Args:
device ('obj'): Device object
directory ('str'): working directory to perform the operation
protected ('list'): list of file patterns that won't be deleted. If it begins
and ends with (), it will be considered as a regex
files_to_delete('list') list of files that should be deleted unless they are not protected
dir_output ('str'): output of dir command, if not provided execute the cmd on device to get the output
Returns:
None
"""
protected_set = set()
fu_device = FileUtils.from_device(device)
file_set = set(device.parse('dir {}'.format(directory), output=dir_output).get('files',{}).keys())
if isinstance(protected, str):
protected = [protected]
elif not isinstance(protected, (list, set)):
raise TypeError("'{p}' must be a list")
for pattern in protected:
# it's a regex!
if pattern.startswith('(') and pattern.endswith(')'):
regexp = re.compile(pattern)
protected_set.update(set(filter(regexp.match, file_set)))
# just file names, exact match only
elif pattern in file_set:
protected_set.add(pattern)
Args:
device ('Device'): Device object
remote_path ('str'): remote file path on the server
local_path ('str'): local file path to copy to on the device
server ('str'): hostname or address of the server
protocol('str'): file transfer protocol to be used
vrf ('str'): vrf to use (optional)
timeout('int'): timeout value in seconds, default 300
compact('bool'): compress image option for n9k, defaults False
use_kstack('bool'): Use faster version of copy, defaults False
Not supported with a file transfer protocol
prompting for a username and password
Returns:
None
"""
fu = FileUtils.from_device(device)
# build the source address
source = '{p}://{s}/{f}'.format(p=protocol, s=server, f=remote_path)
try:
if vrf:
fu.copyfile(source=source,
destination=local_path,
device=device,
vrf=vrf,
timeout_seconds=timeout,
compact=compact,
use_kstack=use_kstack,
**kwargs)
else:
fu.copyfile(source=source,
destination=local_path,