<s> from oslo_serialization import jsonutils <EOL> from nailgun import consts <EOL> from nailgun . db . sqlalchemy . models import IPAddr <EOL> from nailgun . db . sqlalchemy . models import NetworkGroup <EOL> from nailgun import objects <EOL> from nailgun . utils import reverse <EOL> from nailgun . test . base import BaseIntegrationTest <EOL> class BaseIPAddrTest ( BaseIntegrationTest ) : <EOL> __test__ = False <EOL> def setUp ( self ) : <EOL> self . maxDiff = None <EOL> super ( BaseIPAddrTest , self ) . setUp ( ) <EOL> self . cluster = self . env . create ( <EOL> release_kwargs = { '<STR_LIT:version>' : '<STR_LIT>' } , <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT>' : [ '<STR_LIT>' ] } ] <EOL> ) <EOL> net_manager = objects . Cluster . get_network_manager ( self . cluster ) <EOL> net_manager . assign_vips_for_net_groups ( self . cluster ) <EOL> self . management_net = net_manager . get_network_by_netname ( <EOL> consts . NETWORKS . management , self . cluster . network_groups <EOL> ) <EOL> self . management_vips = self . db . query ( IPAddr ) . filter ( <EOL> IPAddr . vip_name . isnot ( None ) , <EOL> IPAddr . network == self . management_net . id <EOL> ) . all ( ) <EOL> self . vip_ids = [ v . id for v in self . management_vips ] <EOL> self . expected_vips = [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> ] <EOL> self . non_existing_id = <NUM_LIT> <EOL> def _remove_from_response ( self , response , fields ) : <EOL> list_given = isinstance ( response , list ) <EOL> if not list_given : <EOL> response = [ response ] <EOL> clean_response = [ ] <EOL> for resp_item in response : <EOL> resp_item_clone = resp_item . copy ( ) <EOL> for f in fields : <EOL> resp_item_clone . pop ( f , None ) <EOL> clean_response . append ( resp_item_clone ) <EOL> return clean_response if list_given else clean_response [ <NUM_LIT:0> ] <EOL> def _check_ip_intersection ( self , ip_addr ) : <EOL> handlers_info = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] . id <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : ip_addr , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] } , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : self . management_vips [ <NUM_LIT:0> ] . id , <EOL> '<STR_LIT>' : ip_addr , <EOL> '<STR_LIT>' : True <EOL> } <EOL> ] <EOL> } <EOL> } <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = handlers_info [ self . handler_name ] [ '<STR_LIT>' ] , <EOL> ) , <EOL> params = jsonutils . dumps ( <EOL> handlers_info [ self . handler_name ] [ '<STR_LIT>' ] ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> net_group = self . db . query ( NetworkGroup ) . filter_by ( id = self . management_net . id ) . first ( ) <EOL> err_msg = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> . format ( ip_addr , net_group . name , net_group . cidr ) <EOL> ) <EOL> self . assertIn ( err_msg , resp . json_body [ '<STR_LIT:message>' ] ) <EOL> def test_update_user_defined_fail_if_ip_addr_intersection ( self ) : <EOL> intersecting_vip = self . management_vips [ <NUM_LIT:1> ] . ip_addr <EOL> self . _check_ip_intersection ( intersecting_vip ) <EOL> class TestIPAddrList ( BaseIPAddrTest ) : <EOL> __test__ = True <EOL> handler_name = '<STR_LIT>' <EOL> def check_create_vip ( self , create_data ) : <EOL> resp = self . app . post ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] } <EOL> ) , <EOL> params = jsonutils . dumps ( create_data ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> return resp <EOL> def test_create_vip_for_cluster ( self ) : <EOL> create_data = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : self . management_net . id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True <EOL> } <EOL> self . check_create_vip ( create_data ) <EOL> def test_create_vip_for_cluster_wo_is_user_defined_flag ( self ) : <EOL> create_data = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : self . management_net . id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> create_resp = self . check_create_vip ( create_data ) <EOL> get_resp = self . app . get ( <EOL> reverse ( <EOL> '<STR_LIT>' , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : create_resp . json_body [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertTrue ( get_resp . json_body [ '<STR_LIT>' ] ) <EOL> def test_vips_list_for_cluster ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] } <EOL> ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertItemsEqual ( <EOL> self . expected_vips , <EOL> self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT:id>' , '<STR_LIT>' ] <EOL> ) , <EOL> ) <EOL> def test_vips_list_with_two_clusters ( self ) : <EOL> self . second_cluster = self . env . create_cluster ( api = False ) <EOL> self . env . create_ip_addrs_by_rules ( <EOL> self . second_cluster , <EOL> { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> ) <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] } <EOL> ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertItemsEqual ( <EOL> self . expected_vips , <EOL> self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT:id>' , '<STR_LIT>' ] <EOL> ) , <EOL> ) <EOL> def test_wrong_cluster ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { '<STR_LIT>' : <NUM_LIT> } <EOL> ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_update ( self ) : <EOL> update_data = [ <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ] <EOL> expected_data = [ <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] [ "<STR_LIT>" ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:1> ] [ "<STR_LIT>" ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ] <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( update_data ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertEqual ( expected_data , self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) ) <EOL> def test_update_fail_with_no_id ( self ) : <EOL> new_data = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ] <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertIn ( "<STR_LIT>" , resp . json_body [ "<STR_LIT:message>" ] ) <EOL> def test_update_fail_with_non_updatable_field ( self ) : <EOL> new_data = [ <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . non_existing_id <EOL> } <EOL> ] <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertIn ( "<STR_LIT>" , resp . json_body [ "<STR_LIT:message>" ] ) <EOL> def test_update_pass_with_non_updatable_not_changed_field ( self ) : <EOL> new_data = [ <EOL> { <EOL> '<STR_LIT:id>' : self . management_vips [ <NUM_LIT:0> ] . id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] . network <EOL> } <EOL> ] <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> def test_update_fail_with_not_found_id ( self ) : <EOL> new_data = [ { <EOL> '<STR_LIT:id>' : self . non_existing_id , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ] <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertIn ( str ( self . non_existing_id ) , resp . json_body [ "<STR_LIT:message>" ] ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_update_fail_on_dict_request ( self ) : <EOL> new_data = { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_update_failing_on_empty_request ( self ) : <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = "<STR_LIT>" , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_update_not_failing_on_empty_list_request ( self ) : <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( [ ] ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> def test_update_fail_on_wrong_fields ( self ) : <EOL> new_data_suites = [ <EOL> [ { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : self . non_existing_id , <EOL> "<STR_LIT>" : "<STR_LIT:value>" <EOL> } ] <EOL> ] <EOL> for new_data in new_data_suites : <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_update_fail_on_ip_from_wrong_cluster ( self ) : <EOL> another_cluster = self . env . create_cluster ( api = False ) <EOL> new_data_suites = [ <EOL> [ { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : self . non_existing_id <EOL> } ] <EOL> ] <EOL> for new_data in new_data_suites : <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : another_cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertIn ( <EOL> "<STR_LIT>" , <EOL> resp . json_body . get ( "<STR_LIT:message>" ) <EOL> ) <EOL> def test_all_data_not_changed_on_single_error ( self ) : <EOL> old_all_addr = [ dict ( a ) for a in self . db . query ( IPAddr ) . all ( ) ] <EOL> self . db . commit ( ) <EOL> new_data = [ <EOL> { <EOL> '<STR_LIT:id>' : self . vip_ids [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ] <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( new_data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> new_all_addr = [ dict ( a ) for a in self . db . query ( IPAddr ) . all ( ) ] <EOL> self . assertEqual ( old_all_addr , new_all_addr ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_ipaddr_filter_by_network_id ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = { "<STR_LIT>" : self . management_vips [ <NUM_LIT:0> ] [ '<STR_LIT>' ] } , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( [ dict ( v ) for v in self . management_vips ] , <EOL> resp . json_body ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> def test_ipaddr_filter_by_missing_network_id ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = { "<STR_LIT>" : self . non_existing_id } , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( [ ] , resp . json_body ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> def test_ipaddr_filter_by_network_role ( self ) : <EOL> public_net_vip_names = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = self . default_headers <EOL> ) <EOL> expected_vips = [ vip for vip in self . expected_vips <EOL> if vip [ '<STR_LIT>' ] in public_net_vip_names ] <EOL> self . assertItemsEqual ( <EOL> expected_vips , <EOL> self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT:id>' , '<STR_LIT>' ] <EOL> ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> def test_ipaddr_filter_by_missing_network_role ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( [ ] , resp . json_body ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> class TestIPAddrHandler ( BaseIPAddrTest ) : <EOL> __test__ = True <EOL> handler_name = '<STR_LIT>' <EOL> def test_get_ip_addr ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . vip_ids [ <NUM_LIT:0> ] <EOL> } <EOL> ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertIn ( <EOL> self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT:id>' , '<STR_LIT>' ] <EOL> ) , <EOL> self . expected_vips <EOL> ) <EOL> def test_delete_fail ( self ) : <EOL> resp = self . app . delete ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . vip_ids [ <NUM_LIT:0> ] <EOL> } <EOL> ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_fail_on_no_vip_metadata ( self ) : <EOL> not_vip_ip_addr = IPAddr ( <EOL> network = self . cluster . network_groups [ <NUM_LIT:0> ] . id , <EOL> ip_addr = "<STR_LIT:127.0.0.1>" , <EOL> vip_name = None <EOL> ) <EOL> self . db . add ( not_vip_ip_addr ) <EOL> self . db . flush ( ) <EOL> not_vip_id = not_vip_ip_addr . get ( '<STR_LIT:id>' ) <EOL> resp = self . app . get ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : not_vip_id <EOL> } <EOL> ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True <EOL> ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_update_ip_addr ( self ) : <EOL> update_data = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> expected_data = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( update_data ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertEqual ( <EOL> expected_data , <EOL> self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> ) <EOL> def test_update_ip_addr_with_put ( self ) : <EOL> update_data = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> expected_data = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> resp = self . app . put ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . management_vips [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( update_data ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertEqual ( <EOL> expected_data , <EOL> self . _remove_from_response ( <EOL> resp . json_body , <EOL> [ '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> ) <EOL> def test_vip_namespase_update ( self ) : <EOL> changed_vip = self . management_vips [ <NUM_LIT:0> ] <EOL> update_data = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> self . handler_name , <EOL> kwargs = { <EOL> '<STR_LIT>' : self . cluster [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : changed_vip [ '<STR_LIT:id>' ] <EOL> } <EOL> ) , <EOL> params = jsonutils . dumps ( update_data ) , <EOL> headers = self . default_headers <EOL> ) <EOL> resp = self . env . neutron_networks_get ( self . cluster . id ) <EOL> net_conf = resp . json_body <EOL> self . assertIn ( changed_vip [ '<STR_LIT>' ] , net_conf [ '<STR_LIT>' ] ) <EOL> vip = net_conf [ '<STR_LIT>' ] [ changed_vip [ '<STR_LIT>' ] ] <EOL> self . assertEqual ( vip [ '<STR_LIT>' ] , update_data [ '<STR_LIT>' ] ) </s>
