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_auth_keyfiles(self, mock_get_key, mock_auth_public_key):
key = paramiko.PKey()
mock_get_key.return_value = key
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
obj._transport = paramiko.Transport(MagicMock())
obj._auth('user', 'password', ["key_file_name"], False, True)
self.assertEqual(
(mock_auth_public_key.call_args_list[0][0][1]).__repr__(),
key.__repr__())
def test_auth_default_keyfiles_exception(self, mock_get_key,
mock_auth_public_key, mock_is_file):
key = paramiko.PKey()
mock_is_file.return_value = True
mock_get_key.side_effect = paramiko.ssh_exception.PasswordRequiredException
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
obj._transport = paramiko.Transport(MagicMock())
self.assertRaises(AuthenticationError,
obj._auth,'user', None, [], False, True)
def test_auth_keyfiles_exception(self, mock_get_key, mock_auth_public_key):
key = paramiko.PKey()
mock_get_key.side_effect = paramiko.ssh_exception.PasswordRequiredException
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
obj._transport = paramiko.Transport(MagicMock())
self.assertRaises(AuthenticationError,
obj._auth,'user', None, ["key_file_name"], False, True)
def test_make_device_handler_provided_handler(self):
device_handler = manager.make_device_handler(
{'handler': JunosDeviceHandler})
self.assertEqual(
device_handler.__class__.__name__,
"JunosDeviceHandler")
def test_run_send_py2(self, mock_error, mock_selector, mock_send, mock_ready, mock_close):
mock_selector.select.return_value = False
mock_ready.return_value = True
mock_send.return_value = -1
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
obj._channel = paramiko.Channel("c100")
obj._q.put("rpc")
obj.run()
self.assertEqual(mock_send.call_args_list[0][0][0], "rpc]]>]]>")
self.assertTrue(
isinstance(
mock_error.call_args_list[0][0][0],
SessionCloseError))
def test_dispatch_message_error(self, mock_log, mock_parse_root):
mock_parse_root.side_effect = Exception
cap = [':candidate']
obj = Session(cap)
device_handler = JunosDeviceHandler({'name': 'junos'})
obj._device_handler = device_handler
listener = HelloHandler(None, None)
obj._listeners.add(listener)
obj._dispatch_message(rpc_reply)
self.assertNotEqual(
mock_log.call_args_list[0][0][0].find("error parsing dispatch message"), -1)
def test_run_receive_py2(self, mock_error, mock_selector, mock_recv, mock_close):
mock_selector.select.return_value = True
mock_recv.return_value = 0
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
obj._channel = paramiko.Channel("c100")
obj.run()
self.assertTrue(
isinstance(
mock_error.call_args_list[0][0][0],
SessionCloseError))
def _test_parsemethod(self, mock_dispatch, parsemethod, reply, ok_chunk,
expected_messages):
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
if sys.version >= "3.0":
obj._buffer.write(bytes(reply, "utf-8"))
remainder = bytes(ok_chunk, "utf-8")
else:
obj._buffer.write(reply)
remainder = ok_chunk
parsemethod(obj)
for i in range(0, len(expected_messages)):
call = mock_dispatch.call_args_list[i][0][0]
self.assertEqual(call, expected_messages[i])
self.assertEqual(obj._buffer.getvalue(), remainder)
def test_parse11(self, mock_dispatch):
device_handler = JunosDeviceHandler({'name': 'junos'})
obj = SSHSession(device_handler)
if sys.version >= "3.0":
obj._buffer.write(bytes(rpc_reply11, "utf-8"))
remainder = bytes(reply_ok_partial_chunk, "utf-8")
else:
obj._buffer.write(rpc_reply11)
remainder = reply_ok_partial_chunk
obj.parser._parse11()
expected_messages = [reply_data, reply_ok]
for i in range(0, len(expected_messages)):
call = mock_dispatch.call_args_list[i][0][0]
self.assertEqual(call, expected_messages[i])
self.assertEqual(obj._buffer.getvalue(), remainder)
self._ssh_private_key_file = kvargs.get('ssh_private_key_file') \
or self._conf_ssh_private_key_file
self._timeout = kvargs.get('timeout', '0.5')
self._normalize = kvargs.get('normalize', False)
self._attempts = kvargs.get('attempts', 10)
self._gather_facts = kvargs.get('gather_facts', False)
self._fact_style = kvargs.get('fact_style', 'new')
self._huge_tree = kvargs.get('huge_tree', False)
if self._fact_style != 'new':
warnings.warn('fact-style %s will be removed in '
'a future release.' %
(self._fact_style), RuntimeWarning)
self.console_has_banner = kvargs.get('console_has_banner', False)
self.rpc = _RpcMetaExec(self)
self._manages = []
self.junos_dev_handler = JunosDeviceHandler(
device_params={'name': 'junos',
'local': False})
self._j2ldr = _Jinja2ldr
if self._fact_style == 'old':
self.facts = self.ofacts
else:
self.facts = _FactCache(self)