<s> from __future__ import absolute_import , unicode_literals <EOL> from kombu . async . http . curl import READ , WRITE , CurlClient <EOL> from kombu . tests . case import ( <EOL> HubCase , Mock , call , patch , set_module_symbol , skip , <EOL> ) <EOL> @ skip . if_pypy ( ) <EOL> @ skip . unless_module ( '<STR_LIT>' ) <EOL> class test_CurlClient ( HubCase ) : <EOL> class Client ( CurlClient ) : <EOL> Curl = Mock ( name = '<STR_LIT>' ) <EOL> def test_when_pycurl_missing ( self ) : <EOL> with set_module_symbol ( '<STR_LIT>' , '<STR_LIT>' , None ) : <EOL> with self . assertRaises ( ImportError ) : <EOL> self . Client ( ) <EOL> def test_max_clients_set ( self ) : <EOL> x = self . Client ( max_clients = <NUM_LIT> ) <EOL> self . assertEqual ( x . max_clients , <NUM_LIT> ) <EOL> def test_init ( self ) : <EOL> with patch ( '<STR_LIT>' ) as _pycurl : <EOL> x = self . Client ( ) <EOL> self . assertIsNotNone ( x . _multi ) <EOL> self . assertIsNotNone ( x . _pending ) <EOL> self . assertIsNotNone ( x . _free_list ) <EOL> self . assertIsNotNone ( x . _fds ) <EOL> self . assertEqual ( <EOL> x . _socket_action , x . _multi . socket_action , <EOL> ) <EOL> self . assertEqual ( len ( x . _curls ) , x . max_clients ) <EOL> self . assertTrue ( x . _timeout_check_tref ) <EOL> x . _multi . setopt . assert_has_calls ( [ <EOL> call ( _pycurl . M_TIMERFUNCTION , x . _set_timeout ) , <EOL> call ( _pycurl . M_SOCKETFUNCTION , x . _handle_socket ) , <EOL> ] ) <EOL> def test_close ( self ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> x = self . Client ( ) <EOL> x . _timeout_check_tref = Mock ( name = '<STR_LIT>' ) <EOL> x . close ( ) <EOL> x . _timeout_check_tref . cancel . assert_called_with ( ) <EOL> for _curl in x . _curls : <EOL> _curl . close . assert_called_with ( ) <EOL> x . _multi . close . assert_called_with ( ) <EOL> def test_add_request ( self ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> x = self . Client ( ) <EOL> x . _process_queue = Mock ( name = '<STR_LIT>' ) <EOL> x . _set_timeout = Mock ( name = '<STR_LIT>' ) <EOL> request = Mock ( name = '<STR_LIT>' ) <EOL> x . add_request ( request ) <EOL> self . assertIn ( request , x . _pending ) <EOL> x . _process_queue . assert_called_with ( ) <EOL> x . _set_timeout . assert_called_with ( <NUM_LIT:0> ) <EOL> def test_handle_socket ( self ) : <EOL> with patch ( '<STR_LIT>' ) as _pycurl : <EOL> hub = Mock ( name = '<STR_LIT>' ) <EOL> x = self . Client ( hub ) <EOL> fd = Mock ( name = '<STR_LIT>' ) <EOL> x . _fds [ fd ] = fd <EOL> x . _handle_socket ( _pycurl . POLL_REMOVE , fd , x . _multi , None , _pycurl ) <EOL> hub . remove . assert_called_with ( fd ) <EOL> self . assertNotIn ( fd , x . _fds ) <EOL> x . _handle_socket ( _pycurl . POLL_REMOVE , fd , x . _multi , None , _pycurl ) <EOL> hub = x . hub = Mock ( name = '<STR_LIT>' ) <EOL> fds = [ fd , Mock ( name = '<STR_LIT>' ) , Mock ( name = '<STR_LIT>' ) ] <EOL> x . _fds = { f : f for f in fds } <EOL> x . _handle_socket ( _pycurl . POLL_IN , fd , x . _multi , None , _pycurl ) <EOL> hub . remove . assert_has_calls ( [ call ( fd ) ] ) <EOL> hub . add_reader . assert_called_with ( fd , x . on_readable , fd ) <EOL> self . assertEqual ( x . _fds [ fd ] , READ ) <EOL> hub = x . hub = Mock ( name = '<STR_LIT>' ) <EOL> x . _handle_socket ( _pycurl . POLL_OUT , fd , x . _multi , None , _pycurl ) <EOL> hub . add_writer . assert_called_with ( fd , x . on_writable , fd ) <EOL> self . assertEqual ( x . _fds [ fd ] , WRITE ) <EOL> hub = x . hub = Mock ( name = '<STR_LIT>' ) <EOL> x . _handle_socket ( _pycurl . POLL_INOUT , fd , x . _multi , None , _pycurl ) <EOL> hub . add_reader . assert_called_with ( fd , x . on_readable , fd ) <EOL> hub . add_writer . assert_called_with ( fd , x . on_writable , fd ) <EOL> self . assertEqual ( x . _fds [ fd ] , READ | WRITE ) <EOL> hub = x . hub = Mock ( name = '<STR_LIT>' ) <EOL> x . _handle_socket ( <NUM_LIT> , fd , x . _multi , None , _pycurl ) <EOL> hub = x . hub = Mock ( name = '<STR_LIT>' ) <EOL> x . _fds . clear ( ) <EOL> x . _handle_socket ( <NUM_LIT> , fd , x . _multi , None , _pycurl ) <EOL> hub . remove . assert_not_called ( ) <EOL> def test_set_timeout ( self ) : <EOL> x = self . Client ( ) <EOL> x . _set_timeout ( <NUM_LIT:100> ) <EOL> def test_timeout_check ( self ) : <EOL> with patch ( '<STR_LIT>' ) as _pycurl : <EOL> x = self . Client ( ) <EOL> x . _process_pending_requests = Mock ( name = '<STR_LIT>' ) <EOL> x . _multi . socket_all . return_value = <NUM_LIT> , <NUM_LIT:1> <EOL> _pycurl . error = KeyError <EOL> x . _timeout_check ( _pycurl = _pycurl ) <EOL> x . _multi . socket_all . return_value = None <EOL> x . _multi . socket_all . side_effect = _pycurl . error ( <NUM_LIT> ) <EOL> x . _timeout_check ( _pycurl = _pycurl ) <EOL> def test_on_readable_on_writeable ( self ) : <EOL> with patch ( '<STR_LIT>' ) as _pycurl : <EOL> x = self . Client ( ) <EOL> x . _on_event = Mock ( name = '<STR_LIT>' ) <EOL> fd = Mock ( name = '<STR_LIT>' ) <EOL> x . on_readable ( fd , _pycurl = _pycurl ) <EOL> x . _on_event . assert_called_with ( fd , _pycurl . CSELECT_IN ) <EOL> x . on_writable ( fd , _pycurl = _pycurl ) <EOL> x . _on_event . assert_called_with ( fd , _pycurl . CSELECT_OUT ) </s>
