<s> import ddt <EOL> import mock <EOL> from rally import exceptions <EOL> from rally . plugins . openstack . scenarios . neutron import utils <EOL> from tests . unit import test <EOL> NEUTRON_UTILS = "<STR_LIT>" <EOL> @ ddt . ddt <EOL> class NeutronScenarioTestCase ( test . ScenarioTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( NeutronScenarioTestCase , self ) . setUp ( ) <EOL> self . network = mock . Mock ( ) <EOL> self . scenario = utils . NeutronScenario ( self . context ) <EOL> self . random_name = "<STR_LIT>" <EOL> self . scenario . generate_random_name = mock . Mock ( <EOL> return_value = self . random_name ) <EOL> def test__get_network_id ( self ) : <EOL> networks = [ { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT:name>" : "<STR_LIT>" } , <EOL> { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT:name>" : "<STR_LIT>" } ] <EOL> network_id = "<STR_LIT>" <EOL> network = "<STR_LIT>" <EOL> self . scenario . _list_networks = mock . Mock ( return_value = networks ) <EOL> resultant_network_id = self . scenario . _get_network_id ( network ) <EOL> self . assertEqual ( network_id , resultant_network_id ) <EOL> self . scenario . _list_networks . assert_called_once_with ( <EOL> atomic_action = False ) <EOL> self . scenario . _list_networks . reset_mock ( ) <EOL> network = "<STR_LIT>" <EOL> resultant_network_id = self . scenario . _get_network_id ( network ) <EOL> self . assertEqual ( network_id , resultant_network_id ) <EOL> self . scenario . _list_networks . assert_called_once_with ( <EOL> atomic_action = False ) <EOL> self . scenario . _list_networks . reset_mock ( ) <EOL> network = "<STR_LIT>" <EOL> self . assertRaises ( exceptions . NotFoundException , <EOL> self . scenario . _get_network_id , network ) <EOL> self . scenario . _list_networks . assert_called_once_with ( <EOL> atomic_action = False ) <EOL> def test_create_network ( self ) : <EOL> self . clients ( "<STR_LIT>" ) . create_network . return_value = self . network <EOL> network_data = { "<STR_LIT>" : False } <EOL> expected_network_data = { "<STR_LIT>" : network_data } <EOL> network = self . scenario . _create_network ( network_data ) <EOL> self . assertEqual ( self . network , network ) <EOL> self . clients ( "<STR_LIT>" ) . create_network . assert_called_once_with ( <EOL> expected_network_data ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_list_networks ( self ) : <EOL> networks_list = [ ] <EOL> networks_dict = { "<STR_LIT>" : networks_list } <EOL> self . clients ( "<STR_LIT>" ) . list_networks . return_value = networks_dict <EOL> return_networks_list = self . scenario . _list_networks ( <EOL> atomic_action = False ) <EOL> self . assertEqual ( networks_list , return_networks_list ) <EOL> return_networks_list = self . scenario . _list_networks ( ) <EOL> self . assertEqual ( networks_list , return_networks_list ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_network ( self ) : <EOL> expected_network = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_network . return_value = expected_network <EOL> network = { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> network_update_args = { "<STR_LIT:name>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> result_network = self . scenario . _update_network ( network , <EOL> network_update_args ) <EOL> self . clients ( "<STR_LIT>" ) . update_network . assert_called_once_with ( <EOL> network [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , expected_network ) <EOL> self . assertEqual ( result_network , expected_network ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_network ( self ) : <EOL> network_create_args = { } <EOL> network = self . scenario . _create_network ( network_create_args ) <EOL> self . scenario . _delete_network ( network ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_create_subnet ( self ) : <EOL> network_id = "<STR_LIT>" <EOL> start_cidr = "<STR_LIT>" <EOL> network = { "<STR_LIT>" : { "<STR_LIT:id>" : network_id } } <EOL> expected_subnet_data = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : network_id , <EOL> "<STR_LIT>" : start_cidr , <EOL> "<STR_LIT>" : self . scenario . SUBNET_IP_VERSION , <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value <EOL> } <EOL> } <EOL> subnet_data = { "<STR_LIT>" : network_id } <EOL> self . scenario . _create_subnet ( network , subnet_data , start_cidr ) <EOL> self . clients ( "<STR_LIT>" ) . create_subnet . assert_called_once_with ( <EOL> expected_subnet_data ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> self . clients ( "<STR_LIT>" ) . create_subnet . reset_mock ( ) <EOL> extras = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : [ ] } <EOL> subnet_data . update ( extras ) <EOL> expected_subnet_data [ "<STR_LIT>" ] . update ( extras ) <EOL> self . scenario . _create_subnet ( network , subnet_data ) <EOL> self . clients ( "<STR_LIT>" ) . create_subnet . assert_called_once_with ( <EOL> expected_subnet_data ) <EOL> def test_list_subnets ( self ) : <EOL> subnets = [ { "<STR_LIT:name>" : "<STR_LIT>" } , { "<STR_LIT:name>" : "<STR_LIT>" } ] <EOL> self . clients ( "<STR_LIT>" ) . list_subnets . return_value = { <EOL> "<STR_LIT>" : subnets <EOL> } <EOL> result = self . scenario . _list_subnets ( ) <EOL> self . assertEqual ( subnets , result ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_subnet ( self ) : <EOL> expected_subnet = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_subnet . return_value = expected_subnet <EOL> subnet = { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> subnet_update_args = { "<STR_LIT:name>" : "<STR_LIT:foo>" , "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> result_subnet = self . scenario . _update_subnet ( subnet , <EOL> subnet_update_args ) <EOL> self . clients ( "<STR_LIT>" ) . update_subnet . assert_called_once_with ( <EOL> subnet [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , expected_subnet ) <EOL> self . assertEqual ( result_subnet , expected_subnet ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_subnet ( self ) : <EOL> network = self . scenario . _create_network ( { } ) <EOL> subnet = self . scenario . _create_subnet ( network , { } ) <EOL> self . scenario . _delete_subnet ( subnet ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_create_router ( self ) : <EOL> router = mock . Mock ( ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . return_value = router <EOL> result_router = self . scenario . _create_router ( { } ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . assert_called_once_with ( { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value <EOL> } <EOL> } ) <EOL> self . assertEqual ( result_router , router ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_create_router_with_ext_gw ( self ) : <EOL> router = mock . Mock ( ) <EOL> external_network = [ { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT>" : True } ] <EOL> self . scenario . _list_networks = mock . Mock ( return_value = external_network ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . return_value = router <EOL> gw_info = { "<STR_LIT>" : external_network [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] , <EOL> "<STR_LIT>" : True } <EOL> router_data = { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT>" : gw_info <EOL> } <EOL> result_router = self . scenario . _create_router ( { } , external_gw = True ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . assert_called_once_with ( <EOL> { "<STR_LIT>" : router_data } ) <EOL> self . assertEqual ( result_router , router ) <EOL> self . _test_atomic_action_timer ( <EOL> self . scenario . atomic_actions ( ) , "<STR_LIT>" ) <EOL> def test_create_router_with_ext_gw_but_no_ext_net ( self ) : <EOL> router = mock . Mock ( ) <EOL> external_network = [ { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT>" : False } ] <EOL> self . scenario . _list_networks = mock . Mock ( return_value = external_network ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . return_value = router <EOL> result_router = self . scenario . _create_router ( { } , external_gw = True ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . assert_called_once_with ( { <EOL> "<STR_LIT>" : { "<STR_LIT:name>" : self . scenario . generate_random_name . return_value } <EOL> } ) <EOL> self . assertEqual ( result_router , router ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_create_router_explicit ( self ) : <EOL> router = mock . Mock ( ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . return_value = router <EOL> router_data = { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : True } <EOL> result_router = self . scenario . _create_router ( router_data ) <EOL> self . clients ( "<STR_LIT>" ) . create_router . assert_called_once_with ( <EOL> { "<STR_LIT>" : router_data } ) <EOL> self . assertEqual ( result_router , router ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_list_routers ( self ) : <EOL> routers = [ mock . Mock ( ) ] <EOL> self . clients ( "<STR_LIT>" ) . list_routers . return_value = { <EOL> "<STR_LIT>" : routers } <EOL> self . assertEqual ( routers , self . scenario . _list_routers ( ) ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_router ( self ) : <EOL> expected_router = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_router . return_value = expected_router <EOL> router = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True <EOL> } <EOL> } <EOL> router_update_args = { "<STR_LIT:name>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> result_router = self . scenario . _update_router ( router , <EOL> router_update_args ) <EOL> self . clients ( "<STR_LIT>" ) . update_router . assert_called_once_with ( <EOL> router [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , expected_router ) <EOL> self . assertEqual ( result_router , expected_router ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_router ( self ) : <EOL> router = self . scenario . _create_router ( { } ) <EOL> self . scenario . _delete_router ( router ) <EOL> self . clients ( "<STR_LIT>" ) . delete_router . assert_called_once_with ( <EOL> router [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_remove_interface_router ( self ) : <EOL> subnet = { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } <EOL> router_data = { "<STR_LIT:id>" : <NUM_LIT:1> } <EOL> router = self . scenario . _create_router ( router_data ) <EOL> self . scenario . _add_interface_router ( subnet , router ) <EOL> self . scenario . _remove_interface_router ( subnet , router ) <EOL> mock_remove_router = self . clients ( "<STR_LIT>" ) . remove_interface_router <EOL> mock_remove_router . assert_called_once_with ( <EOL> router [ "<STR_LIT:id>" ] , { "<STR_LIT>" : subnet [ "<STR_LIT:id>" ] } ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_SUBNET_IP_VERSION ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( utils . NeutronScenario . SUBNET_IP_VERSION , <NUM_LIT:4> ) <EOL> def test_create_port ( self ) : <EOL> net_id = "<STR_LIT>" <EOL> net = { "<STR_LIT>" : { "<STR_LIT:id>" : net_id } } <EOL> expected_port_args = { <EOL> "<STR_LIT:port>" : { <EOL> "<STR_LIT>" : net_id , <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value <EOL> } <EOL> } <EOL> port_create_args = { } <EOL> self . scenario . _create_port ( net , port_create_args ) <EOL> self . clients ( "<STR_LIT>" <EOL> ) . create_port . assert_called_once_with ( expected_port_args ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> self . clients ( "<STR_LIT>" ) . create_port . reset_mock ( ) <EOL> port_args = { "<STR_LIT>" : True } <EOL> expected_port_args [ "<STR_LIT:port>" ] . update ( port_args ) <EOL> self . scenario . _create_port ( net , port_args ) <EOL> self . clients ( "<STR_LIT>" <EOL> ) . create_port . assert_called_once_with ( expected_port_args ) <EOL> def test_list_ports ( self ) : <EOL> ports = [ { "<STR_LIT:name>" : "<STR_LIT>" } , { "<STR_LIT:name>" : "<STR_LIT>" } ] <EOL> self . clients ( "<STR_LIT>" ) . list_ports . return_value = { "<STR_LIT>" : ports } <EOL> self . assertEqual ( ports , self . scenario . _list_ports ( ) ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_port ( self ) : <EOL> expected_port = { <EOL> "<STR_LIT:port>" : { <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_port . return_value = expected_port <EOL> port = { <EOL> "<STR_LIT:port>" : { <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True <EOL> } <EOL> } <EOL> port_update_args = { <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> result_port = self . scenario . _update_port ( port , port_update_args ) <EOL> self . clients ( "<STR_LIT>" ) . update_port . assert_called_once_with ( <EOL> port [ "<STR_LIT:port>" ] [ "<STR_LIT:id>" ] , expected_port ) <EOL> self . assertEqual ( result_port , expected_port ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_port ( self ) : <EOL> network = self . scenario . _create_network ( { } ) <EOL> port = self . scenario . _create_port ( network , { } ) <EOL> self . scenario . _delete_port ( port ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> @ ddt . data ( <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : <EOL> [ mock . MagicMock ( ) , mock . MagicMock ( ) ] } } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : <EOL> [ mock . MagicMock ( ) , mock . MagicMock ( ) ] } } } ) <EOL> @ ddt . unpack <EOL> @ mock . patch ( "<STR_LIT>" , side_effect = lambda l : l [ <NUM_LIT:0> ] ) <EOL> def test_get_or_create_network ( self , mock_random_choice , <EOL> network_create_args = None , context = None ) : <EOL> self . scenario . context = context <EOL> self . scenario . _create_network = mock . Mock ( <EOL> return_value = { "<STR_LIT>" : mock . Mock ( ) } ) <EOL> network = self . scenario . _get_or_create_network ( network_create_args ) <EOL> self . assertIn ( "<STR_LIT>" , network ) <EOL> if "<STR_LIT>" in context [ "<STR_LIT>" ] : <EOL> self . assertEqual ( network , <EOL> { "<STR_LIT>" : context [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ <NUM_LIT:0> ] } ) <EOL> self . assertFalse ( self . scenario . _create_network . called ) <EOL> else : <EOL> self . assertEqual ( network , <EOL> self . scenario . _create_network . return_value ) <EOL> self . scenario . _create_network . assert_called_once_with ( <EOL> network_create_args or { } ) <EOL> @ mock . patch ( NEUTRON_UTILS + "<STR_LIT>" ) <EOL> @ mock . patch ( NEUTRON_UTILS + "<STR_LIT>" ) <EOL> def test_create_network_and_subnets ( self , <EOL> mock__create_network , <EOL> mock__create_subnet ) : <EOL> mock__create_network . return_value = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> mock__create_subnet . return_value = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False <EOL> } <EOL> } <EOL> network_create_args = { } <EOL> subnet_create_args = { } <EOL> subnets_per_network = <NUM_LIT:4> <EOL> self . scenario . _create_network_and_subnets ( <EOL> network_create_args = network_create_args , <EOL> subnet_create_args = subnet_create_args , <EOL> subnets_per_network = subnets_per_network ) <EOL> mock__create_network . assert_called_once_with ( { } ) <EOL> mock__create_subnet . assert_has_calls ( <EOL> [ mock . call ( { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } , <EOL> { } , "<STR_LIT>" ) ] * subnets_per_network ) <EOL> mock__create_network . reset_mock ( ) <EOL> mock__create_subnet . reset_mock ( ) <EOL> self . scenario . _create_network_and_subnets ( <EOL> network_create_args = network_create_args , <EOL> subnet_create_args = { "<STR_LIT>" : [ ] } , <EOL> subnet_cidr_start = "<STR_LIT>" , <EOL> subnets_per_network = subnets_per_network ) <EOL> mock__create_network . assert_called_once_with ( { } ) <EOL> mock__create_subnet . assert_has_calls ( <EOL> [ mock . call ( { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } , <EOL> { "<STR_LIT>" : [ ] } , <EOL> "<STR_LIT>" ) ] * subnets_per_network ) <EOL> def test_list_floating_ips ( self ) : <EOL> fips_list = [ { "<STR_LIT:id>" : "<STR_LIT>" } ] <EOL> fips_dict = { "<STR_LIT>" : fips_list } <EOL> self . clients ( "<STR_LIT>" ) . list_floatingips . return_value = fips_dict <EOL> self . assertEqual ( self . scenario . _list_floating_ips ( ) , <EOL> self . clients ( "<STR_LIT>" ) . list_floatingips . return_value ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_floating_ip ( self ) : <EOL> fip = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> self . scenario . _delete_floating_ip ( fip [ "<STR_LIT>" ] ) <EOL> self . clients ( "<STR_LIT>" ) . delete_floatingip . assert_called_once_with ( <EOL> fip [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> @ ddt . data ( <EOL> { } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : False } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : True } , <EOL> "<STR_LIT>" : { "<STR_LIT>" : [ ] } , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:3> , <EOL> "<STR_LIT>" : { "<STR_LIT>" : False } } ) <EOL> @ ddt . unpack <EOL> def test_create_network_structure ( self , network_create_args = None , <EOL> subnet_create_args = None , <EOL> subnet_cidr_start = None , <EOL> subnets_per_network = None , <EOL> router_create_args = None ) : <EOL> network = mock . MagicMock ( ) <EOL> router_create_args = router_create_args or { } <EOL> subnets = [ ] <EOL> routers = [ ] <EOL> router_create_calls = [ ] <EOL> for i in range ( subnets_per_network or <NUM_LIT:1> ) : <EOL> subnets . append ( mock . MagicMock ( ) ) <EOL> routers . append ( mock . MagicMock ( ) ) <EOL> router_create_calls . append ( mock . call ( router_create_args ) ) <EOL> self . scenario . _get_or_create_network = mock . Mock ( return_value = network ) <EOL> self . scenario . _create_subnets = mock . Mock ( return_value = subnets ) <EOL> self . scenario . _create_router = mock . Mock ( side_effect = routers ) <EOL> self . scenario . _add_interface_router = mock . Mock ( ) <EOL> actual = self . scenario . _create_network_structure ( network_create_args , <EOL> subnet_create_args , <EOL> subnet_cidr_start , <EOL> subnets_per_network , <EOL> router_create_args ) <EOL> self . assertEqual ( actual , ( network , subnets , routers ) ) <EOL> self . scenario . _get_or_create_network . assert_called_once_with ( <EOL> network_create_args ) <EOL> self . scenario . _create_subnets . assert_called_once_with ( <EOL> network , <EOL> subnet_create_args , <EOL> subnet_cidr_start , <EOL> subnets_per_network ) <EOL> self . scenario . _create_router . assert_has_calls ( router_create_calls ) <EOL> add_iface_calls = [ mock . call ( subnets [ i ] [ "<STR_LIT>" ] , <EOL> routers [ i ] [ "<STR_LIT>" ] ) <EOL> for i in range ( subnets_per_network or <NUM_LIT:1> ) ] <EOL> self . scenario . _add_interface_router . assert_has_calls ( add_iface_calls ) <EOL> def test_delete_v1_pool ( self ) : <EOL> pool = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> self . scenario . _delete_v1_pool ( pool [ "<STR_LIT>" ] ) <EOL> self . clients ( "<STR_LIT>" ) . delete_pool . assert_called_once_with ( <EOL> pool [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_pool ( self ) : <EOL> expected_pool = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_pool . return_value = expected_pool <EOL> pool = { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> pool_update_args = { "<STR_LIT:name>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> result_pool = self . scenario . _update_v1_pool ( pool , ** pool_update_args ) <EOL> self . assertEqual ( result_pool , expected_pool ) <EOL> self . clients ( "<STR_LIT>" ) . update_pool . assert_called_once_with ( <EOL> pool [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , expected_pool ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_list_v1_pools ( self ) : <EOL> pools_list = [ ] <EOL> pools_dict = { "<STR_LIT>" : pools_list } <EOL> self . clients ( "<STR_LIT>" ) . list_pools . return_value = pools_dict <EOL> return_pools_dict = self . scenario . _list_v1_pools ( ) <EOL> self . assertEqual ( pools_dict , return_pools_dict ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_list_v1_vips ( self ) : <EOL> vips_list = [ ] <EOL> vips_dict = { "<STR_LIT>" : vips_list } <EOL> self . clients ( "<STR_LIT>" ) . list_vips . return_value = vips_dict <EOL> return_vips_dict = self . scenario . _list_v1_vips ( ) <EOL> self . assertEqual ( vips_dict , return_vips_dict ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_v1_vip ( self ) : <EOL> vip = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> self . scenario . _delete_v1_vip ( vip [ "<STR_LIT>" ] ) <EOL> self . clients ( "<STR_LIT>" ) . delete_vip . assert_called_once_with ( <EOL> vip [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_v1_vip ( self ) : <EOL> expected_vip = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT>" : False <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_vip . return_value = expected_vip <EOL> vip = { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> vip_update_args = { "<STR_LIT:name>" : "<STR_LIT:foo>" , "<STR_LIT>" : False } <EOL> result_vip = self . scenario . _update_v1_vip ( vip , ** vip_update_args ) <EOL> self . assertEqual ( result_vip , expected_vip ) <EOL> self . clients ( "<STR_LIT>" ) . update_vip . assert_called_once_with ( <EOL> vip [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , expected_vip ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> @ mock . patch ( NEUTRON_UTILS + "<STR_LIT>" ) <EOL> def test_create_security_group ( self , mock_generate_random_name ) : <EOL> security_group_create_args = { "<STR_LIT:description>" : "<STR_LIT>" } <EOL> expected_security_group = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . create_security_group = mock . Mock ( <EOL> return_value = expected_security_group ) <EOL> security_group_data = { <EOL> "<STR_LIT>" : <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" } <EOL> } <EOL> resultant_security_group = self . scenario . _create_security_group ( <EOL> ** security_group_create_args ) <EOL> self . assertEqual ( expected_security_group , resultant_security_group ) <EOL> self . clients ( "<STR_LIT>" ) . create_security_group . assert_called_once_with ( <EOL> security_group_data ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_list_security_groups ( self ) : <EOL> security_groups_list = [ { "<STR_LIT:id>" : "<STR_LIT>" } ] <EOL> security_groups_dict = { "<STR_LIT>" : security_groups_list } <EOL> self . clients ( "<STR_LIT>" ) . list_security_groups = mock . Mock ( <EOL> return_value = security_groups_dict ) <EOL> self . assertEqual ( <EOL> self . scenario . _list_security_groups ( ) , <EOL> self . clients ( "<STR_LIT>" ) . list_security_groups . return_value ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_security_group ( self ) : <EOL> security_group = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> self . scenario . _delete_security_group ( security_group ) <EOL> self . clients ( "<STR_LIT>" ) . delete_security_group . assert_called_once_with ( <EOL> security_group [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_security_group ( self ) : <EOL> security_group = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> expected_security_group = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . clients ( "<STR_LIT>" ) . update_security_group = mock . Mock ( <EOL> return_value = expected_security_group ) <EOL> result_security_group = self . scenario . _update_security_group ( <EOL> security_group , description = "<STR_LIT>" ) <EOL> self . clients ( "<STR_LIT>" ) . update_security_group . assert_called_once_with ( <EOL> security_group [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , <EOL> { "<STR_LIT>" : { <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : self . scenario . generate_random_name . return_value } } <EOL> ) <EOL> self . assertEqual ( result_security_group , expected_security_group ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> @ ddt . data ( <EOL> { "<STR_LIT>" : [ { "<STR_LIT>" : "<STR_LIT>" } ] } , <EOL> { "<STR_LIT>" : None , "<STR_LIT>" : [ { "<STR_LIT>" : [ "<STR_LIT>" ] } ] } , <EOL> { "<STR_LIT>" : { } , "<STR_LIT>" : [ { "<STR_LIT>" : [ "<STR_LIT>" ] } ] } , <EOL> { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : [ "<STR_LIT>" ] } ] } , <EOL> ) <EOL> @ ddt . unpack <EOL> def test__create_v1_pools ( self , networks , pool_create_args = None ) : <EOL> pool_create_args = pool_create_args or { } <EOL> pool = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> self . scenario . _create_lb_pool = mock . Mock ( return_value = pool ) <EOL> resultant_pools = self . scenario . _create_v1_pools ( <EOL> networks = networks , ** pool_create_args ) <EOL> if networks : <EOL> subnets = [ ] <EOL> [ subnets . extend ( net [ "<STR_LIT>" ] ) for net in networks ] <EOL> self . scenario . _create_lb_pool . assert_has_calls ( <EOL> [ mock . call ( subnet , atomic_action = False , <EOL> ** pool_create_args ) for subnet in subnets ] ) <EOL> self . assertEqual ( resultant_pools , [ pool ] * len ( subnets ) ) <EOL> self . _test_atomic_action_timer ( <EOL> self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" % len ( subnets ) ) <EOL> @ ddt . data ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : None , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : { } , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> { "<STR_LIT>" : None , "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False } , <EOL> { "<STR_LIT>" : { } , "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False } , <EOL> { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> ) <EOL> @ ddt . unpack <EOL> def test__create_lb_pool ( self , subnet_id = None , atomic_action = True , <EOL> pool_create_args = None ) : <EOL> pool = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> pool_create_args = pool_create_args or { } <EOL> if pool_create_args . get ( "<STR_LIT:name>" ) is None : <EOL> self . generate_random_name = mock . Mock ( return_value = "<STR_LIT>" ) <EOL> self . clients ( "<STR_LIT>" ) . create_pool . return_value = pool <EOL> args = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : subnet_id } <EOL> args . update ( pool_create_args ) <EOL> expected_pool_data = { "<STR_LIT>" : args } <EOL> resultant_pool = self . scenario . _create_lb_pool ( <EOL> subnet_id = subnet_id , atomic_action = atomic_action , <EOL> ** pool_create_args ) <EOL> self . assertEqual ( resultant_pool , pool ) <EOL> self . clients ( "<STR_LIT>" ) . create_pool . assert_called_once_with ( <EOL> expected_pool_data ) <EOL> if atomic_action : <EOL> self . _test_atomic_action_timer ( <EOL> self . scenario . atomic_actions ( ) , "<STR_LIT>" ) <EOL> @ ddt . data ( <EOL> { } , <EOL> { "<STR_LIT>" : { } } , <EOL> { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" } } , <EOL> ) <EOL> @ ddt . unpack <EOL> def test__create_v1_vip ( self , vip_create_args = None ) : <EOL> vip = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> pool = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } } <EOL> vip_create_args = vip_create_args or { } <EOL> if vip_create_args . get ( "<STR_LIT:name>" ) is None : <EOL> self . scenario . generate_random_name = mock . Mock ( <EOL> return_value = "<STR_LIT>" ) <EOL> self . clients ( "<STR_LIT>" ) . create_vip . return_value = vip <EOL> args = { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : pool [ "<STR_LIT>" ] [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : pool [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] } <EOL> args . update ( vip_create_args ) <EOL> expected_vip_data = { "<STR_LIT>" : args } <EOL> resultant_vip = self . scenario . _create_v1_vip ( pool , ** vip_create_args ) <EOL> self . assertEqual ( resultant_vip , vip ) <EOL> self . clients ( "<STR_LIT>" ) . create_vip . assert_called_once_with ( <EOL> expected_vip_data ) <EOL> @ ddt . data ( <EOL> { } , <EOL> { "<STR_LIT>" : { } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } , <EOL> ) <EOL> @ ddt . unpack <EOL> def test__create_floating_ip ( self , floating_ip_args = None ) : <EOL> floating_network = "<STR_LIT>" <EOL> fip = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> network_id = "<STR_LIT>" <EOL> floating_ip_args = floating_ip_args or { } <EOL> self . clients ( "<STR_LIT>" ) . create_floatingip . return_value = fip <EOL> mock_get_network_id = self . scenario . _get_network_id = mock . Mock ( ) <EOL> mock_get_network_id . return_value = network_id <EOL> args = { "<STR_LIT>" : network_id } <EOL> args . update ( floating_ip_args ) <EOL> expected_fip_data = { "<STR_LIT>" : args } <EOL> resultant_fip = self . scenario . _create_floatingip ( <EOL> floating_network , ** floating_ip_args ) <EOL> self . assertEqual ( resultant_fip , fip ) <EOL> self . clients ( "<STR_LIT>" ) . create_floatingip . assert_called_once_with ( <EOL> expected_fip_data ) <EOL> mock_get_network_id . assert_called_once_with ( floating_network ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> @ ddt . data ( <EOL> { } , <EOL> { "<STR_LIT>" : { } } , <EOL> { "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" } } , <EOL> { "<STR_LIT>" : False } , <EOL> { "<STR_LIT>" : False , <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" } } , <EOL> { "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : False } , <EOL> ) <EOL> @ ddt . unpack <EOL> def test__create_v1_healthmonitor ( self , atomic_action = True , <EOL> healthmonitor_create_args = None ) : <EOL> hm = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> healthmonitor_create_args = healthmonitor_create_args or { } <EOL> self . clients ( "<STR_LIT>" ) . create_health_monitor . return_value = hm <EOL> args = { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:20> , <EOL> "<STR_LIT>" : <NUM_LIT:10> , "<STR_LIT>" : <NUM_LIT:3> } <EOL> args . update ( healthmonitor_create_args ) <EOL> expected_hm_data = { "<STR_LIT>" : args } <EOL> resultant_hm = self . scenario . _create_v1_healthmonitor ( <EOL> atomic_action = atomic_action , <EOL> ** healthmonitor_create_args ) <EOL> self . assertEqual ( resultant_hm , hm ) <EOL> self . clients ( "<STR_LIT>" ) . create_health_monitor . assert_called_once_with ( <EOL> expected_hm_data ) <EOL> if atomic_action : <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_list_v1_healthmonitors ( self ) : <EOL> hm_list = [ ] <EOL> hm_dict = { "<STR_LIT>" : hm_list } <EOL> self . clients ( "<STR_LIT>" ) . list_health_monitors . return_value = hm_dict <EOL> return_hm_dict = self . scenario . _list_v1_healthmonitors ( ) <EOL> self . assertEqual ( hm_dict , return_hm_dict ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_delete_v1_healthmonitor ( self ) : <EOL> healthmonitor = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> self . scenario . _delete_v1_healthmonitor ( healthmonitor [ "<STR_LIT>" ] ) <EOL> self . clients ( "<STR_LIT>" ) . delete_health_monitor . assert_called_once_with ( <EOL> healthmonitor [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_update_healthmonitor ( self ) : <EOL> expected_hm = { "<STR_LIT>" : { "<STR_LIT>" : False } } <EOL> mock_update = self . clients ( "<STR_LIT>" ) . update_health_monitor <EOL> mock_update . return_value = expected_hm <EOL> hm = { "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } } <EOL> healthmonitor_update_args = { "<STR_LIT>" : False } <EOL> result_hm = self . scenario . _update_v1_healthmonitor ( <EOL> hm , ** healthmonitor_update_args ) <EOL> self . assertEqual ( result_hm , expected_hm ) <EOL> mock_update . assert_called_once_with ( <EOL> hm [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] , expected_hm ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> class NeutronScenarioFunctionalTestCase ( test . FakeClientsScenarioTestCase ) : <EOL> @ mock . patch ( NEUTRON_UTILS + "<STR_LIT>" ) <EOL> def test_functional_create_network_and_subnets ( self , mock_generate_cidr ) : <EOL> scenario = utils . NeutronScenario ( context = self . context ) <EOL> network_create_args = { } <EOL> subnet_create_args = { } <EOL> subnets_per_network = <NUM_LIT:5> <EOL> subnet_cidr_start = "<STR_LIT>" <EOL> cidrs = [ "<STR_LIT>" % i for i in range ( subnets_per_network ) ] <EOL> cidrs_ = iter ( cidrs ) <EOL> mock_generate_cidr . side_effect = lambda ** kw : next ( cidrs_ ) <EOL> network , subnets = scenario . _create_network_and_subnets ( <EOL> network_create_args , <EOL> subnet_create_args , <EOL> subnets_per_network , <EOL> subnet_cidr_start ) <EOL> result_cidrs = sorted ( [ s [ "<STR_LIT>" ] [ "<STR_LIT>" ] for s in subnets ] ) <EOL> self . assertEqual ( cidrs , result_cidrs ) </s>
