<s> import mock <EOL> import munch <EOL> import glanceclient <EOL> from heatclient import client as heat_client <EOL> from neutronclient . common import exceptions as n_exc <EOL> import testtools <EOL> from os_client_config import cloud_config <EOL> import shade <EOL> from shade import _utils <EOL> from shade import exc <EOL> from shade . tests import fakes <EOL> from shade . tests . unit import base <EOL> RANGE_DATA = [ <EOL> dict ( id = <NUM_LIT:1> , key1 = <NUM_LIT:1> , key2 = <NUM_LIT:5> ) , <EOL> dict ( id = <NUM_LIT:2> , key1 = <NUM_LIT:1> , key2 = <NUM_LIT:20> ) , <EOL> dict ( id = <NUM_LIT:3> , key1 = <NUM_LIT:2> , key2 = <NUM_LIT:10> ) , <EOL> dict ( id = <NUM_LIT:4> , key1 = <NUM_LIT:2> , key2 = <NUM_LIT:30> ) , <EOL> dict ( id = <NUM_LIT:5> , key1 = <NUM_LIT:3> , key2 = <NUM_LIT> ) , <EOL> dict ( id = <NUM_LIT:6> , key1 = <NUM_LIT:3> , key2 = <NUM_LIT> ) , <EOL> ] <EOL> class TestShade ( base . TestCase ) : <EOL> def test_openstack_cloud ( self ) : <EOL> self . assertIsInstance ( self . cloud , shade . OpenStackCloud ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_images ( self , mock_search ) : <EOL> image1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ image1 ] <EOL> r = self . cloud . get_image ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( r ) <EOL> self . assertDictEqual ( image1 , r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_image_not_found ( self , mock_search ) : <EOL> mock_search . return_value = [ ] <EOL> r = self . cloud . get_image ( '<STR_LIT>' ) <EOL> self . assertIsNone ( r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_server ( self , mock_search ) : <EOL> server1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ server1 ] <EOL> r = self . cloud . get_server ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( r ) <EOL> self . assertDictEqual ( server1 , r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_server_not_found ( self , mock_search ) : <EOL> mock_search . return_value = [ ] <EOL> r = self . cloud . get_server ( '<STR_LIT>' ) <EOL> self . assertIsNone ( r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_servers_exception ( self , mock_client ) : <EOL> mock_client . servers . list . side_effect = Exception ( ) <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . list_servers ) <EOL> @ mock . patch . object ( cloud_config . CloudConfig , '<STR_LIT>' ) <EOL> @ mock . patch . object ( cloud_config . CloudConfig , '<STR_LIT>' ) <EOL> def test_glance_args ( self , get_legacy_client_mock , get_session_mock ) : <EOL> session_mock = mock . Mock ( ) <EOL> session_mock . get_endpoint . return_value = '<STR_LIT>' <EOL> get_session_mock . return_value = session_mock <EOL> self . cloud . glance_client <EOL> get_legacy_client_mock . assert_called_once_with ( <EOL> service_key = '<STR_LIT:image>' , <EOL> client_class = glanceclient . Client , <EOL> interface_key = None , <EOL> pass_version_arg = True , <EOL> ) <EOL> @ mock . patch . object ( cloud_config . CloudConfig , '<STR_LIT>' ) <EOL> @ mock . patch . object ( cloud_config . CloudConfig , '<STR_LIT>' ) <EOL> def test_heat_args ( self , get_legacy_client_mock , get_session_mock ) : <EOL> session_mock = mock . Mock ( ) <EOL> get_session_mock . return_value = session_mock <EOL> self . cloud . heat_client <EOL> get_legacy_client_mock . assert_called_once_with ( <EOL> service_key = '<STR_LIT>' , <EOL> client_class = heat_client . Client , <EOL> interface_key = None , <EOL> pass_version_arg = True , <EOL> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_networks ( self , mock_neutron ) : <EOL> net1 = { '<STR_LIT:id>' : '<STR_LIT:1>' , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> net2 = { '<STR_LIT:id>' : '<STR_LIT:2>' , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> mock_neutron . list_networks . return_value = { <EOL> '<STR_LIT>' : [ net1 , net2 ] <EOL> } <EOL> nets = self . cloud . list_networks ( ) <EOL> mock_neutron . list_networks . assert_called_once_with ( ) <EOL> self . assertEqual ( [ net1 , net2 ] , nets ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_networks_filtered ( self , mock_neutron ) : <EOL> self . cloud . list_networks ( filters = { '<STR_LIT:name>' : '<STR_LIT:test>' } ) <EOL> mock_neutron . list_networks . assert_called_once_with ( name = '<STR_LIT:test>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_networks_exception ( self , mock_neutron ) : <EOL> mock_neutron . list_networks . side_effect = Exception ( ) <EOL> with testtools . ExpectedException ( <EOL> exc . OpenStackCloudException , <EOL> "<STR_LIT>" <EOL> ) : <EOL> self . cloud . list_networks ( ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_subnet ( self , mock_search ) : <EOL> subnet = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ subnet ] <EOL> r = self . cloud . get_subnet ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( r ) <EOL> self . assertDictEqual ( subnet , r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_router ( self , mock_search ) : <EOL> router1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ router1 ] <EOL> r = self . cloud . get_router ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( r ) <EOL> self . assertDictEqual ( router1 , r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_router_not_found ( self , mock_search ) : <EOL> mock_search . return_value = [ ] <EOL> r = self . cloud . get_router ( '<STR_LIT>' ) <EOL> self . assertIsNone ( r ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_router ( self , mock_client ) : <EOL> self . cloud . create_router ( name = '<STR_LIT>' , admin_state_up = True ) <EOL> self . assertTrue ( mock_client . create_router . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_router_with_enable_snat_True ( self , mock_client ) : <EOL> """<STR_LIT>""" <EOL> self . cloud . create_router ( name = '<STR_LIT>' , admin_state_up = True , <EOL> enable_snat = True ) <EOL> mock_client . create_router . assert_called_once_with ( <EOL> body = dict ( <EOL> router = dict ( <EOL> name = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> ) <EOL> ) <EOL> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_router_with_enable_snat_False ( self , mock_client ) : <EOL> """<STR_LIT>""" <EOL> self . cloud . create_router ( name = '<STR_LIT>' , admin_state_up = True , <EOL> enable_snat = False ) <EOL> mock_client . create_router . assert_called_once_with ( <EOL> body = dict ( <EOL> router = dict ( <EOL> name = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> external_gateway_info = dict ( <EOL> enable_snat = False <EOL> ) <EOL> ) <EOL> ) <EOL> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_add_router_interface ( self , mock_client ) : <EOL> self . cloud . add_router_interface ( { '<STR_LIT:id>' : '<STR_LIT>' } , subnet_id = '<STR_LIT:abc>' ) <EOL> mock_client . add_interface_router . assert_called_once_with ( <EOL> router = '<STR_LIT>' , body = { '<STR_LIT>' : '<STR_LIT:abc>' } <EOL> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_remove_router_interface ( self , mock_client ) : <EOL> self . cloud . remove_router_interface ( { '<STR_LIT:id>' : '<STR_LIT>' } , subnet_id = '<STR_LIT:abc>' ) <EOL> mock_client . remove_interface_router . assert_called_once_with ( <EOL> router = '<STR_LIT>' , body = { '<STR_LIT>' : '<STR_LIT:abc>' } <EOL> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_update_router ( self , mock_client , mock_get ) : <EOL> router1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_get . return_value = router1 <EOL> self . cloud . update_router ( '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> self . assertTrue ( mock_client . update_router . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_router ( self , mock_client , mock_search ) : <EOL> router1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ router1 ] <EOL> self . cloud . delete_router ( '<STR_LIT>' ) <EOL> self . assertTrue ( mock_client . delete_router . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_router_not_found ( self , mock_client , mock_search ) : <EOL> mock_search . return_value = [ ] <EOL> r = self . cloud . delete_router ( '<STR_LIT>' ) <EOL> self . assertFalse ( r ) <EOL> self . assertFalse ( mock_client . delete_router . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_router_multiple_found ( self , mock_client ) : <EOL> router1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> router2 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_client . list_routers . return_value = dict ( routers = [ router1 , <EOL> router2 ] ) <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . delete_router , <EOL> '<STR_LIT>' ) <EOL> self . assertFalse ( mock_client . delete_router . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_router_multiple_using_id ( self , mock_client ) : <EOL> router1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> router2 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_client . list_routers . return_value = dict ( routers = [ router1 , <EOL> router2 ] ) <EOL> self . cloud . delete_router ( '<STR_LIT>' ) <EOL> self . assertTrue ( mock_client . delete_router . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_router_interfaces_no_gw ( self , mock_client , mock_search ) : <EOL> """<STR_LIT>""" <EOL> external_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> port_list = [ external_port ] <EOL> router = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> } <EOL> mock_search . return_value = port_list <EOL> ret = self . cloud . list_router_interfaces ( router , <EOL> interface_type = '<STR_LIT>' ) <EOL> mock_search . assert_called_once_with ( <EOL> filters = { '<STR_LIT>' : router [ '<STR_LIT:id>' ] } <EOL> ) <EOL> self . assertEqual ( [ ] , ret ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_router_interfaces_all ( self , mock_client , mock_search ) : <EOL> internal_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> external_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> port_list = [ internal_port , external_port ] <EOL> router = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> } <EOL> } <EOL> mock_search . return_value = port_list <EOL> ret = self . cloud . list_router_interfaces ( router ) <EOL> mock_search . assert_called_once_with ( <EOL> filters = { '<STR_LIT>' : router [ '<STR_LIT:id>' ] } <EOL> ) <EOL> self . assertEqual ( port_list , ret ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_router_interfaces_internal ( self , mock_client , mock_search ) : <EOL> internal_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> external_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> port_list = [ internal_port , external_port ] <EOL> router = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> } <EOL> } <EOL> mock_search . return_value = port_list <EOL> ret = self . cloud . list_router_interfaces ( router , <EOL> interface_type = '<STR_LIT>' ) <EOL> mock_search . assert_called_once_with ( <EOL> filters = { '<STR_LIT>' : router [ '<STR_LIT:id>' ] } <EOL> ) <EOL> self . assertEqual ( [ internal_port ] , ret ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_list_router_interfaces_external ( self , mock_client , mock_search ) : <EOL> internal_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> external_port = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] } <EOL> port_list = [ internal_port , external_port ] <EOL> router = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> } <EOL> } <EOL> mock_search . return_value = port_list <EOL> ret = self . cloud . list_router_interfaces ( router , <EOL> interface_type = '<STR_LIT>' ) <EOL> mock_search . assert_called_once_with ( <EOL> filters = { '<STR_LIT>' : router [ '<STR_LIT:id>' ] } <EOL> ) <EOL> self . assertEqual ( [ external_port ] , ret ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_subnet ( self , mock_client , mock_search ) : <EOL> net1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ net1 ] <EOL> pool = [ { '<STR_LIT:start>' : '<STR_LIT>' , '<STR_LIT:end>' : '<STR_LIT>' } ] <EOL> dns = [ '<STR_LIT>' ] <EOL> routes = [ { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ] <EOL> self . cloud . create_subnet ( '<STR_LIT>' , '<STR_LIT>' , <EOL> allocation_pools = pool , <EOL> dns_nameservers = dns , <EOL> host_routes = routes ) <EOL> self . assertTrue ( mock_client . create_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_subnet_without_gateway_ip ( self , mock_client , mock_search ) : <EOL> net1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ net1 ] <EOL> pool = [ { '<STR_LIT:start>' : '<STR_LIT>' , '<STR_LIT:end>' : '<STR_LIT>' } ] <EOL> dns = [ '<STR_LIT>' ] <EOL> self . cloud . create_subnet ( '<STR_LIT>' , '<STR_LIT>' , <EOL> allocation_pools = pool , <EOL> dns_nameservers = dns , <EOL> disable_gateway_ip = True ) <EOL> self . assertTrue ( mock_client . create_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_subnet_with_gateway_ip ( self , mock_client , mock_search ) : <EOL> net1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ net1 ] <EOL> pool = [ { '<STR_LIT:start>' : '<STR_LIT>' , '<STR_LIT:end>' : '<STR_LIT>' } ] <EOL> dns = [ '<STR_LIT>' ] <EOL> gateway = '<STR_LIT>' <EOL> self . cloud . create_subnet ( '<STR_LIT>' , '<STR_LIT>' , <EOL> allocation_pools = pool , <EOL> dns_nameservers = dns , <EOL> gateway_ip = gateway ) <EOL> self . assertTrue ( mock_client . create_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_subnet_conflict_gw_ops ( self , mock_client , mock_search ) : <EOL> net1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ net1 ] <EOL> gateway = '<STR_LIT>' <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . create_subnet , '<STR_LIT>' , <EOL> '<STR_LIT>' , gateway_ip = gateway , <EOL> disable_gateway_ip = True ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_subnet_bad_network ( self , mock_client , mock_list ) : <EOL> net1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_list . return_value = [ net1 ] <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . create_subnet , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertFalse ( mock_client . create_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_create_subnet_non_unique_network ( self , mock_client , mock_search ) : <EOL> net1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> net2 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ net1 , net2 ] <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . create_subnet , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertFalse ( mock_client . create_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_subnet ( self , mock_client , mock_search ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_search . return_value = [ subnet1 ] <EOL> self . cloud . delete_subnet ( '<STR_LIT>' ) <EOL> self . assertTrue ( mock_client . delete_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_subnet_not_found ( self , mock_client , mock_search ) : <EOL> mock_search . return_value = [ ] <EOL> r = self . cloud . delete_subnet ( '<STR_LIT>' ) <EOL> self . assertFalse ( r ) <EOL> self . assertFalse ( mock_client . delete_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_subnet_multiple_found ( self , mock_client ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> subnet2 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_client . list_subnets . return_value = dict ( subnets = [ subnet1 , <EOL> subnet2 ] ) <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . delete_subnet , <EOL> '<STR_LIT>' ) <EOL> self . assertFalse ( mock_client . delete_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_delete_subnet_multiple_using_id ( self , mock_client ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> subnet2 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_client . list_subnets . return_value = dict ( subnets = [ subnet1 , <EOL> subnet2 ] ) <EOL> self . cloud . delete_subnet ( '<STR_LIT>' ) <EOL> self . assertTrue ( mock_client . delete_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_update_subnet ( self , mock_client , mock_get ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_get . return_value = subnet1 <EOL> self . cloud . update_subnet ( '<STR_LIT>' , subnet_name = '<STR_LIT>' ) <EOL> self . assertTrue ( mock_client . update_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_update_subnet_gateway_ip ( self , mock_client , mock_get ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_get . return_value = subnet1 <EOL> gateway = '<STR_LIT>' <EOL> self . cloud . update_subnet ( <EOL> '<STR_LIT>' , gateway_ip = gateway ) <EOL> self . assertTrue ( mock_client . update_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_update_subnet_disable_gateway_ip ( self , mock_client , mock_get ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_get . return_value = subnet1 <EOL> self . cloud . update_subnet ( <EOL> '<STR_LIT>' , disable_gateway_ip = True ) <EOL> self . assertTrue ( mock_client . update_subnet . called ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_update_subnet_conflict_gw_ops ( self , mock_client , mock_get ) : <EOL> subnet1 = dict ( id = '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> mock_get . return_value = subnet1 <EOL> gateway = '<STR_LIT>' <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . update_subnet , <EOL> '<STR_LIT>' , gateway_ip = gateway , disable_gateway_ip = True ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_flavor_by_ram ( self , mock_nova_client , mock_compute ) : <EOL> vanilla = fakes . FakeFlavor ( '<STR_LIT:1>' , '<STR_LIT>' , <NUM_LIT:100> ) <EOL> chocolate = fakes . FakeFlavor ( '<STR_LIT:1>' , '<STR_LIT>' , <NUM_LIT:200> ) <EOL> mock_nova_client . flavors . list . return_value = [ vanilla , chocolate ] <EOL> mock_response = mock . Mock ( ) <EOL> mock_response . json . return_value = dict ( extra_specs = [ ] ) <EOL> mock_compute . get . return_value = mock_response <EOL> flavor = self . cloud . get_flavor_by_ram ( ram = <NUM_LIT> ) <EOL> self . assertEquals ( chocolate . id , flavor [ '<STR_LIT:id>' ] ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_flavor_by_ram_and_include ( <EOL> self , mock_nova_client , mock_compute ) : <EOL> vanilla = fakes . FakeFlavor ( '<STR_LIT:1>' , '<STR_LIT>' , <NUM_LIT:100> ) <EOL> chocolate = fakes . FakeFlavor ( '<STR_LIT:2>' , '<STR_LIT>' , <NUM_LIT:200> ) <EOL> strawberry = fakes . FakeFlavor ( '<STR_LIT:3>' , '<STR_LIT>' , <NUM_LIT> ) <EOL> mock_response = mock . Mock ( ) <EOL> mock_response . json . return_value = dict ( extra_specs = [ ] ) <EOL> mock_compute . get . return_value = mock_response <EOL> mock_nova_client . flavors . list . return_value = [ <EOL> vanilla , chocolate , strawberry ] <EOL> flavor = self . cloud . get_flavor_by_ram ( ram = <NUM_LIT> , include = '<STR_LIT>' ) <EOL> self . assertEquals ( strawberry . id , flavor [ '<STR_LIT:id>' ] ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_flavor_by_ram_not_found ( self , mock_nova_client ) : <EOL> mock_nova_client . flavors . list . return_value = [ ] <EOL> self . assertRaises ( shade . OpenStackCloudException , <EOL> self . cloud . get_flavor_by_ram , <EOL> ram = <NUM_LIT:100> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test_get_flavor_string_and_int ( <EOL> self , mock_nova_client , mock_compute ) : <EOL> vanilla = fakes . FakeFlavor ( '<STR_LIT:1>' , '<STR_LIT>' , <NUM_LIT:100> ) <EOL> mock_nova_client . flavors . list . return_value = [ vanilla ] <EOL> mock_response = mock . Mock ( ) <EOL> mock_response . json . return_value = dict ( extra_specs = [ ] ) <EOL> mock_compute . get . return_value = mock_response <EOL> flavor1 = self . cloud . get_flavor ( '<STR_LIT:1>' ) <EOL> self . assertEquals ( vanilla . id , flavor1 [ '<STR_LIT:id>' ] ) <EOL> flavor2 = self . cloud . get_flavor ( <NUM_LIT:1> ) <EOL> self . assertEquals ( vanilla . id , flavor2 [ '<STR_LIT:id>' ] ) <EOL> def test__neutron_exceptions_resource_not_found ( self ) : <EOL> with mock . patch . object ( <EOL> shade . _tasks , '<STR_LIT>' , <EOL> side_effect = n_exc . NotFound ( ) ) : <EOL> self . assertRaises ( exc . OpenStackCloudResourceNotFound , <EOL> self . cloud . list_networks ) <EOL> def test__neutron_exceptions_url_not_found ( self ) : <EOL> with mock . patch . object ( <EOL> shade . _tasks , '<STR_LIT>' , <EOL> side_effect = n_exc . NeutronClientException ( status_code = <NUM_LIT> ) ) : <EOL> self . assertRaises ( exc . OpenStackCloudURINotFound , <EOL> self . cloud . list_networks ) <EOL> def test__neutron_exceptions_neutron_client_generic ( self ) : <EOL> with mock . patch . object ( <EOL> shade . _tasks , '<STR_LIT>' , <EOL> side_effect = n_exc . NeutronClientException ( ) ) : <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . list_networks ) <EOL> def test__neutron_exceptions_generic ( self ) : <EOL> with mock . patch . object ( <EOL> shade . _tasks , '<STR_LIT>' , <EOL> side_effect = Exception ( ) ) : <EOL> self . assertRaises ( exc . OpenStackCloudException , <EOL> self . cloud . list_networks ) <EOL> @ mock . patch . object ( shade . _tasks . ServerList , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_list_servers ( self , mock_add_srv_int , mock_serverlist ) : <EOL> '''<STR_LIT>''' <EOL> server_obj = munch . Munch ( { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:image>' : '<STR_LIT>' } ) <EOL> mock_serverlist . return_value = [ server_obj ] <EOL> mock_add_srv_int . side_effect = [ server_obj ] <EOL> r = self . cloud . list_servers ( ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( r ) ) <EOL> self . assertEquals ( <NUM_LIT:1> , mock_add_srv_int . call_count ) <EOL> self . assertEquals ( '<STR_LIT>' , r [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] ) <EOL> @ mock . patch . object ( shade . _tasks . ServerList , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_list_servers_detailed ( self , <EOL> mock_get_hostvars_from_server , <EOL> mock_serverlist ) : <EOL> '''<STR_LIT>''' <EOL> mock_serverlist . return_value = [ <EOL> fakes . FakeServer ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> fakes . FakeServer ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> mock_get_hostvars_from_server . side_effect = [ <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:id>' : '<STR_LIT:1>' } , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:id>' : '<STR_LIT:2>' } , <EOL> ] <EOL> r = self . cloud . list_servers ( detailed = True ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( r ) ) <EOL> self . assertEquals ( len ( r ) , mock_get_hostvars_from_server . call_count ) <EOL> self . assertEquals ( '<STR_LIT>' , r [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] ) <EOL> self . assertEquals ( '<STR_LIT>' , r [ <NUM_LIT:1> ] [ '<STR_LIT:name>' ] ) <EOL> def test_iterate_timeout_bad_wait ( self ) : <EOL> with testtools . ExpectedException ( <EOL> exc . OpenStackCloudException , <EOL> "<STR_LIT>" ) : <EOL> for count in _utils . _iterate_timeout ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , wait = "<STR_LIT>" ) : <EOL> pass <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_iterate_timeout_str_wait ( self , mock_sleep ) : <EOL> iter = _utils . _iterate_timeout ( <EOL> <NUM_LIT:10> , "<STR_LIT>" , wait = "<STR_LIT>" ) <EOL> next ( iter ) <EOL> next ( iter ) <EOL> mock_sleep . assert_called_with ( <NUM_LIT> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_iterate_timeout_int_wait ( self , mock_sleep ) : <EOL> iter = _utils . _iterate_timeout ( <EOL> <NUM_LIT:10> , "<STR_LIT>" , wait = <NUM_LIT:1> ) <EOL> next ( iter ) <EOL> next ( iter ) <EOL> mock_sleep . assert_called_with ( <NUM_LIT:1.0> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_iterate_timeout_timeout ( self , mock_sleep ) : <EOL> message = "<STR_LIT>" <EOL> with testtools . ExpectedException ( <EOL> exc . OpenStackCloudTimeout , <EOL> message ) : <EOL> for count in _utils . _iterate_timeout ( <NUM_LIT:0.1> , message , wait = <NUM_LIT:1> ) : <EOL> pass <EOL> mock_sleep . assert_called_with ( <NUM_LIT:1.0> ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test__nova_extensions ( self , mock_compute ) : <EOL> body = { <EOL> '<STR_LIT>' : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> ] <EOL> } <EOL> mock_response = mock . Mock ( ) <EOL> mock_response . json . return_value = body <EOL> mock_compute . get . return_value = mock_response <EOL> extensions = self . cloud . _nova_extensions ( ) <EOL> mock_compute . get . assert_called_once_with ( '<STR_LIT>' ) <EOL> self . assertEqual ( set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , extensions ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test__nova_extensions_fails ( self , mock_compute ) : <EOL> mock_compute . get . side_effect = Exception ( ) <EOL> with testtools . ExpectedException ( <EOL> exc . OpenStackCloudException , <EOL> "<STR_LIT>" <EOL> ) : <EOL> self . cloud . _nova_extensions ( ) <EOL> @ mock . patch . object ( shade . OpenStackCloud , '<STR_LIT>' ) <EOL> def test__has_nova_extension ( self , mock_compute ) : <EOL> body = { <EOL> '<STR_LIT>' : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> ] <EOL> } <EOL> mock_response = mock . Mock ( ) <EOL> mock_response . json . return_value = body <EOL> mock_compute . get . return_value = mock_response <EOL> self . assertTrue ( self . cloud . _has_nova_extension ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( self . cloud . _has_nova_extension ( '<STR_LIT>' ) ) <EOL> def test_range_search ( self ) : <EOL> filters = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> retval = self . cloud . range_search ( RANGE_DATA , filters ) <EOL> self . assertIsInstance ( retval , list ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( retval ) ) <EOL> self . assertEqual ( [ RANGE_DATA [ <NUM_LIT:1> ] ] , retval ) <EOL> def test_range_search_2 ( self ) : <EOL> filters = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> retval = self . cloud . range_search ( RANGE_DATA , filters ) <EOL> self . assertIsInstance ( retval , list ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( retval ) ) <EOL> self . assertEqual ( [ RANGE_DATA [ <NUM_LIT:1> ] , RANGE_DATA [ <NUM_LIT:3> ] ] , retval ) <EOL> def test_range_search_3 ( self ) : <EOL> filters = { "<STR_LIT>" : "<STR_LIT:2>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> retval = self . cloud . range_search ( RANGE_DATA , filters ) <EOL> self . assertIsInstance ( retval , list ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( retval ) ) <EOL> def test_range_search_4 ( self ) : <EOL> filters = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> retval = self . cloud . range_search ( RANGE_DATA , filters ) <EOL> self . assertIsInstance ( retval , list ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( retval ) ) <EOL> def test_range_search_5 ( self ) : <EOL> filters = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> retval = self . cloud . range_search ( RANGE_DATA , filters ) <EOL> self . assertIsInstance ( retval , list ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( retval ) ) <EOL> self . assertEqual ( [ RANGE_DATA [ <NUM_LIT:0> ] ] , retval ) </s>
