<s> import mock <EOL> from mox3 import mox <EOL> from oslo_config import cfg <EOL> from oslo_serialization import jsonutils <EOL> from oslo_utils import encodeutils <EOL> import webob <EOL> from nova . api . openstack . compute . legacy_v2 . contrib import security_groups as secgroups_v2 <EOL> from nova . api . openstack . compute import security_groups as secgroups_v21 <EOL> from nova . api . openstack import wsgi <EOL> from nova import compute <EOL> from nova . compute import power_state <EOL> from nova import context as context_maker <EOL> import nova . db <EOL> from nova import exception <EOL> from nova import objects <EOL> from nova import quota <EOL> from nova import test <EOL> from nova . tests . unit . api . openstack import fakes <EOL> from nova . tests . unit import fake_instance <EOL> from nova . tests import uuidsentinel as uuids <EOL> CONF = cfg . CONF <EOL> FAKE_UUID1 = '<STR_LIT>' <EOL> FAKE_UUID2 = '<STR_LIT>' <EOL> UUID_SERVER = uuids . server <EOL> class AttrDict ( dict ) : <EOL> def __getattr__ ( self , k ) : <EOL> return self [ k ] <EOL> def security_group_request_template ( ** kwargs ) : <EOL> sg = kwargs . copy ( ) <EOL> sg . setdefault ( '<STR_LIT:name>' , '<STR_LIT:test>' ) <EOL> sg . setdefault ( '<STR_LIT:description>' , '<STR_LIT>' ) <EOL> return sg <EOL> def security_group_template ( ** kwargs ) : <EOL> sg = kwargs . copy ( ) <EOL> sg . setdefault ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> sg . setdefault ( '<STR_LIT:name>' , '<STR_LIT:test>' ) <EOL> sg . setdefault ( '<STR_LIT:description>' , '<STR_LIT>' ) <EOL> return sg <EOL> def security_group_db ( security_group , id = None ) : <EOL> attrs = security_group . copy ( ) <EOL> if '<STR_LIT>' in attrs : <EOL> attrs [ '<STR_LIT>' ] = attrs . pop ( '<STR_LIT>' ) <EOL> if id is not None : <EOL> attrs [ '<STR_LIT:id>' ] = id <EOL> attrs . setdefault ( '<STR_LIT>' , [ ] ) <EOL> attrs . setdefault ( '<STR_LIT>' , [ ] ) <EOL> return AttrDict ( attrs ) <EOL> def security_group_rule_template ( ** kwargs ) : <EOL> rule = kwargs . copy ( ) <EOL> rule . setdefault ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> rule . setdefault ( '<STR_LIT>' , <NUM_LIT> ) <EOL> rule . setdefault ( '<STR_LIT>' , <NUM_LIT> ) <EOL> rule . setdefault ( '<STR_LIT>' , <NUM_LIT:2> ) <EOL> return rule <EOL> def security_group_rule_db ( rule , id = None ) : <EOL> attrs = rule . copy ( ) <EOL> if '<STR_LIT>' in attrs : <EOL> attrs [ '<STR_LIT>' ] = attrs . pop ( '<STR_LIT>' ) <EOL> return AttrDict ( attrs ) <EOL> def return_server ( context , server_id , <EOL> columns_to_join = None , use_slave = False ) : <EOL> return fake_instance . fake_db_instance ( <EOL> ** { '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:host>' : "<STR_LIT:localhost>" , <EOL> '<STR_LIT>' : server_id , <EOL> '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> def return_server_by_uuid ( context , server_uuid , <EOL> columns_to_join = None , <EOL> use_slave = False ) : <EOL> return fake_instance . fake_db_instance ( <EOL> ** { '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:host>' : "<STR_LIT:localhost>" , <EOL> '<STR_LIT>' : server_uuid , <EOL> '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> def return_non_running_server ( context , server_id , columns_to_join = None ) : <EOL> return fake_instance . fake_db_instance ( <EOL> ** { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : power_state . SHUTDOWN , <EOL> '<STR_LIT>' : server_id , '<STR_LIT:host>' : "<STR_LIT:localhost>" , '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> def return_security_group_by_name ( context , project_id , group_name ) : <EOL> return { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : group_name , <EOL> "<STR_LIT>" : [ { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : UUID_SERVER } ] } <EOL> def return_security_group_without_instances ( context , project_id , group_name ) : <EOL> return { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : group_name } <EOL> def return_server_nonexistent ( context , server_id , columns_to_join = None ) : <EOL> raise exception . InstanceNotFound ( instance_id = server_id ) <EOL> class TestSecurityGroupsV21 ( test . TestCase ) : <EOL> secgrp_ctl_cls = secgroups_v21 . SecurityGroupController <EOL> server_secgrp_ctl_cls = secgroups_v21 . ServerSecurityGroupController <EOL> secgrp_act_ctl_cls = secgroups_v21 . SecurityGroupActionController <EOL> def setUp ( self ) : <EOL> super ( TestSecurityGroupsV21 , self ) . setUp ( ) <EOL> self . controller = self . secgrp_ctl_cls ( ) <EOL> self . server_controller = self . server_secgrp_ctl_cls ( ) <EOL> self . manager = self . secgrp_act_ctl_cls ( ) <EOL> if self . controller . security_group_api . id_is_uuid : <EOL> self . fake_id = '<STR_LIT>' <EOL> else : <EOL> self . fake_id = '<STR_LIT>' <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> self . admin_req = fakes . HTTPRequest . blank ( '<STR_LIT>' , use_admin_context = True ) <EOL> def _assert_no_security_groups_reserved ( self , context ) : <EOL> """<STR_LIT>""" <EOL> result = quota . QUOTAS . get_project_quotas ( context , context . project_id ) <EOL> self . assertEqual ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> def _assert_security_groups_in_use ( self , project_id , user_id , in_use ) : <EOL> context = context_maker . get_admin_context ( ) <EOL> result = quota . QUOTAS . get_user_quotas ( context , project_id , user_id ) <EOL> self . assertEqual ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] , in_use ) <EOL> def test_create_security_group ( self ) : <EOL> sg = security_group_request_template ( ) <EOL> res_dict = self . controller . create ( self . req , { '<STR_LIT>' : sg } ) <EOL> self . assertEqual ( res_dict [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , '<STR_LIT:test>' ) <EOL> self . assertEqual ( res_dict [ '<STR_LIT>' ] [ '<STR_LIT:description>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_create_security_group_with_no_name ( self ) : <EOL> sg = security_group_request_template ( ) <EOL> del sg [ '<STR_LIT:name>' ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . create , self . req , <EOL> { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_no_description ( self ) : <EOL> sg = security_group_request_template ( ) <EOL> del sg [ '<STR_LIT:description>' ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_empty_description ( self ) : <EOL> sg = security_group_request_template ( ) <EOL> sg [ '<STR_LIT:description>' ] = "<STR_LIT>" <EOL> try : <EOL> self . controller . create ( self . req , { '<STR_LIT>' : sg } ) <EOL> self . fail ( '<STR_LIT>' ) <EOL> except webob . exc . HTTPBadRequest as exc : <EOL> self . assertEqual ( '<STR_LIT>' <EOL> '<STR_LIT>' , exc . explanation ) <EOL> except exception . InvalidInput : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_blank_name ( self ) : <EOL> sg = security_group_request_template ( name = '<STR_LIT>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_whitespace_name ( self ) : <EOL> sg = security_group_request_template ( name = '<STR_LIT:U+0020>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_blank_description ( self ) : <EOL> sg = security_group_request_template ( description = '<STR_LIT>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_whitespace_description ( self ) : <EOL> sg = security_group_request_template ( description = '<STR_LIT:U+0020>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_duplicate_name ( self ) : <EOL> sg = security_group_request_template ( ) <EOL> self . controller . create ( self . req , { '<STR_LIT>' : sg } ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_no_body ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . create , self . req , None ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_with_no_security_group ( self ) : <EOL> body = { '<STR_LIT>' : None } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . create , self . req , body ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_above_255_characters_name ( self ) : <EOL> sg = security_group_request_template ( name = '<STR_LIT>' * <NUM_LIT> ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_above_255_characters_description ( self ) : <EOL> sg = security_group_request_template ( description = '<STR_LIT>' * <NUM_LIT> ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_non_string_name ( self ) : <EOL> sg = security_group_request_template ( name = <NUM_LIT:12> ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_non_string_description ( self ) : <EOL> sg = security_group_request_template ( description = <NUM_LIT:12> ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> self . _assert_no_security_groups_reserved ( <EOL> self . req . environ [ '<STR_LIT>' ] ) <EOL> def test_create_security_group_quota_limit ( self ) : <EOL> for num in range ( <NUM_LIT:1> , CONF . quota_security_groups ) : <EOL> name = '<STR_LIT>' % num <EOL> sg = security_group_request_template ( name = name ) <EOL> res_dict = self . controller . create ( self . req , { '<STR_LIT>' : sg } ) <EOL> self . assertEqual ( res_dict [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , name ) <EOL> sg = security_group_request_template ( ) <EOL> self . assertRaises ( webob . exc . HTTPForbidden , self . controller . create , <EOL> self . req , { '<STR_LIT>' : sg } ) <EOL> def test_get_security_group_list ( self ) : <EOL> groups = [ ] <EOL> for i , name in enumerate ( [ '<STR_LIT:default>' , '<STR_LIT:test>' ] ) : <EOL> sg = security_group_template ( id = i + <NUM_LIT:1> , <EOL> name = name , <EOL> description = name + '<STR_LIT>' , <EOL> rules = [ ] ) <EOL> groups . append ( sg ) <EOL> expected = { '<STR_LIT>' : groups } <EOL> def return_security_groups ( context , project_id ) : <EOL> return [ security_group_db ( sg ) for sg in groups ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_groups ) <EOL> res_dict = self . controller . index ( self . req ) <EOL> self . assertEqual ( res_dict , expected ) <EOL> def test_get_security_group_list_missing_group_id_rule ( self ) : <EOL> groups = [ ] <EOL> rule1 = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = <NUM_LIT:1> , <EOL> group_id = { } , id = <NUM_LIT> , <EOL> protocol = '<STR_LIT>' ) <EOL> rule2 = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = <NUM_LIT:1> , <EOL> id = <NUM_LIT> , protocol = <NUM_LIT> , <EOL> group_id = '<STR_LIT>' ) <EOL> sg = security_group_template ( id = <NUM_LIT:1> , <EOL> name = '<STR_LIT:test>' , <EOL> description = '<STR_LIT>' , <EOL> rules = [ rule1 , rule2 ] ) <EOL> groups . append ( sg ) <EOL> expected_rule = security_group_rule_template ( <EOL> ip_range = { '<STR_LIT>' : '<STR_LIT>' } , parent_group_id = <NUM_LIT:1> , <EOL> group = { } , id = <NUM_LIT> , ip_protocol = '<STR_LIT>' ) <EOL> expected = security_group_template ( id = <NUM_LIT:1> , <EOL> name = '<STR_LIT:test>' , <EOL> description = '<STR_LIT>' , <EOL> rules = [ expected_rule ] ) <EOL> expected = { '<STR_LIT>' : [ expected ] } <EOL> def return_security_groups ( context , project , search_opts ) : <EOL> return [ security_group_db ( sg ) for sg in groups ] <EOL> self . stubs . Set ( self . controller . security_group_api , '<STR_LIT:list>' , <EOL> return_security_groups ) <EOL> res_dict = self . controller . index ( self . req ) <EOL> self . assertEqual ( res_dict , expected ) <EOL> def test_get_security_group_list_all_tenants ( self ) : <EOL> all_groups = [ ] <EOL> tenant_groups = [ ] <EOL> for i , name in enumerate ( [ '<STR_LIT:default>' , '<STR_LIT:test>' ] ) : <EOL> sg = security_group_template ( id = i + <NUM_LIT:1> , <EOL> name = name , <EOL> description = name + '<STR_LIT>' , <EOL> rules = [ ] ) <EOL> all_groups . append ( sg ) <EOL> if name == '<STR_LIT:default>' : <EOL> tenant_groups . append ( sg ) <EOL> all = { '<STR_LIT>' : all_groups } <EOL> tenant_specific = { '<STR_LIT>' : tenant_groups } <EOL> def return_all_security_groups ( context ) : <EOL> return [ security_group_db ( sg ) for sg in all_groups ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_all_security_groups ) <EOL> def return_tenant_security_groups ( context , project_id ) : <EOL> return [ security_group_db ( sg ) for sg in tenant_groups ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_tenant_security_groups ) <EOL> path = '<STR_LIT>' <EOL> req = fakes . HTTPRequest . blank ( path , use_admin_context = True ) <EOL> res_dict = self . controller . index ( req ) <EOL> self . assertEqual ( res_dict , tenant_specific ) <EOL> req = fakes . HTTPRequest . blank ( '<STR_LIT>' % path , <EOL> use_admin_context = True ) <EOL> res_dict = self . controller . index ( req ) <EOL> self . assertEqual ( res_dict , all ) <EOL> def test_get_security_group_by_instance ( self ) : <EOL> groups = [ ] <EOL> for i , name in enumerate ( [ '<STR_LIT:default>' , '<STR_LIT:test>' ] ) : <EOL> sg = security_group_template ( id = i + <NUM_LIT:1> , <EOL> name = name , <EOL> description = name + '<STR_LIT>' , <EOL> rules = [ ] ) <EOL> groups . append ( sg ) <EOL> expected = { '<STR_LIT>' : groups } <EOL> def return_instance ( context , server_id , <EOL> columns_to_join = None , use_slave = False ) : <EOL> self . assertEqual ( server_id , FAKE_UUID1 ) <EOL> return return_server_by_uuid ( context , server_id ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_instance ) <EOL> def return_security_groups ( context , instance_uuid ) : <EOL> self . assertEqual ( instance_uuid , FAKE_UUID1 ) <EOL> return [ security_group_db ( sg ) for sg in groups ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_groups ) <EOL> res_dict = self . server_controller . index ( self . req , FAKE_UUID1 ) <EOL> self . assertEqual ( res_dict , expected ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' , return_value = [ ] ) <EOL> def test_get_security_group_empty_for_instance ( self , mock_sec_group , <EOL> mock_db_get_ins ) : <EOL> expected = { '<STR_LIT>' : [ ] } <EOL> def return_instance ( context , server_id , <EOL> columns_to_join = None , use_slave = False ) : <EOL> self . assertEqual ( server_id , FAKE_UUID1 ) <EOL> return return_server_by_uuid ( context , server_id ) <EOL> mock_db_get_ins . side_effect = return_instance <EOL> res_dict = self . server_controller . index ( self . req , FAKE_UUID1 ) <EOL> self . assertEqual ( expected , res_dict ) <EOL> mock_sec_group . assert_called_once_with ( <EOL> self . req . environ [ '<STR_LIT>' ] , FAKE_UUID1 ) <EOL> def test_get_security_group_by_instance_non_existing ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server_nonexistent ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_nonexistent ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . server_controller . index , self . req , '<STR_LIT:1>' ) <EOL> def test_get_security_group_by_instance_invalid_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . server_controller . index , self . req , '<STR_LIT>' ) <EOL> def test_get_security_group_by_id ( self ) : <EOL> sg = security_group_template ( id = <NUM_LIT:2> , rules = [ ] ) <EOL> def return_security_group ( context , group_id ) : <EOL> self . assertEqual ( sg [ '<STR_LIT:id>' ] , group_id ) <EOL> return security_group_db ( sg ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group ) <EOL> res_dict = self . controller . show ( self . req , '<STR_LIT:2>' ) <EOL> expected = { '<STR_LIT>' : sg } <EOL> self . assertEqual ( res_dict , expected ) <EOL> def test_get_security_group_by_invalid_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . delete , <EOL> self . req , '<STR_LIT>' ) <EOL> def test_get_security_group_by_non_existing_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPNotFound , self . controller . delete , <EOL> self . req , self . fake_id ) <EOL> def test_update_security_group ( self ) : <EOL> sg = security_group_template ( id = <NUM_LIT:2> , rules = [ ] ) <EOL> sg_update = security_group_template ( id = <NUM_LIT:2> , rules = [ ] , <EOL> name = '<STR_LIT>' , description = '<STR_LIT>' ) <EOL> def return_security_group ( context , group_id ) : <EOL> self . assertEqual ( sg [ '<STR_LIT:id>' ] , group_id ) <EOL> return security_group_db ( sg ) <EOL> def return_update_security_group ( context , group_id , values , <EOL> columns_to_join = None ) : <EOL> self . assertEqual ( sg_update [ '<STR_LIT:id>' ] , group_id ) <EOL> self . assertEqual ( sg_update [ '<STR_LIT:name>' ] , values [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( sg_update [ '<STR_LIT:description>' ] , values [ '<STR_LIT:description>' ] ) <EOL> return security_group_db ( sg_update ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_update_security_group ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group ) <EOL> res_dict = self . controller . update ( self . req , '<STR_LIT:2>' , <EOL> { '<STR_LIT>' : sg_update } ) <EOL> expected = { '<STR_LIT>' : sg_update } <EOL> self . assertEqual ( res_dict , expected ) <EOL> def test_update_security_group_name_to_default ( self ) : <EOL> sg = security_group_template ( id = <NUM_LIT:2> , rules = [ ] , name = '<STR_LIT:default>' ) <EOL> def return_security_group ( context , group_id ) : <EOL> self . assertEqual ( sg [ '<STR_LIT:id>' ] , group_id ) <EOL> return security_group_db ( sg ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . update , <EOL> self . req , '<STR_LIT:2>' , { '<STR_LIT>' : sg } ) <EOL> def test_update_default_security_group_fail ( self ) : <EOL> sg = security_group_template ( ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . update , <EOL> self . req , '<STR_LIT:1>' , { '<STR_LIT>' : sg } ) <EOL> def test_delete_security_group_by_id ( self ) : <EOL> sg = security_group_template ( id = <NUM_LIT:1> , project_id = '<STR_LIT>' , <EOL> user_id = '<STR_LIT>' , rules = [ ] ) <EOL> self . called = False <EOL> def security_group_destroy ( context , id ) : <EOL> self . called = True <EOL> def return_security_group ( context , group_id ) : <EOL> self . assertEqual ( sg [ '<STR_LIT:id>' ] , group_id ) <EOL> return security_group_db ( sg ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> security_group_destroy ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group ) <EOL> self . controller . delete ( self . req , '<STR_LIT:1>' ) <EOL> self . assertTrue ( self . called ) <EOL> def test_delete_security_group_by_admin ( self ) : <EOL> sg = security_group_request_template ( ) <EOL> self . controller . create ( self . req , { '<STR_LIT>' : sg } ) <EOL> context = self . req . environ [ '<STR_LIT>' ] <EOL> self . _assert_security_groups_in_use ( context . project_id , <EOL> context . user_id , <NUM_LIT:2> ) <EOL> self . controller . delete ( self . admin_req , '<STR_LIT:2>' ) <EOL> self . _assert_security_groups_in_use ( context . project_id , <EOL> context . user_id , <NUM_LIT:1> ) <EOL> def test_delete_security_group_by_invalid_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . delete , <EOL> self . req , '<STR_LIT>' ) <EOL> def test_delete_security_group_by_non_existing_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPNotFound , self . controller . delete , <EOL> self . req , self . fake_id ) <EOL> def test_delete_security_group_in_use ( self ) : <EOL> sg = security_group_template ( id = <NUM_LIT:1> , rules = [ ] ) <EOL> def security_group_in_use ( context , id ) : <EOL> return True <EOL> def return_security_group ( context , group_id ) : <EOL> self . assertEqual ( sg [ '<STR_LIT:id>' ] , group_id ) <EOL> return security_group_db ( sg ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> security_group_in_use ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . delete , <EOL> self . req , '<STR_LIT:1>' ) <EOL> def test_associate_by_non_existing_security_group_name ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self . assertEqual ( return_server ( None , '<STR_LIT:1>' ) , <EOL> nova . db . instance_get ( None , '<STR_LIT:1>' ) ) <EOL> body = dict ( addSecurityGroup = dict ( name = '<STR_LIT>' ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . manager . _addSecurityGroup , self . req , '<STR_LIT:1>' , body ) <EOL> def test_associate_by_invalid_server_id ( self ) : <EOL> body = dict ( addSecurityGroup = dict ( name = '<STR_LIT:test>' ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . manager . _addSecurityGroup , self . req , <EOL> '<STR_LIT>' , body ) <EOL> def test_associate_without_body ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = dict ( addSecurityGroup = None ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _addSecurityGroup , self . req , '<STR_LIT:1>' , body ) <EOL> def test_associate_no_security_group_name ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = dict ( addSecurityGroup = dict ( ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _addSecurityGroup , self . req , '<STR_LIT:1>' , body ) <EOL> def test_associate_security_group_name_with_whitespaces ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = dict ( addSecurityGroup = dict ( name = "<STR_LIT:U+0020>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _addSecurityGroup , self . req , '<STR_LIT:1>' , body ) <EOL> def test_associate_non_existing_instance ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server_nonexistent ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_nonexistent ) <EOL> body = dict ( addSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . manager . _addSecurityGroup , self . req , '<STR_LIT:1>' , body ) <EOL> def test_associate_non_running_instance ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_non_running_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_non_running_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_without_instances ) <EOL> body = dict ( addSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . manager . _addSecurityGroup ( self . req , UUID_SERVER , body ) <EOL> def test_associate_already_associated_security_group_to_instance ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_by_uuid ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_by_name ) <EOL> body = dict ( addSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _addSecurityGroup , self . req , <EOL> UUID_SERVER , body ) <EOL> def test_associate ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_by_uuid ) <EOL> self . mox . StubOutWithMock ( nova . db , '<STR_LIT>' ) <EOL> nova . db . instance_add_security_group ( mox . IgnoreArg ( ) , <EOL> mox . IgnoreArg ( ) , <EOL> mox . IgnoreArg ( ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_without_instances ) <EOL> self . mox . ReplayAll ( ) <EOL> body = dict ( addSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . manager . _addSecurityGroup ( self . req , UUID_SERVER , body ) <EOL> def test_disassociate_by_non_existing_security_group_name ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self . assertEqual ( return_server ( None , '<STR_LIT:1>' ) , <EOL> nova . db . instance_get ( None , '<STR_LIT:1>' ) ) <EOL> body = dict ( removeSecurityGroup = dict ( name = '<STR_LIT>' ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . manager . _removeSecurityGroup , self . req , <EOL> UUID_SERVER , body ) <EOL> def test_disassociate_by_invalid_server_id ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_by_name ) <EOL> body = dict ( removeSecurityGroup = dict ( name = '<STR_LIT:test>' ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . manager . _removeSecurityGroup , self . req , <EOL> '<STR_LIT>' , body ) <EOL> def test_disassociate_without_body ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = dict ( removeSecurityGroup = None ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _removeSecurityGroup , self . req , <EOL> '<STR_LIT:1>' , body ) <EOL> def test_disassociate_no_security_group_name ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = dict ( removeSecurityGroup = dict ( ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _removeSecurityGroup , self . req , <EOL> '<STR_LIT:1>' , body ) <EOL> def test_disassociate_security_group_name_with_whitespaces ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = dict ( removeSecurityGroup = dict ( name = "<STR_LIT:U+0020>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _removeSecurityGroup , self . req , <EOL> '<STR_LIT:1>' , body ) <EOL> def test_disassociate_non_existing_instance ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server_nonexistent ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_by_name ) <EOL> body = dict ( removeSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . manager . _removeSecurityGroup , <EOL> self . req , '<STR_LIT:1>' , body ) <EOL> def test_disassociate_non_running_instance ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_non_running_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_non_running_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_by_name ) <EOL> body = dict ( removeSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . manager . _removeSecurityGroup ( self . req , UUID_SERVER , body ) <EOL> def test_disassociate_already_associated_security_group_to_instance ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_by_uuid ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_without_instances ) <EOL> body = dict ( removeSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . manager . _removeSecurityGroup , self . req , <EOL> UUID_SERVER , body ) <EOL> def test_disassociate ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_by_uuid ) <EOL> self . mox . StubOutWithMock ( nova . db , '<STR_LIT>' ) <EOL> nova . db . instance_remove_security_group ( mox . IgnoreArg ( ) , <EOL> mox . IgnoreArg ( ) , <EOL> mox . IgnoreArg ( ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group_by_name ) <EOL> self . mox . ReplayAll ( ) <EOL> body = dict ( removeSecurityGroup = dict ( name = "<STR_LIT:test>" ) ) <EOL> self . manager . _removeSecurityGroup ( self . req , UUID_SERVER , body ) <EOL> class TestSecurityGroupsV2 ( TestSecurityGroupsV21 ) : <EOL> secgrp_ctl_cls = secgroups_v2 . SecurityGroupController <EOL> server_secgrp_ctl_cls = secgroups_v2 . ServerSecurityGroupController <EOL> secgrp_act_ctl_cls = secgroups_v2 . SecurityGroupActionController <EOL> class TestSecurityGroupRulesV21 ( test . TestCase ) : <EOL> secgrp_ctl_cls = secgroups_v21 . SecurityGroupRulesController <EOL> def setUp ( self ) : <EOL> super ( TestSecurityGroupRulesV21 , self ) . setUp ( ) <EOL> self . controller = self . secgrp_ctl_cls ( ) <EOL> if self . controller . security_group_api . id_is_uuid : <EOL> id1 = '<STR_LIT>' <EOL> id2 = '<STR_LIT>' <EOL> self . invalid_id = '<STR_LIT>' <EOL> else : <EOL> id1 = <NUM_LIT:1> <EOL> id2 = <NUM_LIT:2> <EOL> self . invalid_id = '<STR_LIT>' <EOL> self . sg1 = security_group_template ( id = id1 ) <EOL> self . sg2 = security_group_template ( <EOL> id = id2 , name = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) <EOL> db1 = security_group_db ( self . sg1 ) <EOL> db2 = security_group_db ( self . sg2 ) <EOL> def return_security_group ( context , group_id , columns_to_join = None ) : <EOL> if group_id == db1 [ '<STR_LIT:id>' ] : <EOL> return db1 <EOL> if group_id == db2 [ '<STR_LIT:id>' ] : <EOL> return db2 <EOL> raise exception . SecurityGroupNotFound ( security_group_id = group_id ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_security_group ) <EOL> self . parent_security_group = db2 <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> def test_create_by_cidr ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_create_by_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = self . sg1 [ '<STR_LIT:id>' ] , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg2 [ '<STR_LIT:id>' ] ) <EOL> def test_create_by_same_group_id ( self ) : <EOL> rule1 = security_group_rule_template ( group_id = self . sg1 [ '<STR_LIT:id>' ] , <EOL> from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . parent_security_group [ '<STR_LIT>' ] = [ security_group_rule_db ( rule1 ) ] <EOL> rule2 = security_group_rule_template ( group_id = self . sg1 [ '<STR_LIT:id>' ] , <EOL> from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule2 } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> def test_create_none_value_from_to_port ( self ) : <EOL> rule = { '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] } <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertIsNone ( security_group_rule [ '<STR_LIT>' ] ) <EOL> self . assertIsNone ( security_group_rule [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , '<STR_LIT:test>' ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg1 [ '<STR_LIT:id>' ] ) <EOL> def test_create_none_value_from_to_port_icmp ( self ) : <EOL> rule = { '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , - <NUM_LIT:1> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , - <NUM_LIT:1> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , '<STR_LIT:test>' ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg1 [ '<STR_LIT:id>' ] ) <EOL> def test_create_none_value_from_to_port_tcp ( self ) : <EOL> rule = { '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , '<STR_LIT:test>' ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg1 [ '<STR_LIT:id>' ] ) <EOL> def test_create_by_invalid_cidr_json ( self ) : <EOL> rule = security_group_rule_template ( <EOL> ip_protocol = "<STR_LIT>" , <EOL> from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] , <EOL> cidr = "<STR_LIT>" ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_by_invalid_tcp_port_json ( self ) : <EOL> rule = security_group_rule_template ( <EOL> ip_protocol = "<STR_LIT>" , <EOL> from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] , <EOL> cidr = "<STR_LIT>" ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_by_invalid_icmp_port_json ( self ) : <EOL> rule = security_group_rule_template ( <EOL> ip_protocol = "<STR_LIT>" , <EOL> from_port = <NUM_LIT:1> , <EOL> to_port = <NUM_LIT> , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] , <EOL> cidr = "<STR_LIT>" ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_add_existing_rules_by_cidr ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . parent_security_group [ '<STR_LIT>' ] = [ security_group_rule_db ( rule ) ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_add_existing_rules_by_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = <NUM_LIT:1> ) <EOL> self . parent_security_group [ '<STR_LIT>' ] = [ security_group_rule_db ( rule ) ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_no_body ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . create , self . req , None ) <EOL> def test_create_with_no_security_group_rule_in_body ( self ) : <EOL> rules = { '<STR_LIT:test>' : '<STR_LIT:test>' } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . create , self . req , rules ) <EOL> def test_create_with_invalid_parent_group_id ( self ) : <EOL> rule = security_group_rule_template ( parent_group_id = '<STR_LIT>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_non_existing_parent_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = None , <EOL> parent_group_id = self . invalid_id ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_non_existing_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_invalid_protocol ( self ) : <EOL> rule = security_group_rule_template ( ip_protocol = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_no_protocol ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> del rule [ '<STR_LIT>' ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_invalid_from_port ( self ) : <EOL> rule = security_group_rule_template ( from_port = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_invalid_to_port ( self ) : <EOL> rule = security_group_rule_template ( to_port = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_non_numerical_from_port ( self ) : <EOL> rule = security_group_rule_template ( from_port = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_non_numerical_to_port ( self ) : <EOL> rule = security_group_rule_template ( to_port = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_no_from_port ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> del rule [ '<STR_LIT>' ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_no_to_port ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> del rule [ '<STR_LIT>' ] <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_invalid_cidr ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_no_cidr_group ( self ) : <EOL> rule = security_group_rule_template ( parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . parent_security_group [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_create_with_invalid_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_empty_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_nonexist_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = self . invalid_id , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_with_same_group_parent_id_and_group_id ( self ) : <EOL> rule = security_group_rule_template ( group_id = self . sg1 [ '<STR_LIT:id>' ] , <EOL> parent_group_id = self . sg1 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . sg1 [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , <EOL> self . sg1 [ '<STR_LIT:name>' ] ) <EOL> def _test_create_with_no_ports_and_no_group ( self , proto ) : <EOL> rule = { '<STR_LIT>' : proto , '<STR_LIT>' : self . sg2 [ '<STR_LIT:id>' ] } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def _test_create_with_no_ports ( self , proto ) : <EOL> rule = { '<STR_LIT>' : proto , '<STR_LIT>' : self . sg2 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] } <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> expected_rule = { <EOL> '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT:test>' } , <EOL> '<STR_LIT>' : proto , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <EOL> self . sg2 [ '<STR_LIT:id>' ] , '<STR_LIT>' : { } , '<STR_LIT:id>' : security_group_rule [ '<STR_LIT:id>' ] <EOL> } <EOL> if proto == '<STR_LIT>' : <EOL> expected_rule [ '<STR_LIT>' ] = - <NUM_LIT:1> <EOL> expected_rule [ '<STR_LIT>' ] = - <NUM_LIT:1> <EOL> self . assertEqual ( expected_rule , security_group_rule ) <EOL> def test_create_with_no_ports_icmp ( self ) : <EOL> self . _test_create_with_no_ports_and_no_group ( '<STR_LIT>' ) <EOL> self . _test_create_with_no_ports ( '<STR_LIT>' ) <EOL> def test_create_with_no_ports_tcp ( self ) : <EOL> self . _test_create_with_no_ports_and_no_group ( '<STR_LIT>' ) <EOL> self . _test_create_with_no_ports ( '<STR_LIT>' ) <EOL> def test_create_with_no_ports_udp ( self ) : <EOL> self . _test_create_with_no_ports_and_no_group ( '<STR_LIT>' ) <EOL> self . _test_create_with_no_ports ( '<STR_LIT>' ) <EOL> def _test_create_with_ports ( self , proto , from_port , to_port ) : <EOL> rule = { <EOL> '<STR_LIT>' : proto , '<STR_LIT>' : from_port , '<STR_LIT>' : to_port , <EOL> '<STR_LIT>' : self . sg2 [ '<STR_LIT:id>' ] , '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] <EOL> } <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> expected_rule = { <EOL> '<STR_LIT>' : from_port , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT:test>' } , <EOL> '<STR_LIT>' : proto , '<STR_LIT>' : to_port , '<STR_LIT>' : <EOL> self . sg2 [ '<STR_LIT:id>' ] , '<STR_LIT>' : { } , '<STR_LIT:id>' : security_group_rule [ '<STR_LIT:id>' ] <EOL> } <EOL> self . assertEqual ( proto , security_group_rule [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( from_port , security_group_rule [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( to_port , security_group_rule [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( expected_rule , security_group_rule ) <EOL> def test_create_with_ports_icmp ( self ) : <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> def test_create_with_ports_tcp ( self ) : <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT> ) <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) <EOL> def test_create_with_ports_udp ( self ) : <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT> ) <EOL> self . _test_create_with_ports ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) <EOL> def test_delete ( self ) : <EOL> rule = security_group_rule_template ( id = self . sg2 [ '<STR_LIT:id>' ] , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> def security_group_rule_get ( context , id ) : <EOL> return security_group_rule_db ( rule ) <EOL> def security_group_rule_destroy ( context , id ) : <EOL> pass <EOL> self . stub_out ( '<STR_LIT>' , <EOL> security_group_rule_get ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> security_group_rule_destroy ) <EOL> self . controller . delete ( self . req , self . sg2 [ '<STR_LIT:id>' ] ) <EOL> def test_delete_invalid_rule_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . delete , <EOL> self . req , '<STR_LIT>' ) <EOL> def test_delete_non_existing_rule_id ( self ) : <EOL> self . assertRaises ( webob . exc . HTTPNotFound , self . controller . delete , <EOL> self . req , self . invalid_id ) <EOL> def test_create_rule_quota_limit ( self ) : <EOL> for num in range ( <NUM_LIT:100> , <NUM_LIT:100> + CONF . quota_security_group_rules ) : <EOL> rule = { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : num , <EOL> '<STR_LIT>' : num , '<STR_LIT>' : self . sg2 [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] <EOL> } <EOL> self . controller . create ( self . req , { '<STR_LIT>' : rule } ) <EOL> rule = { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . sg2 [ '<STR_LIT:id>' ] , '<STR_LIT>' : self . sg1 [ '<STR_LIT:id>' ] <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPForbidden , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> def test_create_rule_cidr_allow_all ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . parent_security_group [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_create_rule_cidr_ipv6_allow_all ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . parent_security_group [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_create_rule_cidr_allow_some ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' , <EOL> parent_group_id = self . sg2 [ '<STR_LIT:id>' ] ) <EOL> res_dict = self . controller . create ( self . req , <EOL> { '<STR_LIT>' : rule } ) <EOL> security_group_rule = res_dict [ '<STR_LIT>' ] <EOL> self . assertNotEqual ( security_group_rule [ '<STR_LIT:id>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] , <EOL> self . parent_security_group [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( security_group_rule [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_create_rule_cidr_bad_netmask ( self ) : <EOL> rule = security_group_rule_template ( cidr = '<STR_LIT>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , self . controller . create , <EOL> self . req , { '<STR_LIT>' : rule } ) <EOL> class TestSecurityGroupRulesV2 ( TestSecurityGroupRulesV21 ) : <EOL> secgrp_ctl_cls = secgroups_v2 . SecurityGroupRulesController <EOL> UUID1 = '<STR_LIT>' <EOL> UUID2 = '<STR_LIT>' <EOL> UUID3 = '<STR_LIT>' <EOL> def fake_compute_get_all ( * args , ** kwargs ) : <EOL> base = { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:description>' : '<STR_LIT:foo>' , '<STR_LIT>' : '<STR_LIT:bar>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : False , '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , '<STR_LIT>' : None } <EOL> inst_list = [ <EOL> fakes . stub_instance_obj ( <EOL> None , <NUM_LIT:1> , uuid = UUID1 , <EOL> security_groups = [ dict ( base , ** { '<STR_LIT:name>' : '<STR_LIT>' } ) , <EOL> dict ( base , ** { '<STR_LIT:name>' : '<STR_LIT>' } ) ] ) , <EOL> fakes . stub_instance_obj ( <EOL> None , <NUM_LIT:2> , uuid = UUID2 , <EOL> security_groups = [ dict ( base , ** { '<STR_LIT:name>' : '<STR_LIT>' } ) , <EOL> dict ( base , ** { '<STR_LIT:name>' : '<STR_LIT>' } ) ] ) <EOL> ] <EOL> return objects . InstanceList ( objects = inst_list ) <EOL> def fake_compute_get ( * args , ** kwargs ) : <EOL> secgroups = objects . SecurityGroupList ( ) <EOL> secgroups . objects = [ <EOL> objects . SecurityGroup ( name = '<STR_LIT>' ) , <EOL> objects . SecurityGroup ( name = '<STR_LIT>' ) , <EOL> ] <EOL> inst = fakes . stub_instance_obj ( None , <NUM_LIT:1> , uuid = UUID3 ) <EOL> inst . security_groups = secgroups <EOL> return inst <EOL> def fake_compute_create ( * args , ** kwargs ) : <EOL> return ( [ fake_compute_get ( * args , ** kwargs ) ] , '<STR_LIT>' ) <EOL> def fake_get_instances_security_groups_bindings ( inst , context , servers ) : <EOL> groups = { UUID1 : [ { '<STR_LIT:name>' : '<STR_LIT>' } , { '<STR_LIT:name>' : '<STR_LIT>' } ] , <EOL> UUID2 : [ { '<STR_LIT:name>' : '<STR_LIT>' } , { '<STR_LIT:name>' : '<STR_LIT>' } ] , <EOL> UUID3 : [ { '<STR_LIT:name>' : '<STR_LIT>' } , { '<STR_LIT:name>' : '<STR_LIT>' } ] } <EOL> result = { } <EOL> for server in servers : <EOL> result [ server [ '<STR_LIT:id>' ] ] = groups . get ( server [ '<STR_LIT:id>' ] ) <EOL> return result <EOL> class SecurityGroupsOutputTestV21 ( test . TestCase ) : <EOL> base_url = '<STR_LIT>' <EOL> content_type = '<STR_LIT:application/json>' <EOL> def setUp ( self ) : <EOL> super ( SecurityGroupsOutputTestV21 , self ) . setUp ( ) <EOL> fakes . stub_out_nw_api ( self ) <EOL> self . stubs . Set ( compute . api . API , '<STR_LIT>' , fake_compute_get ) <EOL> self . stubs . Set ( compute . api . API , '<STR_LIT>' , fake_compute_get_all ) <EOL> self . stubs . Set ( compute . api . API , '<STR_LIT>' , fake_compute_create ) <EOL> self . flags ( <EOL> osapi_compute_extension = [ <EOL> '<STR_LIT>' ] , <EOL> osapi_compute_ext_list = [ '<STR_LIT>' ] ) <EOL> self . app = self . _setup_app ( ) <EOL> def _setup_app ( self ) : <EOL> return fakes . wsgi_app_v21 ( init_only = ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def _make_request ( self , url , body = None ) : <EOL> req = webob . Request . blank ( url ) <EOL> if body : <EOL> req . method = '<STR_LIT:POST>' <EOL> req . body = encodeutils . safe_encode ( self . _encode_body ( body ) ) <EOL> req . content_type = self . content_type <EOL> req . headers [ '<STR_LIT>' ] = self . content_type <EOL> res = req . get_response ( self . app ) <EOL> return res <EOL> def _encode_body ( self , body ) : <EOL> return jsonutils . dumps ( body ) <EOL> def _get_server ( self , body ) : <EOL> return jsonutils . loads ( body ) . get ( '<STR_LIT>' ) <EOL> def _get_servers ( self , body ) : <EOL> return jsonutils . loads ( body ) . get ( '<STR_LIT>' ) <EOL> def _get_groups ( self , server ) : <EOL> return server . get ( '<STR_LIT>' ) <EOL> def test_create ( self ) : <EOL> image_uuid = '<STR_LIT>' <EOL> server = dict ( name = '<STR_LIT>' , imageRef = image_uuid , flavorRef = <NUM_LIT:2> ) <EOL> res = self . _make_request ( self . base_url , { '<STR_LIT>' : server } ) <EOL> self . assertEqual ( res . status_int , <NUM_LIT> ) <EOL> server = self . _get_server ( res . body ) <EOL> for i , group in enumerate ( self . _get_groups ( server ) ) : <EOL> name = '<STR_LIT>' % i <EOL> self . assertEqual ( group . get ( '<STR_LIT:name>' ) , name ) <EOL> def test_show ( self ) : <EOL> url = self . base_url + '<STR_LIT:/>' + UUID3 <EOL> res = self . _make_request ( url ) <EOL> self . assertEqual ( res . status_int , <NUM_LIT:200> ) <EOL> server = self . _get_server ( res . body ) <EOL> for i , group in enumerate ( self . _get_groups ( server ) ) : <EOL> name = '<STR_LIT>' % i <EOL> self . assertEqual ( group . get ( '<STR_LIT:name>' ) , name ) <EOL> def test_detail ( self ) : <EOL> url = self . base_url + '<STR_LIT>' <EOL> res = self . _make_request ( url ) <EOL> self . assertEqual ( res . status_int , <NUM_LIT:200> ) <EOL> for i , server in enumerate ( self . _get_servers ( res . body ) ) : <EOL> for j , group in enumerate ( self . _get_groups ( server ) ) : <EOL> name = '<STR_LIT>' % ( i , j ) <EOL> self . assertEqual ( group . get ( '<STR_LIT:name>' ) , name ) <EOL> def test_no_instance_passthrough_404 ( self ) : <EOL> def fake_compute_get ( * args , ** kwargs ) : <EOL> raise exception . InstanceNotFound ( instance_id = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute . api . API , '<STR_LIT>' , fake_compute_get ) <EOL> url = self . base_url + '<STR_LIT>' <EOL> res = self . _make_request ( url ) <EOL> self . assertEqual ( res . status_int , <NUM_LIT> ) <EOL> class SecurityGroupsOutputTestV2 ( SecurityGroupsOutputTestV21 ) : <EOL> def _setup_app ( self ) : <EOL> return fakes . wsgi_app ( init_only = ( '<STR_LIT>' , ) ) <EOL> class SecurityGroupsOutputPolicyEnforcementV21 ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SecurityGroupsOutputPolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . controller = secgroups_v21 . SecurityGroupsOutputController ( ) <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> self . rule_name = "<STR_LIT>" <EOL> self . rule = { self . rule_name : "<STR_LIT>" } <EOL> self . policy . set_rules ( self . rule ) <EOL> self . fake_res = wsgi . ResponseObject ( { <EOL> '<STR_LIT>' : { '<STR_LIT:id>' : '<STR_LIT:0>' } , <EOL> '<STR_LIT>' : [ { '<STR_LIT:id>' : '<STR_LIT:0>' } , { '<STR_LIT:id>' : '<STR_LIT:2>' } ] } ) <EOL> @ mock . patch . object ( secgroups_v21 , "<STR_LIT>" ) <EOL> def test_show_policy_softauth_is_called ( self , mock_softauth ) : <EOL> mock_softauth . return_value = False <EOL> self . controller . show ( self . req , self . fake_res , FAKE_UUID1 ) <EOL> self . assertTrue ( mock_softauth . called ) <EOL> @ mock . patch . object ( nova . network . security_group . openstack_driver , <EOL> "<STR_LIT>" ) <EOL> def test_show_policy_failed ( self , is_neutron_security_groups ) : <EOL> self . controller . show ( self . req , self . fake_res , FAKE_UUID1 ) <EOL> self . assertFalse ( is_neutron_security_groups . called ) <EOL> @ mock . patch . object ( secgroups_v21 , "<STR_LIT>" ) <EOL> def test_create_policy_softauth_is_called ( self , mock_softauth ) : <EOL> mock_softauth . return_value = False <EOL> self . controller . show ( self . req , self . fake_res , { } ) <EOL> self . assertTrue ( mock_softauth . called ) <EOL> @ mock . patch . object ( nova . network . security_group . openstack_driver , <EOL> "<STR_LIT>" ) <EOL> def test_create_policy_failed ( self , is_neutron_security_groups ) : <EOL> self . controller . create ( self . req , self . fake_res , { } ) <EOL> self . assertFalse ( is_neutron_security_groups . called ) <EOL> @ mock . patch . object ( secgroups_v21 , "<STR_LIT>" ) <EOL> def test_detail_policy_softauth_is_called ( self , mock_softauth ) : <EOL> mock_softauth . return_value = False <EOL> self . controller . detail ( self . req , self . fake_res ) <EOL> self . assertTrue ( mock_softauth . called ) <EOL> @ mock . patch . object ( nova . network . security_group . openstack_driver , <EOL> "<STR_LIT>" ) <EOL> def test_detail_policy_failed ( self , is_neutron_security_groups ) : <EOL> self . controller . detail ( self . req , self . fake_res ) <EOL> self . assertFalse ( is_neutron_security_groups . called ) <EOL> class PolicyEnforcementV21 ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( PolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> self . rule_name = "<STR_LIT>" <EOL> self . rule = { self . rule_name : "<STR_LIT>" } <EOL> def _common_policy_check ( self , func , * arg , ** kwarg ) : <EOL> self . policy . set_rules ( self . rule ) <EOL> exc = self . assertRaises ( <EOL> exception . PolicyNotAuthorized , func , * arg , ** kwarg ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" % self . rule_name , <EOL> exc . format_message ( ) ) <EOL> class SecurityGroupPolicyEnforcementV21 ( PolicyEnforcementV21 ) : <EOL> def setUp ( self ) : <EOL> super ( SecurityGroupPolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . controller = secgroups_v21 . SecurityGroupController ( ) <EOL> def test_create_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . create , self . req , { } ) <EOL> def test_show_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . show , self . req , FAKE_UUID1 ) <EOL> def test_delete_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . delete , self . req , FAKE_UUID1 ) <EOL> def test_index_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . index , self . req ) <EOL> def test_update_policy_failed ( self ) : <EOL> self . _common_policy_check ( <EOL> self . controller . update , self . req , FAKE_UUID1 , { } ) <EOL> class ServerSecurityGroupPolicyEnforcementV21 ( PolicyEnforcementV21 ) : <EOL> def setUp ( self ) : <EOL> super ( ServerSecurityGroupPolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . controller = secgroups_v21 . ServerSecurityGroupController ( ) <EOL> def test_index_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . index , self . req , FAKE_UUID1 ) <EOL> class SecurityGroupRulesPolicyEnforcementV21 ( PolicyEnforcementV21 ) : <EOL> def setUp ( self ) : <EOL> super ( SecurityGroupRulesPolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . controller = secgroups_v21 . SecurityGroupRulesController ( ) <EOL> def test_create_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . create , self . req , { } ) <EOL> def test_delete_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . delete , self . req , FAKE_UUID1 ) <EOL> class SecurityGroupActionPolicyEnforcementV21 ( PolicyEnforcementV21 ) : <EOL> def setUp ( self ) : <EOL> super ( SecurityGroupActionPolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . controller = secgroups_v21 . SecurityGroupActionController ( ) <EOL> def test_add_security_group_policy_failed ( self ) : <EOL> self . _common_policy_check ( <EOL> self . controller . _addSecurityGroup , self . req , FAKE_UUID1 , { } ) <EOL> def test_remove_security_group_policy_failed ( self ) : <EOL> self . _common_policy_check ( <EOL> self . controller . _removeSecurityGroup , self . req , FAKE_UUID1 , { } ) </s>
