Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_file_transfer(scp_file_transfer):
"""Test Netmiko file_transfer function."""
ssh_conn, file_system = scp_file_transfer
source_file = "test9.txt"
dest_file = "test9.txt"
direction = "put"
transfer_dict = file_transfer(
ssh_conn,
source_file=source_file,
dest_file=dest_file,
file_system=file_system,
direction=direction,
overwrite_file=True,
)
# No file on device at the beginning
assert (
transfer_dict["file_exists"]
and transfer_dict["file_transferred"]
and transfer_dict["file_verified"]
)
# File exists on device at this point
'device_type': 'arista_eos',
'host': 'localhost',
'username': 'arista',
'password': 'arista',
'file_system': '/mnt/flash',
'port': 2023
}
source_file = "test_file.txt"
dest_file = "test_file.txt"
direction = 'put'
for net_device in (eos1, eos2):
file_system = net_device.pop('file_system')
ssh_conn = ConnectHandler(**net_device)
transfer_dict = file_transfer(ssh_conn,
source_file=source_file,
dest_file=dest_file,
file_system=file_system,
direction=direction,
overwrite_file=True)
print(transfer_dict)
pause = input("Hit enter to continue: ")
print("-" * 50)
# Extract file and file system variables
file_system = net_device.pop('file_system')
source_file = net_device.pop('source_file')
dest_file = net_device.pop('dest_file')
# Establish SSH control channel
print(".establishing SSH connection.")
ssh_conn = ConnectHandler(**net_device)
# SCP new image file
print(".transferring image file.")
enable_transfer = True
if enable_transfer:
transfer_dict = file_transfer(ssh_conn, source_file=source_file, dest_file=dest_file,
file_system=file_system, direction='put',
overwrite_file=False)
else:
transfer_dict = {}
# Check the file exists and the MD5 matches the source file
if not transfer_dict.get('file_exists') or not transfer_dict.get('file_verified'):
raise ValueError("File doesn't exist or MD5 doesn't match on the remote system")
print(".verifying new image file.")
file_size = '42628912'
verify_image(ssh_conn, file_system, dest_file, file_size)
print()
print(".checking current boot commands")
boot_cmd = check_boot_var(ssh_conn)
}
for net_device in (cisco1,):
file_system = net_device.pop('file_system')
# Create the Netmiko SSH connection
ssh_conn = ConnectHandler(**net_device)
print(ssh_conn.find_prompt())
# Transfer the IOS image to device
source_file = "test1.bin"
dest_file = "test1.bin"
scp_transfer = False
if scp_transfer:
transfer_dict = file_transfer(
ssh_conn,
source_file=source_file,
dest_file=dest_file,
file_system=file_system,
direction='put',
overwrite_file=False,
)
if not transfer_dict['file_exists'] or not transfer_dict['file_verified']:
raise ValueError("The SCP file transfer failed.")
else:
transfer_dict = {}
print(transfer_dict)
any_key()
def load_replace_candidate(self, filename=None, config=None):
if not filename and not config:
raise ReplaceConfigException(
"filename or config parameter must be provided."
)
if not filename:
tmp_file = self._create_tmp_file(config)
filename = tmp_file
else:
if not os.path.isfile(filename):
raise ReplaceConfigException("File {} not found".format(filename))
try:
transfer_result = file_transfer(
self._netmiko_device,
source_file=filename,
dest_file=self.candidate_cfg,
file_system=self._dest_file_system,
direction="put",
overwrite_file=True,
)
if not transfer_result["file_exists"]:
raise ValueError()
except Exception:
msg = (
"Could not transfer file. There was an error "
"during transfer. Please make sure remote "
"permissions are set."
)
raise ReplaceConfigException(msg)
def job(self, run, payload, device):
netmiko_connection = run.netmiko_connection(device)
source = run.sub(run.source_file, locals())
destination = run.sub(run.destination_file, locals())
run.log("info", f"Transferring file {source}", device)
transfer_dict = file_transfer(
netmiko_connection,
source_file=source,
dest_file=destination,
file_system=run.file_system,
direction=run.direction,
overwrite_file=run.overwrite_file,
disable_md5=run.disable_md5,
inline_transfer=run.inline_transfer,
)
return {"success": True, "result": transfer_dict}
password = getpass()
cisco = {
"device_type": "cisco_ios",
"host": "cisco1.twb-tech.com",
"username": "pyclass",
"password": password,
}
source_file = "test1.txt"
dest_file = "test1.txt"
direction = "put"
file_system = "flash:"
ssh_conn = ConnectHandler(**cisco)
transfer_dict = file_transfer(
ssh_conn,
source_file=source_file,
dest_file=dest_file,
file_system=file_system,
direction=direction,
overwrite_file=True,
)
print(transfer_dict)
def job(self, run, payload, device):
netmiko_connection = run.netmiko_connection(device)
source = run.sub(run.source_file, locals())
destination = run.sub(run.destination_file, locals())
run.log("info", f"Transferring file {source}", device)
transfer_dict = file_transfer(
netmiko_connection,
source_file=source,
dest_file=destination,
file_system=run.file_system,
direction=run.direction,
overwrite_file=run.overwrite_file,
disable_md5=run.disable_md5,
inline_transfer=run.inline_transfer,
)
return {"success": True, "result": transfer_dict}