<s> import webob . exc <EOL> from neutron import context <EOL> from neutron . plugins . common import constants <EOL> from oslo_config import cfg <EOL> from oslo_utils import uuidutils <EOL> from gbpservice . neutron . db import servicechain_db as svcchain_db <EOL> from gbpservice . neutron . extensions import servicechain as service_chain <EOL> from gbpservice . neutron . tests . unit import common as cm <EOL> from gbpservice . neutron . tests . unit . db . grouppolicy import test_group_policy_db <EOL> JSON_FORMAT = '<STR_LIT>' <EOL> GP_PLUGIN_KLASS = ( <EOL> "<STR_LIT>" ) <EOL> class ServiceChainDBTestBase ( test_group_policy_db . GroupPolicyDBTestBase ) : <EOL> def _get_resource_plural ( self , resource ) : <EOL> if resource . endswith ( '<STR_LIT:y>' ) : <EOL> resource_plural = resource . replace ( '<STR_LIT:y>' , '<STR_LIT>' ) <EOL> else : <EOL> resource_plural = resource + '<STR_LIT:s>' <EOL> return resource_plural <EOL> def _test_list_resources ( self , resource , items , <EOL> neutron_context = None , <EOL> query_params = None ) : <EOL> resource_plural = self . _get_resource_plural ( resource ) <EOL> res = self . _list ( resource_plural , <EOL> neutron_context = neutron_context , <EOL> query_params = query_params ) <EOL> params = query_params . split ( '<STR_LIT:&>' ) <EOL> params = dict ( ( x . split ( '<STR_LIT:=>' ) [ <NUM_LIT:0> ] , x . split ( '<STR_LIT:=>' ) [ <NUM_LIT:1> ] . split ( '<STR_LIT:U+002C>' ) ) <EOL> for x in params ) <EOL> count = getattr ( self . plugin , '<STR_LIT>' % resource_plural ) ( <EOL> neutron_context or context . get_admin_context ( ) , params ) <EOL> self . assertEqual ( len ( res [ resource_plural ] ) , count ) <EOL> resource = resource . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) <EOL> self . assertEqual ( sorted ( [ i [ '<STR_LIT:id>' ] for i in res [ resource_plural ] ] ) , <EOL> sorted ( [ i [ resource ] [ '<STR_LIT:id>' ] for i in items ] ) ) <EOL> def _create_profiled_servicechain_node ( <EOL> self , service_type = constants . LOADBALANCER , shared_profile = False , <EOL> profile_tenant_id = None , ** kwargs ) : <EOL> prof = self . create_service_profile ( <EOL> service_type = service_type , <EOL> shared = shared_profile , <EOL> tenant_id = profile_tenant_id or self . _tenant_id ) [ '<STR_LIT>' ] <EOL> return self . create_servicechain_node ( <EOL> service_profile_id = prof [ '<STR_LIT:id>' ] , ** kwargs ) <EOL> class ServiceChainDBTestPlugin ( svcchain_db . ServiceChainDbPlugin ) : <EOL> supported_extension_aliases = [ '<STR_LIT>' ] + ( <EOL> test_group_policy_db . UNSUPPORTED_REQUIRED_EXTS ) <EOL> path_prefix = "<STR_LIT>" <EOL> DB_GP_PLUGIN_KLASS = ( ServiceChainDBTestPlugin . __module__ + '<STR_LIT:.>' + <EOL> ServiceChainDBTestPlugin . __name__ ) <EOL> class ServiceChainDbTestCase ( test_group_policy_db . GroupPolicyDbTestCase ) : <EOL> def setUp ( self , core_plugin = None , sc_plugin = None , service_plugins = None , <EOL> ext_mgr = None , gp_plugin = None ) : <EOL> super ( ServiceChainDbTestCase , self ) . setUp ( <EOL> gp_plugin = gp_plugin or GP_PLUGIN_KLASS , core_plugin = core_plugin , <EOL> sc_plugin = sc_plugin , service_plugins = service_plugins , <EOL> ext_mgr = ext_mgr ) <EOL> self . plugin = self . _sc_plugin <EOL> class TestServiceChainResources ( ServiceChainDbTestCase ) : <EOL> def _test_show_resource ( self , resource , resource_id , attrs ) : <EOL> resource_plural = self . _get_resource_plural ( resource ) <EOL> req = self . new_show_request ( resource_plural , resource_id , <EOL> fmt = self . fmt ) <EOL> res = self . deserialize ( self . fmt , <EOL> req . get_response ( self . ext_api ) ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , res [ resource ] [ k ] ) <EOL> def test_create_servicechain_specs_same_node ( self ) : <EOL> template1 = '<STR_LIT>' <EOL> sp = self . create_service_profile ( <EOL> service_type = constants . FIREWALL ) [ '<STR_LIT>' ] <EOL> scn = self . create_servicechain_node ( <EOL> config = template1 , service_profile_id = sp [ '<STR_LIT:id>' ] ) <EOL> scn_id = scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> spec1 = { "<STR_LIT>" : { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . _tenant_id , <EOL> '<STR_LIT>' : [ scn_id ] } } <EOL> spec_req = self . new_create_request ( '<STR_LIT>' , <EOL> spec1 , <EOL> self . fmt ) <EOL> spec_res = spec_req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPCreated . code , spec_res . status_int ) <EOL> res = self . deserialize ( self . fmt , spec_res ) <EOL> self . assertIn ( '<STR_LIT>' , res ) <EOL> self . assertEqual ( [ scn_id ] , res [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> spec2 = { "<STR_LIT>" : { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . _tenant_id , <EOL> '<STR_LIT>' : [ scn_id ] } } <EOL> spec_req = self . new_create_request ( '<STR_LIT>' , <EOL> spec2 , <EOL> self . fmt ) <EOL> spec_res = spec_req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPCreated . code , spec_res . status_int ) <EOL> res = self . deserialize ( self . fmt , spec_res ) <EOL> self . assertIn ( '<STR_LIT>' , res ) <EOL> self . assertEqual ( [ scn_id ] , res [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_create_and_show_servicechain_node ( self ) : <EOL> profile = self . create_service_profile ( service_type = constants . FIREWALL ) <EOL> attrs = cm . get_create_servicechain_node_default_attrs ( <EOL> service_profile_id = profile [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> config = "<STR_LIT>" ) <EOL> scn = self . create_servicechain_node ( <EOL> service_profile_id = profile [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> config = "<STR_LIT>" ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , scn [ '<STR_LIT>' ] [ k ] ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> attrs ) <EOL> def test_list_servicechain_nodes ( self ) : <EOL> scns = [ <EOL> self . _create_profiled_servicechain_node ( name = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) , <EOL> self . _create_profiled_servicechain_node ( name = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) , <EOL> self . _create_profiled_servicechain_node ( name = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) ] <EOL> self . _test_list_resources ( '<STR_LIT>' , scns , <EOL> query_params = '<STR_LIT>' ) <EOL> def test_update_servicechain_node ( self ) : <EOL> name = '<STR_LIT>' <EOL> description = '<STR_LIT>' <EOL> config = '<STR_LIT>' <EOL> profile = self . create_service_profile ( service_type = constants . FIREWALL ) <EOL> attrs = cm . get_create_servicechain_node_default_attrs ( <EOL> name = name , description = description , <EOL> config = config , <EOL> service_profile_id = profile [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> scn = self . create_servicechain_node ( <EOL> service_profile_id = profile [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> data = { '<STR_LIT>' : { '<STR_LIT:name>' : name , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : config } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , <EOL> scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . deserialize ( self . fmt , req . get_response ( self . ext_api ) ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , res [ '<STR_LIT>' ] [ k ] ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> attrs ) <EOL> def test_delete_servicechain_node ( self ) : <EOL> ctx = context . get_admin_context ( ) <EOL> scn = self . _create_profiled_servicechain_node ( ) <EOL> scn_id = scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> scs = self . create_servicechain_spec ( nodes = [ scn_id ] ) <EOL> scs_id = scs [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . assertRaises ( service_chain . ServiceChainNodeInUse , <EOL> self . plugin . delete_servicechain_node , ctx , scn_id ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , scs_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , scn_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceChainNodeNotFound , <EOL> self . plugin . get_servicechain_node , <EOL> ctx , scn_id ) <EOL> def test_create_and_show_servicechain_spec ( self ) : <EOL> name = "<STR_LIT>" <EOL> scn = self . _create_profiled_servicechain_node ( ) <EOL> scn_id = scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> attrs = cm . get_create_servicechain_spec_default_attrs ( <EOL> name = name , nodes = [ scn_id ] ) <EOL> scs = self . create_servicechain_spec ( name = name , nodes = [ scn_id ] ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , scs [ '<STR_LIT>' ] [ k ] ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> scs [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> attrs ) <EOL> def test_create_spec_multiple_nodes ( self ) : <EOL> name = "<STR_LIT>" <EOL> scn1 = self . _create_profiled_servicechain_node ( ) <EOL> scn1_id = scn1 [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> scn2 = self . _create_profiled_servicechain_node ( ) <EOL> scn2_id = scn2 [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> attrs = cm . get_create_servicechain_spec_default_attrs ( <EOL> name = name , nodes = [ scn1_id , scn2_id ] ) <EOL> scs = self . create_servicechain_spec ( <EOL> name = name , nodes = [ scn1_id , scn2_id ] ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , scs [ '<STR_LIT>' ] [ k ] ) <EOL> def test_list_servicechain_specs ( self ) : <EOL> scs = [ self . create_servicechain_spec ( name = '<STR_LIT>' , description = '<STR_LIT>' ) , <EOL> self . create_servicechain_spec ( name = '<STR_LIT>' , description = '<STR_LIT>' ) , <EOL> self . create_servicechain_spec ( name = '<STR_LIT>' , description = '<STR_LIT>' ) ] <EOL> self . _test_list_resources ( '<STR_LIT>' , scs , <EOL> query_params = '<STR_LIT>' ) <EOL> def test_node_ordering_list_servicechain_specs ( self ) : <EOL> scn1_id = self . _create_profiled_servicechain_node ( ) [ <EOL> '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> scn2_id = self . _create_profiled_servicechain_node ( ) [ <EOL> '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> nodes_list = [ scn1_id , scn2_id ] <EOL> scs = self . create_servicechain_spec ( name = '<STR_LIT>' , <EOL> nodes = nodes_list ) <EOL> self . assertEqual ( nodes_list , scs [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> res = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( res [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( nodes_list , res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , <EOL> scs [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> nodes_list . reverse ( ) <EOL> scs = self . create_servicechain_spec ( name = '<STR_LIT>' , <EOL> nodes = nodes_list ) <EOL> self . assertEqual ( scs [ '<STR_LIT>' ] [ '<STR_LIT>' ] , nodes_list ) <EOL> res = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( res [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( nodes_list , res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> def test_update_servicechain_spec ( self ) : <EOL> name = "<STR_LIT>" <EOL> description = '<STR_LIT>' <EOL> scn_id = self . _create_profiled_servicechain_node ( ) [ <EOL> '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> attrs = cm . get_create_servicechain_spec_default_attrs ( <EOL> name = name , description = description , nodes = [ scn_id ] ) <EOL> scs = self . create_servicechain_spec ( ) <EOL> data = { '<STR_LIT>' : { '<STR_LIT:name>' : name , '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : [ scn_id ] } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , <EOL> scs [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . deserialize ( self . fmt , req . get_response ( self . ext_api ) ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , res [ '<STR_LIT>' ] [ k ] ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> scs [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , attrs ) <EOL> def test_delete_servicechain_spec ( self ) : <EOL> ctx = context . get_admin_context ( ) <EOL> scs = self . create_servicechain_spec ( ) <EOL> scs_id = scs [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> req = self . new_delete_request ( '<STR_LIT>' , scs_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceChainSpecNotFound , <EOL> self . plugin . get_servicechain_spec , ctx , scs_id ) <EOL> def test_delete_spec_in_use_by_policy_action_rejected ( self ) : <EOL> ctx = context . get_admin_context ( ) <EOL> scs_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> data = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . _tenant_id , <EOL> '<STR_LIT>' : scs_id } } <EOL> pa_req = self . new_create_request ( '<STR_LIT>' , <EOL> data , self . fmt ) <EOL> res = pa_req . get_response ( self . ext_api ) <EOL> if res . status_int >= webob . exc . HTTPClientError . code : <EOL> raise webob . exc . HTTPClientError ( code = res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceChainSpecInUse , <EOL> self . plugin . delete_servicechain_spec , ctx , scs_id ) <EOL> def test_delete_spec_in_use_by_instance_rejected ( self ) : <EOL> ctx = context . get_admin_context ( ) <EOL> scs_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> sci = self . create_servicechain_instance ( servicechain_specs = [ scs_id ] ) <EOL> sci_id = sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . assertRaises ( service_chain . ServiceChainSpecInUse , <EOL> self . plugin . delete_servicechain_spec , ctx , scs_id ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , sci_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceChainInstanceNotFound , <EOL> self . plugin . get_servicechain_instance , <EOL> ctx , sci_id ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , scs_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceChainSpecNotFound , <EOL> self . plugin . get_servicechain_spec , ctx , scs_id ) <EOL> def test_create_and_show_servicechain_instance ( self ) : <EOL> scs_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> policy_target_group_id = uuidutils . generate_uuid ( ) <EOL> classifier_id = uuidutils . generate_uuid ( ) <EOL> config_param_values = "<STR_LIT:{}>" <EOL> attrs = cm . get_create_servicechain_instance_default_attrs ( <EOL> servicechain_specs = [ scs_id ] , <EOL> provider_ptg_id = policy_target_group_id , <EOL> consumer_ptg_id = policy_target_group_id , <EOL> management_ptg_id = policy_target_group_id , <EOL> classifier_id = classifier_id , <EOL> config_param_values = config_param_values ) <EOL> sci = self . create_servicechain_instance ( <EOL> servicechain_specs = [ scs_id ] , <EOL> provider_ptg_id = policy_target_group_id , <EOL> consumer_ptg_id = policy_target_group_id , <EOL> management_ptg_id = policy_target_group_id , <EOL> classifier_id = classifier_id , <EOL> config_param_values = config_param_values ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , sci [ '<STR_LIT>' ] [ k ] ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> attrs ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> req . get_response ( self . ext_api ) <EOL> def test_list_servicechain_instances ( self ) : <EOL> servicechain_instances = [ self . create_servicechain_instance ( <EOL> name = '<STR_LIT>' , description = '<STR_LIT>' ) , <EOL> self . create_servicechain_instance ( name = '<STR_LIT>' , description = '<STR_LIT>' ) , <EOL> self . create_servicechain_instance ( name = '<STR_LIT>' , description = '<STR_LIT>' ) ] <EOL> self . _test_list_resources ( '<STR_LIT>' , <EOL> servicechain_instances , <EOL> query_params = '<STR_LIT>' ) <EOL> def test_spec_ordering_list_servicechain_instances ( self ) : <EOL> scs1_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> scs2_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> specs_list = [ scs1_id , scs2_id ] <EOL> sci = self . create_servicechain_instance ( name = '<STR_LIT>' , <EOL> servicechain_specs = specs_list ) <EOL> self . assertEqual ( specs_list , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> res = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( res [ '<STR_LIT>' ] ) ) <EOL> result_instance = res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( specs_list , result_instance [ '<STR_LIT>' ] ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> specs_list . reverse ( ) <EOL> sci = self . create_servicechain_instance ( name = '<STR_LIT>' , <EOL> servicechain_specs = specs_list ) <EOL> self . assertEqual ( specs_list , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> res = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( res [ '<STR_LIT>' ] ) ) <EOL> result_instance = res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( specs_list , <EOL> result_instance [ '<STR_LIT>' ] ) <EOL> def test_update_servicechain_instance ( self ) : <EOL> name = "<STR_LIT>" <EOL> description = '<STR_LIT>' <EOL> config_param_values = "<STR_LIT:{}>" <EOL> scs_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> provider_ptg_id = uuidutils . generate_uuid ( ) <EOL> consumer_ptg_id = uuidutils . generate_uuid ( ) <EOL> management_ptg_id = uuidutils . generate_uuid ( ) <EOL> classifier_id = uuidutils . generate_uuid ( ) <EOL> attrs = cm . get_create_servicechain_instance_default_attrs ( <EOL> name = name , description = description , servicechain_specs = [ scs_id ] , <EOL> provider_ptg_id = provider_ptg_id , consumer_ptg_id = consumer_ptg_id , <EOL> management_ptg_id = management_ptg_id , <EOL> classifier_id = classifier_id , <EOL> config_param_values = config_param_values ) <EOL> sci = self . create_servicechain_instance ( <EOL> servicechain_specs = [ scs_id ] , provider_ptg_id = provider_ptg_id , <EOL> consumer_ptg_id = consumer_ptg_id , <EOL> management_ptg_id = management_ptg_id , classifier_id = classifier_id , <EOL> config_param_values = config_param_values ) <EOL> new_classifier_id = uuidutils . generate_uuid ( ) <EOL> new_scs_id = self . create_servicechain_spec ( ) [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> attrs . update ( { '<STR_LIT>' : [ new_scs_id ] , <EOL> '<STR_LIT>' : new_classifier_id } ) <EOL> data = { '<STR_LIT>' : { '<STR_LIT:name>' : name , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : [ new_scs_id ] , <EOL> '<STR_LIT>' : new_classifier_id } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . deserialize ( self . fmt , req . get_response ( self . ext_api ) ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( v , res [ '<STR_LIT>' ] [ k ] ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , attrs ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , <EOL> sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> req . get_response ( self . ext_api ) <EOL> def test_delete_servicechain_instance ( self ) : <EOL> ctx = context . get_admin_context ( ) <EOL> sci = self . create_servicechain_instance ( ) <EOL> sci_id = sci [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> req = self . new_delete_request ( '<STR_LIT>' , sci_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceChainInstanceNotFound , <EOL> self . plugin . get_servicechain_instance , <EOL> ctx , sci_id ) <EOL> def test_create_and_show_service_profile ( self ) : <EOL> attrs = cm . get_create_service_profile_default_attrs ( <EOL> service_type = constants . FIREWALL , vendor = "<STR_LIT>" ) <EOL> scn = self . create_service_profile ( <EOL> service_type = constants . FIREWALL , vendor = "<STR_LIT>" ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( scn [ '<STR_LIT>' ] [ k ] , v ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , attrs ) <EOL> def test_list_service_profile ( self ) : <EOL> scns = [ self . create_service_profile ( name = '<STR_LIT>' , description = '<STR_LIT>' , <EOL> service_type = '<STR_LIT>' ) , <EOL> self . create_service_profile ( name = '<STR_LIT>' , description = '<STR_LIT>' , <EOL> service_type = '<STR_LIT>' ) , <EOL> self . create_service_profile ( name = '<STR_LIT>' , description = '<STR_LIT>' , <EOL> service_type = '<STR_LIT>' ) ] <EOL> self . _test_list_resources ( '<STR_LIT>' , scns , <EOL> query_params = '<STR_LIT>' ) <EOL> def test_update_service_profile ( self ) : <EOL> name = '<STR_LIT>' <EOL> description = '<STR_LIT>' <EOL> attrs = cm . get_create_service_profile_default_attrs ( <EOL> name = name , description = description , <EOL> service_type = constants . FIREWALL ) <EOL> scn = self . create_service_profile ( service_type = constants . FIREWALL ) <EOL> data = { '<STR_LIT>' : { '<STR_LIT:name>' : name , <EOL> '<STR_LIT:description>' : description } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , <EOL> scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . deserialize ( self . fmt , req . get_response ( self . ext_api ) ) <EOL> for k , v in attrs . iteritems ( ) : <EOL> self . assertEqual ( res [ '<STR_LIT>' ] [ k ] , v ) <EOL> self . _test_show_resource ( '<STR_LIT>' , <EOL> scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , attrs ) <EOL> def test_delete_service_profile ( self ) : <EOL> ctx = context . get_admin_context ( ) <EOL> sp = self . create_service_profile ( service_type = '<STR_LIT>' ) <EOL> sp_id = sp [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> scn = self . create_servicechain_node ( service_profile_id = sp_id ) <EOL> scn_id = scn [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . assertRaises ( service_chain . ServiceProfileInUse , <EOL> self . plugin . delete_service_profile , ctx , sp_id ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , scn_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , sp_id ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . assertRaises ( service_chain . ServiceProfileNotFound , <EOL> self . plugin . get_service_profile , <EOL> ctx , sp_id ) <EOL> class TestQuotasForServiceChain ( ServiceChainDbTestCase ) : <EOL> def setUp ( self , core_plugin = None , sc_plugin = None , <EOL> gp_plugin = None , service_plugins = None , ext_mgr = None ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> super ( TestQuotasForServiceChain , self ) . setUp ( <EOL> core_plugin = core_plugin , sc_plugin = sc_plugin , <EOL> gp_plugin = gp_plugin , service_plugins = service_plugins , <EOL> ext_mgr = ext_mgr ) <EOL> def tearDown ( self ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , - <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , - <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , - <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , - <NUM_LIT:1> , <EOL> group = '<STR_LIT>' ) <EOL> super ( TestQuotasForServiceChain , self ) . tearDown ( ) <EOL> def test_servicechain_node_quota ( self ) : <EOL> self . create_servicechain_node ( ) <EOL> self . assertRaises ( webob . exc . HTTPClientError , <EOL> self . create_servicechain_node ) <EOL> def test_servicechain_spec_quota ( self ) : <EOL> self . create_servicechain_spec ( ) <EOL> self . assertRaises ( webob . exc . HTTPClientError , <EOL> self . create_servicechain_spec ) <EOL> def test_servicechain_instance_quota ( self ) : <EOL> self . create_servicechain_instance ( ) <EOL> self . assertRaises ( webob . exc . HTTPClientError , <EOL> self . create_servicechain_instance ) <EOL> def test_service_profile ( self ) : <EOL> self . create_service_profile ( service_type = constants . FIREWALL ) <EOL> self . assertRaises ( webob . exc . HTTPClientError , <EOL> self . create_service_profile , <EOL> service_type = constants . FIREWALL ) </s>
