# Path: ncclient/devices/junos.py
# class JunosDeviceHandler(DefaultDeviceHandler):
#     """
#     Juniper handler for device specific information.
# 
#     """
# 
#     def __init__(self, device_params):
#         super(JunosDeviceHandler, self).__init__(device_params)
#         self.__reply_parsing_error_transform_by_cls = {
#             GetSchemaReply: fix_get_schema_reply
#         }
# 
#     def add_additional_operations(self):
#         dict = {}
#         dict["rpc"] = ExecuteRpc
#         dict["get_configuration"] = GetConfiguration
#         dict["load_configuration"] = LoadConfiguration
#         dict["compare_configuration"] = CompareConfiguration
#         dict["command"] = Command
#         dict["reboot"] = Reboot
#         dict["halt"] = Halt
#         dict["commit"] = Commit
#         dict["rollback"] = Rollback
#         return dict
# 
#     def perform_qualify_check(self):
#         return False
# 
#     def handle_raw_dispatch(self, raw):
#         if 'routing-engine' in raw:
#             raw = re.sub(r'<ok/>', '</routing-engine>\n<ok/>', raw)
#             return raw
#         # check if error is during capabilities exchange itself
#         elif re.search(r'<rpc-reply>.*?</rpc-reply>.*</hello>?', raw, re.M | re.S):
#             errs = re.findall(
#                 r'<rpc-error>.*?</rpc-error>', raw, re.M | re.S)
#             err_list = []
#             if errs:
#                 add_ns = """
#                         <xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
#                           <xsl:output indent="yes"/>
#                             <xsl:template match="*">
#                             <xsl:element name="{local-name()}" namespace="urn:ietf:params:xml:ns:netconf:base:1.0">
#                             <xsl:apply-templates select="@*|node()"/>
#                             </xsl:element>
#                           </xsl:template>
#                         </xsl:stylesheet>"""
#                 for err in errs:
#                     doc = etree.ElementTree(etree.XML(err))
#                     # Adding namespace using xslt
#                     xslt = etree.XSLT(etree.XML(add_ns))
#                     transformed_xml = etree.XML(etree.tostring(xslt(doc)))
#                     err_list.append(RPCError(transformed_xml))
#                 return RPCError(to_ele("<rpc-reply>"+''.join(errs)+"</rpc-reply>"), err_list)
#         else:
#             return False
# 
#     def handle_connection_exceptions(self, sshsession):
#         c = sshsession._channel = sshsession._transport.open_channel(
#             kind="session")
#         c.set_name("netconf-command-" + str(sshsession._channel_id))
#         c.exec_command("xml-mode netconf need-trailer")
#         return True
# 
#     def reply_parsing_error_transform(self, reply_cls):
#         # return transform function if found, else None
#         return self.__reply_parsing_error_transform_by_cls.get(reply_cls)
# 
#     def transform_reply(self):
#         reply = '''<xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
#         <xsl:output method="xml" indent="no"/>
# 
#         <xsl:template match="/|comment()|processing-instruction()">
#             <xsl:copy>
#                 <xsl:apply-templates/>
#             </xsl:copy>
#         </xsl:template>
# 
#         <xsl:template match="*">
#             <xsl:element name="{local-name()}">
#                 <xsl:apply-templates select="@*|node()"/>
#             </xsl:element>
#         </xsl:template>
# 
#         <xsl:template match="@*">
#             <xsl:attribute name="{local-name()}">
#                 <xsl:value-of select="."/>
#             </xsl:attribute>
#         </xsl:template>
#         </xsl:stylesheet>
#         '''
#         import sys
#         if sys.version < '3':
#             return reply
#         else:
#             return reply.encode('UTF-8')
# 
#     def get_xml_parser(self, session):
#         # use_filter in device_params can be used to enabled using SAX parsing
#         if self.device_params.get('use_filter', False):
#             l = session.get_listener_instance(SAXParserHandler)
#             if l:
#                 session.remove_listener(l)
#                 del l
#             session.add_listener(SAXParserHandler(session))
#             return JunosXMLParser(session)
#         else:
#             return DefaultXMLParser(session)


import unittest
import logging
    from unittest.mock import patch  # Python 3.4 and later
    from mock import patch
from ncclient.transport.session import *
from ncclient.devices.junos import JunosDeviceHandler
    from Queue import Queue, Empty
    from queue import Queue, Empty
      <eventTime>2007-07-08T00:01:00Z</eventTime>
      <event xmlns="http://example.com/event/1.0">
         <eventClass>fault</eventClass>
         <reportingEntity>
             <card>Ethernet0</card>
         </reportingEntity>
         <severity>major</severity>
       </event>
   </notification>
"""

class TestSession(unittest.TestCase):

    @patch('ncclient.transport.session.HelloHandler.callback')
    def test_dispatch_message(self, mock_handler):
        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)
        mock_handler.assert_called_once_with(parse_root(rpc_reply), rpc_reply)

    @patch('ncclient.transport.session.parse_root')
    @patch('ncclient.logging_.SessionLoggerAdapter.error')
    def test_dispatch_message_error(self, mock_log, mock_parse_root):
        mock_parse_root.side_effect = Exception
        cap = [':candidate']
        obj = Session(cap)

# Path: ncclient/devices/junos.py
# class JunosDeviceHandler(DefaultDeviceHandler):
#     """
#     Juniper handler for device specific information.
# 
#     """
# 
#     def __init__(self, device_params):
#         super(JunosDeviceHandler, self).__init__(device_params)
#         self.__reply_parsing_error_transform_by_cls = {
#             GetSchemaReply: fix_get_schema_reply
#         }
# 
#     def add_additional_operations(self):
#         dict = {}
#         dict["rpc"] = ExecuteRpc
#         dict["get_configuration"] = GetConfiguration
#         dict["load_configuration"] = LoadConfiguration
#         dict["compare_configuration"] = CompareConfiguration
#         dict["command"] = Command
#         dict["reboot"] = Reboot
#         dict["halt"] = Halt
#         dict["commit"] = Commit
#         dict["rollback"] = Rollback
#         return dict
# 
#     def perform_qualify_check(self):
#         return False
# 
#     def handle_raw_dispatch(self, raw):
#         if 'routing-engine' in raw:
#             raw = re.sub(r'<ok/>', '</routing-engine>\n<ok/>', raw)
#             return raw
#         # check if error is during capabilities exchange itself
#         elif re.search(r'<rpc-reply>.*?</rpc-reply>.*</hello>?', raw, re.M | re.S):
#             errs = re.findall(
#                 r'<rpc-error>.*?</rpc-error>', raw, re.M | re.S)
#             err_list = []
#             if errs:
#                 add_ns = """
#                         <xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
#                           <xsl:output indent="yes"/>
#                             <xsl:template match="*">
#                             <xsl:element name="{local-name()}" namespace="urn:ietf:params:xml:ns:netconf:base:1.0">
#                             <xsl:apply-templates select="@*|node()"/>
#                             </xsl:element>
#                           </xsl:template>
#                         </xsl:stylesheet>"""
#                 for err in errs:
#                     doc = etree.ElementTree(etree.XML(err))
#                     # Adding namespace using xslt
#                     xslt = etree.XSLT(etree.XML(add_ns))
#                     transformed_xml = etree.XML(etree.tostring(xslt(doc)))
#                     err_list.append(RPCError(transformed_xml))
#                 return RPCError(to_ele("<rpc-reply>"+''.join(errs)+"</rpc-reply>"), err_list)
#         else:
#             return False
# 
#     def handle_connection_exceptions(self, sshsession):
#         c = sshsession._channel = sshsession._transport.open_channel(
#             kind="session")
#         c.set_name("netconf-command-" + str(sshsession._channel_id))
#         c.exec_command("xml-mode netconf need-trailer")
#         return True
# 
#     def reply_parsing_error_transform(self, reply_cls):
#         # return transform function if found, else None
#         return self.__reply_parsing_error_transform_by_cls.get(reply_cls)
# 
#     def transform_reply(self):
#         reply = '''<xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
#         <xsl:output method="xml" indent="no"/>
# 
#         <xsl:template match="/|comment()|processing-instruction()">
#             <xsl:copy>
#                 <xsl:apply-templates/>
#             </xsl:copy>
#         </xsl:template>
# 
#         <xsl:template match="*">
#             <xsl:element name="{local-name()}">
#                 <xsl:apply-templates select="@*|node()"/>
#             </xsl:element>
#         </xsl:template>
# 
#         <xsl:template match="@*">
#             <xsl:attribute name="{local-name()}">
#                 <xsl:value-of select="."/>
#             </xsl:attribute>
#         </xsl:template>
#         </xsl:stylesheet>
#         '''
#         import sys
#         if sys.version < '3':
#             return reply
#         else:
#             return reply.encode('UTF-8')
# 
#     def get_xml_parser(self, session):
#         # use_filter in device_params can be used to enabled using SAX parsing
#         if self.device_params.get('use_filter', False):
#             l = session.get_listener_instance(SAXParserHandler)
#             if l:
#                 session.remove_listener(l)
#                 del l
#             session.add_listener(SAXParserHandler(session))
#             return JunosXMLParser(session)
#         else:
#             return DefaultXMLParser(session)


# Path: test/unit/transport/test_session.py
import unittest
import logging
    from unittest.mock import patch  # Python 3.4 and later
    from mock import patch
from ncclient.transport.session import *
from ncclient.devices.junos import JunosDeviceHandler
    from Queue import Queue, Empty
    from queue import Queue, Empty

      <eventTime>2007-07-08T00:01:00Z</eventTime>
      <event xmlns="http://example.com/event/1.0">
         <eventClass>fault</eventClass>
         <reportingEntity>
             <card>Ethernet0</card>
         </reportingEntity>
         <severity>major</severity>
       </event>
   </notification>
"""

class TestSession(unittest.TestCase):

    @patch('ncclient.transport.session.HelloHandler.callback')
    def test_dispatch_message(self, mock_handler):
        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)
        mock_handler.assert_called_once_with(parse_root(rpc_reply), rpc_reply)

    @patch('ncclient.transport.session.parse_root')
    @patch('ncclient.logging_.SessionLoggerAdapter.error')
    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'})