How to use the ncclient.transport.ssh.SSHSession function in ncclient

]]>]]>
    
    
    22450
    0
    31992
    0
    0
    0
    0
    0
    ]]>]]>"""]
            session = SSHSession(device_handler)
            session._connected = True
            session._channel = paramiko.Channel("c100")
            session.parser = session._device_handler.get_xml_parser(session)
            obj = ExecuteRpc(session, device_handler, raise_mode=RaiseMode.ALL)
            obj = ExecuteRpc(session, device_handler, raise_mode=RaiseMode.ALL)
            obj._filter_xml = ''
            session.run()
            resp = obj.request(rpc)._NCElement__doc[0]
            self.assertEqual(len(resp.xpath('pfe-traffic-statistics')), 1)

def test_filter_xml_delimiter_rpc_reply(self, mock_uuid4, mock_select,
                                            mock_session, mock_recv, mock_close,
                                            mock_send, mock_send_ready,
                                            mock_connected):
        mock_send.return_value = True
        mock_send_ready.return_value = -1
        mock_uuid4.return_value = type('dummy', (), {'urn': "urn:uuid:e0a7abe3-fffa-11e5-b78e-b8e85604f858"})
        device_handler = manager.make_device_handler({'name': 'junos', 'use_filter': True})
        rpc = ''
        mock_recv.side_effect = self._read_file('get-software-information.xml')[:-1] + [b"]]>", b"]]>"]
        session = SSHSession(device_handler)
        session._connected = True
        session._channel = paramiko.Channel("c100")
        session.parser = session._device_handler.get_xml_parser(session)
        obj = ExecuteRpc(session, device_handler, raise_mode=RaiseMode.ALL)
        obj._filter_xml = ''
        session.run()
        resp = obj.request(rpc)._NCElement__doc[0]
        self.assertEqual(len(resp.xpath('multi-routing-engine-item/re-name')), 2)
        self.assertEqual(len(resp.xpath('multi-routing-engine-item/software-information')), 0)

def test_auth_default_keyfiles(self, mock_get_key, mock_auth_public_key,
                                   mock_is_file):
        key = paramiko.PKey()
        mock_get_key.return_value = key
        mock_is_file.return_value = True
        device_handler = JunosDeviceHandler({'name': 'junos'})
        obj = SSHSession(device_handler)
        obj._transport = paramiko.Transport(MagicMock())
        obj._auth('user', 'password', [], False, True)
        self.assertEqual(
            (mock_auth_public_key.call_args_list[0][0][1]).__repr__(),
            key.__repr__())

def test_run_receive_py3(self, mock_error, mock_selector, mock_recv, mock_close):
        mock_selector.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_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_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_load_host_key(self, mock_load):
        device_handler = JunosDeviceHandler({'name': 'junos'})
        obj = SSHSession(device_handler)
        obj.load_known_hosts("file_name")
        mock_load.assert_called_once_with("file_name")

def test_close(self, mock_close):
        device_handler = JunosDeviceHandler({'name': 'junos'})
        obj = SSHSession(device_handler)
        obj._transport = paramiko.Transport(MagicMock())
        obj._transport.active = True
        obj._connected = True
        obj.close()
        mock_close.assert_called_once_with()
        self.assertFalse(obj._connected)

def __init__(self, device_handler):
        SSHSession.__init__(self, device_handler)
        self._host_keys = None
        self._transport = None
        self._connected = False
        self._channel = None
        self._channel_id = None
        self._channel_name = None
        self._buffer = compat.StringIO()  # for incoming data
        # parsing-related, see _parse()
        self._parsing_state = 0
        self._parsing_pos = 0
        self._device_handler = device_handler

import os
from select import select
from subprocess import Popen, PIPE, STDOUT

from ncclient import compat
from ncclient.transport.errors import SessionCloseError, TransportError
from ncclient.transport.ssh import SSHSession

MSG_DELIM = "]]>]]>"
TICK = 0.1
NETCONF_SHELL = 'xml-mode netconf need-trailer'


class IOProc(SSHSession):

    def __init__(self, device_handler):
        SSHSession.__init__(self, device_handler)
        self._host_keys = None
        self._transport = None
        self._connected = False
        self._channel = None
        self._channel_id = None
        self._channel_name = None
        self._buffer = compat.StringIO()  # for incoming data
        # parsing-related, see _parse()
        self._parsing_state = 0
        self._parsing_pos = 0
        self._device_handler = device_handler

    def close(self):