Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def tcl_package_require_caas_parsers():
tcl_package_require_caas()
tcl.call('package', 'require', 'IOS_Parser')
tcl.call('package', 'require', 'IOSXE_Parser')
tcl.call('package', 'require', 'NXOS_Parser')
tcl.call('package', 'require', 'IOSXR_Parser')
tcl.call('package', 'require', 'MULTIOS_Parser')
def try_cast_number(value):
if type(value) in (tuple, int, float, bool):
# Already a type recognized by Tk
return value
try:
return pyats.tcl.cast_int(value)
except (TypeError, ValueError):
pass
try:
return pyats.tcl.cast_double(value)
except (TypeError, ValueError):
pass
return tclstr(value)
def xml(self):
''' parsing mechanism: xml
Function xml() defines the xml type output parsing mechanism which
typically contains 3 steps: executing, transforming, returning
'''
output = tcl.q.caas.abstract(device=self.device.handle,
exec='show ip rip statistics | xml')
result = tcl.cast_any(output[1])
return result
def try_cast_number(value):
if type(value) in (tuple, int, float, bool):
# Already a type recognized by Tk
return value
try:
return pyats.tcl.cast_int(value)
except (TypeError, ValueError):
pass
try:
return pyats.tcl.cast_double(value)
except (TypeError, ValueError):
pass
return tclstr(value)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
tcl.call('package', 'require', 'IOSXR_Parser')
# --------------------------
tcl.q.set('test_params(useTGN)', tgn_enable)
tcl.q.set('test_params(TGNEnableARP)', tgntcl_enable_arp)
tcl.q.set('test_params(loadTGNAfterRouterConfig)', tgn_config_post_device_config)
tcl.q.set('test_params(waitAfterTGNRouting)', tgn_routing_threshold)
tcl.q.set('test_params(TGNWaitConvergence)', tgn_traffic_convergence_threshold)
tcl.q.set('test_params(TGNWaitTimeNoTrafficMs)', tgn_traffic_loss_recovery_threshold)
tcl.q.set('test_params(TGNWaitForRefRateMinutes)', tgn_profile_snapshot_threshold)
tcl.q.set('test_params(TGNWaitBeforeFirstSample)', tgn_first_sample_threshold)
tcl.q.set('test_params(TGNMinWaitForReferenceRate)', tgn_reference_rate_threshold)
tcl.q.set('test_params(TGNThreshold)', tgntcl_stream_sample_rate_percentage)
tcl.q.set('test_params(checkTrafficLoss)', tgn_enable_traffic_loss_check)
tcl.q.set('test_params(trafficLossPercentageTolerance)', tgn_traffic_loss_tolerance_percentage)
tcl.q.set('test_params(TGNStopTraffic)', tgn_disable_traffic_post_execution)
tcl.q.set('test_params(TGNLearnExcessAfterNSamples)', tgntcl_learn_after_n_samples)
tcl.q.set('test_params(TGNWaitMultiplier)', tgntcl_wait_multiplier)
# Set default values
# ------------------
tcl.q.set('test_params(TGNWaitInterval)', 1)
tcl.q.set('test_params(TGNReportWorstStreams)', 0)
tcl.q.set('test_params(TGNFeaturePatternMatch)', '{}')
tcl.q.set('test_params(TGNretries)', 10)
tcl.q.set('test_params(TGNarpTimeout)', 0)
tcl.q.set('test_params(TGNarpInterval)', 0)
tcl.q.set('test_params(TGNfIndex)', 0)
tcl.q.set('test_params(TGNiIndex)', 1)
tcl.q.set('test_params(TGNeIndex)', 2)
tcl.q.set('test_params(TGNtIndex)', 3)
tcl.q.set('test_params(TGNStatisticsSet)', '')
tcl.q.set('test_params(TGNPortStatisticsSet)', '')
if 'ENA_TESTS' in os.environ and os.path.isdir(os.environ['ENA_TESTS']):
psat_tgn = os.path.join(os.environ['ENA_TESTS'], 'psat/psat_tgn.tcl')
psat_lib = os.path.join(os.environ['ENA_TESTS'], 'psat/psat_lib.tcl')
else:
raise Exception("env(ENA_TESTS) not set or its path does not exist")
# Check for required env vars
required_env_vars = ['AUTOTEST', 'ATS_EASY', 'IXIA_HOME', 'IXIA_VERSION',
'IXIA_HLTAPI_LIBRARY', 'XBU_SHARED', 'TCLLIBPATH',
'TCL_PKG_PREFER_LATEST']
for env_var in required_env_vars:
if env_var not in os.environ:
raise Exception("env({}) has not been set".format(env_var))
# Required TCL Packages
tcl.q.package('require', 'cAAs')
tcl.q.package('require', 'psat-ng')
tcl.q.package('require', 'enaTgnUtils')
tcl.eval('namespace import -force ::enaTgnUtils::*')
# Source PSAT lib files
try:
tcl.q.source(psat_tgn)
tcl.q.source(psat_lib)
except:
raise Exception("Unable to source PSAT files required for TGN")
# Get arguments from TGN testbed YAML file
try:
address = self.connection_info['address']
controller = self.connection_info['controller']
handle = self.connection_info['handle']
# Need to get rid of the non-canonical string representations
# and recurse into children
if sort:
def _clean_cli(clie):
clie = pyats.tcl.cast_list(clie)
if len(clie) == 2:
return (clie[0], tuple(sorted(
(_clean_cli(e) for e in pyats.tcl.cast_list(clie[1])),
key=functools.cmp_to_key(_DictionaryCompare_index0))))
else:
return (clie[0], None)
tree = tuple(sorted(
(_clean_cli(e) for e in pyats.tcl.cast_list(cli)),
key=functools.cmp_to_key(_DictionaryCompare_index0)))
else:
def _clean_cli(clie):
clie = pyats.tcl.cast_list(clie)
if len(clie) == 2:
return (clie[0], tuple(
(_clean_cli(e) for e in pyats.tcl.cast_list(clie[1]))))
else:
return (clie[0], None)
tree = tuple(
(_clean_cli(e) for e in pyats.tcl.cast_list(cli)))
return tree
def tcl_invoke_ats_cmd(cmd, *, cast_=None, **kwargs):
cmd = TclCommand(cmd, keywords=kwargs, cast=tcl.cast_list)
result = cmd()
result_code = result[0]
result_msg = result[1] if len(result) == 2 else result[1:]
try:
result_code = tcl.cast_int(result[0])
except ValueError:
result_code = tclstr(result[0])
if result_code in ('passed', 1):
if cast_:
result_msg = cast_(result_msg)
return result_msg
else:
raise RuntimeError(tclstr(result_msg))
def cli(self):
''' parsing mechanism: cli
Function cli() defines the cli type output parsing mechanism which
typically contains 3 steps: executing, transforming, returning
'''
result = tcl.q.caas.abstract(device=self.device.handle,
exec='show ipv6 rip vrf all')
# # To leverage router_show parsers:
# result = tcl.q.router_show(device=device, cmd='show version')
return tcl.cast_any(result[1])