<s> import unittest <EOL> from mock import patch <EOL> from ncclient . transport . session import * <EOL> from ncclient . devices . junos import JunosDeviceHandler <EOL> rpc_reply = """<STR_LIT>""" <EOL> hello_rpc_reply = """<STR_LIT>""" <EOL> class TestSession ( unittest . TestCase ) : <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_dispatch_message ( self , mock_handler ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> device_handler = JunosDeviceHandler ( { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> obj . _device_handler = device_handler <EOL> listener = HelloHandler ( None , None ) <EOL> obj . _listeners . add ( listener ) <EOL> obj . _dispatch_message ( rpc_reply ) <EOL> mock_handler . assert_called_once_with ( parse_root ( rpc_reply ) , rpc_reply ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_dispatch_message_error ( self , mock_log , mock_parse_root ) : <EOL> mock_parse_root . side_effect = Exception <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> device_handler = JunosDeviceHandler ( { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> obj . _device_handler = device_handler <EOL> listener = HelloHandler ( None , None ) <EOL> obj . _listeners . add ( listener ) <EOL> obj . _dispatch_message ( rpc_reply ) <EOL> self . assertNotEqual ( <EOL> mock_log . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . find ( "<STR_LIT>" ) , - <NUM_LIT:1> ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_dispatch_error ( self , mock_handler ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> listener = HelloHandler ( None , None ) <EOL> obj . _listeners . add ( listener ) <EOL> obj . _dispatch_error ( "<STR_LIT>" ) <EOL> mock_handler . assert_called_once_with ( "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_post_connect ( self , mock_lock , mock_handler , mock_log ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> device_handler = JunosDeviceHandler ( { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> obj . _device_handler = device_handler <EOL> obj . _connected = True <EOL> obj . _id = <NUM_LIT:100> <EOL> obj . _server_capabilities = cap <EOL> obj . _post_connect ( ) <EOL> log_call = mock_log . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertNotEqual ( log_call . find ( "<STR_LIT>" ) , - <NUM_LIT:1> ) <EOL> self . assertNotEqual ( log_call . find ( "<STR_LIT>" ) , - <NUM_LIT:1> ) <EOL> self . assertNotEqual ( <EOL> log_call . find ( "<STR_LIT>" ) , - <NUM_LIT:1> ) <EOL> def test_add_listener ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> listener = HelloHandler ( None , None ) <EOL> obj . add_listener ( listener ) <EOL> self . assertEqual ( obj . _listeners , { listener } ) <EOL> def test_add_listener_exception ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> listener = Session ( None ) <EOL> with self . assertRaises ( SessionError ) : <EOL> obj . add_listener ( listener ) <EOL> def test_remove_listener ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> listener = HelloHandler ( None , None ) <EOL> obj . _listeners . add ( listener ) <EOL> obj . remove_listener ( listener ) <EOL> self . assertEqual ( len ( obj . _listeners ) , <NUM_LIT:0> ) <EOL> def test_get_listener_instance ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> listener = HelloHandler ( None , None ) <EOL> obj . _listeners . add ( listener ) <EOL> ret = obj . get_listener_instance ( HelloHandler ) <EOL> self . assertEqual ( ret , listener ) <EOL> def test_send_connected ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> obj . _connected = True <EOL> obj . send ( "<STR_LIT>" ) <EOL> self . assertEqual ( "<STR_LIT>" , obj . _q . get ( ) ) <EOL> def test_send_disconnected ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> obj . _connected = False <EOL> with self . assertRaises ( TransportError ) : <EOL> obj . send ( "<STR_LIT>" ) <EOL> def test_connected ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> obj . _connected = True <EOL> self . assertTrue ( obj . connected ) <EOL> def test_client_capability ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> self . assertEqual ( obj . client_capabilities , cap ) <EOL> def test_server_capability ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> obj . _server_capabilities = cap <EOL> self . assertEqual ( obj . server_capabilities , cap ) <EOL> def test_id ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> obj . _id = "<STR_LIT>" <EOL> self . assertEqual ( obj . id , "<STR_LIT>" ) <EOL> def test_hello_handler_callback_ok ( self ) : <EOL> def ok_cb ( id , capabilities ) : <EOL> self . _id = id <EOL> self . _server_capabilities = capabilities <EOL> def err_cb ( err ) : <EOL> self . error = err <EOL> listener = HelloHandler ( ok_cb , err_cb ) <EOL> listener . callback ( parse_root ( hello_rpc_reply ) , hello_rpc_reply ) <EOL> caps = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> self . assertEqual ( self . _id , "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> self . _server_capabilities . _dict , <EOL> Capabilities ( caps ) . _dict ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_hello_handler_callback_error ( self , mock_parse ) : <EOL> def ok_cb ( id , capabilities ) : <EOL> self . _id = id <EOL> self . _server_capabilities = capabilities <EOL> def err_cb ( err ) : <EOL> self . error = "<STR_LIT>" <EOL> mock_parse . side_effect = Exception <EOL> listener = HelloHandler ( ok_cb , err_cb ) <EOL> listener . callback ( parse_root ( hello_rpc_reply ) , hello_rpc_reply ) <EOL> self . assertEqual ( self . error , "<STR_LIT>" ) <EOL> def test_hello_handler_build ( self ) : <EOL> cap = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> listener = HelloHandler ( None , None ) <EOL> result = listener . build ( cap , None ) <EOL> node = new_ele ( "<STR_LIT:hello>" ) <EOL> caps = sub_ele ( node , "<STR_LIT>" ) <EOL> sub_ele ( caps , "<STR_LIT>" ) . text = "<STR_LIT>" <EOL> sub_ele ( caps , "<STR_LIT>" ) . text = "<STR_LIT>" <EOL> node = to_xml ( node ) <EOL> self . assertEqual ( node , result ) <EOL> def test_hello_handler_parse ( self ) : <EOL> cap = [ '<STR_LIT>' ] <EOL> obj = Session ( cap ) <EOL> listener = HelloHandler ( None , None ) <EOL> obj . _listeners . add ( listener ) <EOL> id , capabilities = listener . parse ( hello_rpc_reply ) <EOL> self . assertEqual ( id , "<STR_LIT>" ) <EOL> caps = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> self . assertEqual ( capabilities . _dict , Capabilities ( caps ) . _dict ) </s>
