Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def unshut_interface_adjacent_interfaces(
device, link_name, adjacent_interfaces=None, steps=Steps(), num=1
):
""" Unshut adjacent interfaces
Args:
device ('obj'): Device object
link_name ('str'): Interface alias in topology
num ('int'): Number of interfaces to return
adjacent_interfaces ('list'): List of EthernetInterface objects
steps ('obj'): Context manager object
Returns:
None
Raises:
SubCommandFailure
"""
def verify_bgp_as_path_prepend(
route_dict, as_path_prepend, as_path, device, route, steps=Steps()
):
""" Verify if AS value is added at the beginning of the AS path on route
Args:
route_dict ('obj') : Dict containing all routes on device
as_path_prepend ('int') : AS number to be checked on route
as_path ('int') : AS path on route
route ('str') : Route beeing checked
steps ('obj') : Context manager step
Returns:
None
Raises:
pyATS Results
"""
if not route_dict:
def action_parallel(self, steps, testbed, section, data):
# When called run all the actions
# below the keyword parallel concurently
pcall_payloads = []
with steps.start('Executing actions in parallel', continue_=True) as steps:
for action_item in data:
for action, action_kwargs in action_item.items():
# for future use - Enhancement needed in pyATS
# with steps.start("Implementing action '{a}' in parallel".format(a=actions)) as step:
# on parallel it is not possible to set continue to False and benefit from that feature
step = Steps()
kwargs = {'steps': step, 'testbed': testbed, 'section': section, 'data': [{action:action_kwargs}]}
pcall_payloads.append(kwargs)
pcall_returns = pcall(self.dispatcher, ikwargs=pcall_payloads)
# Each action return is a dictionary containing the action name, possible saved_variable
# Action results, and device name that action is being implemented on
# These value would be lost when the child processor that executes the action end the process.
# It is being implemented this way in order to add these values to the main processor.
for each_return in pcall_returns:
if each_return.get('saved_vars'):
for saved_var_name, saved_var_data in each_return.get('saved_vars').items():
if each_return.get('filters'):
log.info('Applied filter: {} to the action {} output'.format(each_return['filters'], action))
save_variable(self, saved_var_data, saved_var_name)
def verify_bgp_config_operational_state_summary(
device,
bgp_summary,
vrf,
address_family,
bgp_config="",
address_family_config="ipv4",
steps=Steps(),
):
""" Verify that bgp running config matches operational state from:
'show ip bgp {address_family} vrf {vrf} summary'
'show ip bgp {address_family} all summary'
Args:
device ('obj'): Device object
bgp_config ('dict'): Parsed output from libs.bgp.get.get_bgp_running_config
Example {'VRF1':
{'neighbors': {
'192.168.0.1': {
'remote_as': 65555}}}}
bgp_summary ('dict'): Parsed output from:
'show ip bgp {address_family} vrf {vrf} summary'
'show ip bgp {address_family} all summary'
def verify_bgp_rd_table(
device,
address_family,
vrf,
default_rd,
routes_list,
as_path_prepend,
as_path,
steps=Steps(),
):
"""Verify BGP table on neighbors
Args:
device ('obj') : Device object
address_family ('str') : Address family to be used in show command
routes_list ('dict') : Routes to be verified
vrf ('str') : VRF value
default_rd ('str') : Route distinguisher value
as_path_prepend ('str') : AS path prepend value to be checked on routes
as_path ('str') : AS path from routes
steps ('obj') : Context manager object
Raises:
Exception
Returns:
None
def verify_bgp_config_operational_state_neighbors(
device,
bgp_neighbors,
vrf,
bgp_config="",
steps=Steps(),
address_family_config="ipv4",
):
""" Verify that bgp running config matches operational state from:
'show ip bgp {address_family} vrf {vrf} neighbors'
'show ip bgp {address_family} all neighbors'
Args:
device ('obj'): Device object
bgp_config ('dict'): Parsed output from libs.bgp.get.get_bgp_running_config
Example {'VRF1':
{'neighbors': {
'192.168.0.1': {
'remote_as': 65555}}}}
bgp_neighbors ('dict'): Parsed output from:
'show ip bgp {address_family} vrf {vrf} neighbors'
'show ip bgp {address_family} all neighbors'
def verify_bgp_table_uut(
address_family,
device,
routes_list,
default_rd,
as_path,
as_path_prepend,
neighbor_address,
vrf=None,
steps=Steps(),
):
"""Verify BGP table on device under test
Args:
device ('obj') : Device object
address_family ('str') : Address family to be used in show command
vrf ('str') : VRF value of neighbors
as_path_prepend ('str') : AS path prepend value to be checked on routes
routes_list ('dict') : Routes to be verified
default_rd ('str') : Route distinguisher value
as_path_prepend ('str') : AS path prepend value to be checked on routes
as_path ('str') : AS path from routes
steps ('obj') : Context manager object
Raises:
Exception
Returns:
def shut_interface_adjacent_interfaces(
device, link_name, adjacent_interfaces=None, steps=Steps(), num=1
):
""" Shut adjacent interfaces
Args:
device ('obj'): Device object
link_name ('str'): Interface alias in topology
adjacent_interfaces ('list'): List of EthernetInterface objects
steps ('obj'): Context manager object
num ('int'): Number of interfaces to return
Returns:
None
Raises:
SubCommandFailure
def perform_issu(device, image, disk, timeout=1200, reconnect_via=None, steps=Steps()):
""" Execute ISSU on device
Args:
device ('obj'): Device object
image ('str'): Image name on disk
disk ('str'): Disk where is located image
timeout ('int'): Timeout in second for each section
Raise:
None
Returns:
None
"""
with steps.start("Command 'issu loadversion'") as step:
slot_number = get_platform_standby_rp(device=device)