<s> import copy <EOL> import mock <EOL> import six <EOL> from openstack . cluster . v1 import receiver as sdk_receiver <EOL> from openstack import exceptions as sdk_exc <EOL> from openstackclient . common import exceptions as exc <EOL> from senlinclient . tests . unit . v1 import fakes <EOL> from senlinclient . v1 import receiver as osc_receiver <EOL> class TestReceiver ( fakes . TestClusteringv1 ) : <EOL> def setUp ( self ) : <EOL> super ( TestReceiver , self ) . setUp ( ) <EOL> self . mock_client = self . app . client_manager . clustering <EOL> class TestReceiverList ( TestReceiver ) : <EOL> columns = [ '<STR_LIT:id>' , '<STR_LIT:name>' , '<STR_LIT:type>' , '<STR_LIT>' , '<STR_LIT:action>' , '<STR_LIT>' ] <EOL> response = { "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:action>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ <EOL> "<STR_LIT>" <EOL> ] <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:count>" : "<STR_LIT:1>" <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : '<STR_LIT:null>' , <EOL> "<STR_LIT:user>" : "<STR_LIT>" <EOL> } <EOL> ] } <EOL> defaults = { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> } <EOL> def setUp ( self ) : <EOL> super ( TestReceiverList , self ) . setUp ( ) <EOL> self . cmd = osc_receiver . ListReceiver ( self . app , None ) <EOL> self . mock_client . receivers = mock . Mock ( <EOL> return_value = self . response ) <EOL> def test_receiver_list_defaults ( self ) : <EOL> arglist = [ ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> columns , data = self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . receivers . assert_called_with ( ** self . defaults ) <EOL> self . assertEqual ( self . columns , columns ) <EOL> def test_receiver_list_full_id ( self ) : <EOL> arglist = [ '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> columns , data = self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . receivers . assert_called_with ( ** self . defaults ) <EOL> self . assertEqual ( self . columns , columns ) <EOL> def test_receiver_list_limit ( self ) : <EOL> kwargs = copy . deepcopy ( self . defaults ) <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:3>' <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT:3>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> columns , data = self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . receivers . assert_called_with ( ** kwargs ) <EOL> self . assertEqual ( self . columns , columns ) <EOL> def test_receiver_list_sort ( self ) : <EOL> kwargs = copy . deepcopy ( self . defaults ) <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> columns , data = self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . receivers . assert_called_with ( ** kwargs ) <EOL> self . assertEqual ( self . columns , columns ) <EOL> def test_receiver_list_sort_invalid_key ( self ) : <EOL> self . mock_client . receivers = mock . Mock ( <EOL> return_value = self . response ) <EOL> kwargs = copy . deepcopy ( self . defaults ) <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . mock_client . receivers . side_effect = sdk_exc . HttpException ( ) <EOL> self . assertRaises ( sdk_exc . HttpException , <EOL> self . cmd . take_action , parsed_args ) <EOL> def test_receiver_list_sort_invalid_direction ( self ) : <EOL> self . mock_client . receivers = mock . Mock ( <EOL> return_value = self . response ) <EOL> kwargs = copy . deepcopy ( self . defaults ) <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . mock_client . receivers . side_effect = sdk_exc . HttpException ( ) <EOL> self . assertRaises ( sdk_exc . HttpException , <EOL> self . cmd . take_action , parsed_args ) <EOL> def test_receiver_list_filter ( self ) : <EOL> kwargs = copy . deepcopy ( self . defaults ) <EOL> kwargs [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> columns , data = self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . receivers . assert_called_with ( ** kwargs ) <EOL> self . assertEqual ( self . columns , columns ) <EOL> def test_receiver_list_marker ( self ) : <EOL> kwargs = copy . deepcopy ( self . defaults ) <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> columns , data = self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . receivers . assert_called_with ( ** kwargs ) <EOL> self . assertEqual ( self . columns , columns ) <EOL> class TestReceiverShow ( TestReceiver ) : <EOL> get_response = { "<STR_LIT>" : { <EOL> "<STR_LIT:action>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ <EOL> "<STR_LIT>" <EOL> ] <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:count>" : "<STR_LIT:1>" <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : '<STR_LIT:null>' , <EOL> "<STR_LIT:user>" : "<STR_LIT>" <EOL> } } <EOL> def setUp ( self ) : <EOL> super ( TestReceiverShow , self ) . setUp ( ) <EOL> self . cmd = osc_receiver . ShowReceiver ( self . app , None ) <EOL> self . mock_client . get_receiver = mock . Mock ( <EOL> return_value = sdk_receiver . Receiver ( <EOL> attrs = self . get_response [ '<STR_LIT>' ] ) ) <EOL> def test_receiver_show ( self ) : <EOL> arglist = [ '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . get_receiver . assert_called_with ( '<STR_LIT>' ) <EOL> def test_receiver_show_not_found ( self ) : <EOL> arglist = [ '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . mock_client . get_receiver . side_effect = sdk_exc . ResourceNotFound ( ) <EOL> error = self . assertRaises ( exc . CommandError , self . cmd . take_action , <EOL> parsed_args ) <EOL> self . assertEqual ( '<STR_LIT>' , str ( error ) ) <EOL> class TestReceiverCreate ( TestReceiver ) : <EOL> response = { "<STR_LIT>" : { <EOL> "<STR_LIT:action>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ <EOL> "<STR_LIT>" <EOL> ] <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:count>" : "<STR_LIT:1>" <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : '<STR_LIT:null>' , <EOL> "<STR_LIT:user>" : "<STR_LIT>" <EOL> } } <EOL> args = { <EOL> "<STR_LIT:action>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:count>" : "<STR_LIT:1>" <EOL> } , <EOL> "<STR_LIT:type>" : "<STR_LIT>" <EOL> } <EOL> def setUp ( self ) : <EOL> super ( TestReceiverCreate , self ) . setUp ( ) <EOL> self . cmd = osc_receiver . CreateReceiver ( self . app , None ) <EOL> self . mock_client . create_receiver = mock . Mock ( <EOL> return_value = sdk_receiver . Receiver ( <EOL> attrs = self . response [ '<STR_LIT>' ] ) ) <EOL> self . mock_client . get_receiver = mock . Mock ( <EOL> return_value = sdk_receiver . Receiver ( <EOL> attrs = self . response [ '<STR_LIT>' ] ) ) <EOL> def test_receiver_create ( self ) : <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . create_receiver . assert_called_with ( ** self . args ) <EOL> class TestReceiverDelete ( TestReceiver ) : <EOL> def setUp ( self ) : <EOL> super ( TestReceiverDelete , self ) . setUp ( ) <EOL> self . cmd = osc_receiver . DeleteReceiver ( self . app , None ) <EOL> self . mock_client . delete_receiver = mock . Mock ( ) <EOL> def test_receiver_delete ( self ) : <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . delete_receiver . assert_has_calls ( <EOL> [ mock . call ( '<STR_LIT>' , False ) , mock . call ( '<STR_LIT>' , False ) , <EOL> mock . call ( '<STR_LIT>' , False ) ] <EOL> ) <EOL> def test_receiver_delete_force ( self ) : <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . cmd . take_action ( parsed_args ) <EOL> self . mock_client . delete_receiver . assert_has_calls ( <EOL> [ mock . call ( '<STR_LIT>' , False ) , mock . call ( '<STR_LIT>' , False ) , <EOL> mock . call ( '<STR_LIT>' , False ) ] <EOL> ) <EOL> def test_receiver_delete_not_found ( self ) : <EOL> arglist = [ '<STR_LIT>' ] <EOL> self . mock_client . delete_receiver . side_effect = ( <EOL> sdk_exc . ResourceNotFound ) <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> error = self . assertRaises ( exc . CommandError , self . cmd . take_action , <EOL> parsed_args ) <EOL> self . assertIn ( '<STR_LIT>' , <EOL> str ( error ) ) <EOL> def test_receiver_delete_one_found_one_not_found ( self ) : <EOL> arglist = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . mock_client . delete_receiver . side_effect = ( <EOL> [ None , sdk_exc . ResourceNotFound ] <EOL> ) <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> error = self . assertRaises ( exc . CommandError , <EOL> self . cmd . take_action , parsed_args ) <EOL> self . mock_client . delete_receiver . assert_has_calls ( <EOL> [ mock . call ( '<STR_LIT>' , False ) , mock . call ( '<STR_LIT>' , False ) ] <EOL> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> str ( error ) ) <EOL> @ mock . patch ( '<STR_LIT>' , spec = six . StringIO ) <EOL> def test_receiver_delete_prompt_yes ( self , mock_stdin ) : <EOL> arglist = [ '<STR_LIT>' ] <EOL> mock_stdin . isatty . return_value = True <EOL> mock_stdin . readline . return_value = '<STR_LIT:y>' <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . cmd . take_action ( parsed_args ) <EOL> mock_stdin . readline . assert_called_with ( ) <EOL> self . mock_client . delete_receiver . assert_called_with ( '<STR_LIT>' , <EOL> False ) <EOL> @ mock . patch ( '<STR_LIT>' , spec = six . StringIO ) <EOL> def test_receiver_delete_prompt_no ( self , mock_stdin ) : <EOL> arglist = [ '<STR_LIT>' ] <EOL> mock_stdin . isatty . return_value = True <EOL> mock_stdin . readline . return_value = '<STR_LIT:n>' <EOL> parsed_args = self . check_parser ( self . cmd , arglist , [ ] ) <EOL> self . cmd . take_action ( parsed_args ) <EOL> mock_stdin . readline . assert_called_with ( ) <EOL> self . mock_client . delete_receiver . assert_not_called ( ) </s>
