Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
% (1002, 'invalid command',
r'Invalid input \(privileged mode required\)')))
for dut in self.duts:
for (cmd, regex) in cases:
try:
# Insert the error in list of valid commands
if cmd != "show running-config":
dut.enable(['show version', cmd, 'show hostname'],
strict=True)
else:
dut.enable(['disable', 'show version', cmd],
strict=True, send_enable=False)
self.fail('A CommandError should have been raised')
except pyeapi.eapilib.CommandError as exc:
# Validate the properties of the exception
if cmd != 'show running-config':
self.assertEqual(len(exc.trace), 4)
else:
self.assertEqual(len(exc.trace), 3)
self.assertIsNotNone(exc.command_error)
self.assertIsNotNone(exc.output)
self.assertIsNotNone(exc.commands)
self.assertRegexpMatches(exc.message, regex)
def test_execute_valid_response(self):
response_dict = dict(jsonrpc='2.0', result=[], id=id(self))
mock_send = Mock(name='send')
mock_send.return_value = json.dumps(response_dict)
instance = pyeapi.eapilib.EapiConnection()
instance.send = mock_send
result = instance.execute(['command'])
self.assertEqual(json.loads(result), response_dict)
def test_create_https_connection(self):
instance = pyeapi.eapilib.HttpsEapiConnection('localhost')
self.assertIsInstance(instance, pyeapi.eapilib.EapiConnection)
self.assertIsNotNone(str(instance.transport))
def test_execute_socket_timeout_error(self, logexception):
for dut in self.duts:
self.assertEqual(dut.connection.transport.timeout, 60)
dut.connection.transport.timeout = 0.001
try:
dut.connection.execute(['show version'], encoding='json')
except pyeapi.eapilib.ConnectionError as err:
error_msg = 'Socket error during eAPI connection: timed out'
self.assertEqual(err.message, error_msg)
logexception.assert_called_once()
dut.connection.transport.timeout = 60
def test_command_error_trace(self):
commands = ['test command', 'test command', 'test command']
output = [{}, 'test output']
result = pyeapi.eapilib.CommandError(9999, 'test', commands=commands,
output=output)
self.assertIsNotNone(result.trace)
if isinstance(config, list):
lines = config
else:
lines = config.splitlines()
for line in lines:
line = line.strip()
if line == '':
continue
if line.startswith('!'):
continue
commands.append(line)
try:
self.device.run_commands(commands)
except pyeapi.eapilib.CommandError as e:
self.discard_config()
if replace:
raise ReplaceConfigException(e.message)
else:
raise MergeConfigException(e.message)
'''
Open the connection to the Arista switch over the eAPI.
'''
proxy_dict = opts.get('proxy', {})
conn_args = proxy_dict.copy()
conn_args.pop('proxytype', None)
opts['multiprocessing'] = conn_args.get('multiprocessing', True)
# This is not a SSH-based proxy, so it should be safe to enable
# multiprocessing.
try:
conn = pyeapi.client.connect(**conn_args)
node = pyeapi.client.Node(conn, enablepwd=conn_args.get('enablepwd'))
pyeapi_device['connection'] = node
pyeapi_device['initialized'] = True
pyeapi_device['up'] = True
except pyeapi.eapilib.ConnectionError as cerr:
log.error('Unable to connect to %s', conn_args['host'], exc_info=True)
return False
return True
continue
commands.append(line)
for start, depth in [
(s, d) for (s, d) in self.HEREDOC_COMMANDS if s in commands
]:
commands = self._multiline_convert(commands, start=start, depth=depth)
commands = self._mode_comment_convert(commands)
try:
if self.eos_autoComplete is not None:
self.device.run_commands(commands, autoComplete=self.eos_autoComplete)
else:
self.device.run_commands(commands)
except pyeapi.eapilib.CommandError as e:
self.discard_config()
msg = str(e)
if replace:
raise ReplaceConfigException(msg)
else:
raise MergeConfigException(msg)
def check_vlan_exists(eapi_conn, vlan_id):
"""
Check if the given VLAN exists
Return either vlan_name or False
"""
vlan_id = six.text_type(vlan_id)
cmd = 'show vlan id {}'.format(vlan_id)
try:
response = eapi_conn.enable(cmd)
check_vlan = pyeapi_result(response)['vlans']
return check_vlan[vlan_id]['name']
except (pyeapi.eapilib.CommandError, KeyError):
pass
return False