<s> import uuid as uuid_lib <EOL> from oslo_config import cfg <EOL> from oslo_utils import timeutils <EOL> from webob import exc <EOL> from nova . api . openstack . compute import cloudpipe as cloudpipe_v21 <EOL> from nova . api . openstack . compute . legacy_v2 . contrib import cloudpipe as cloudpipe_v2 <EOL> from nova . compute import utils as compute_utils <EOL> from nova import exception <EOL> from nova import objects <EOL> from nova import test <EOL> from nova . tests . unit . api . openstack import fakes <EOL> from nova . tests . unit import fake_network <EOL> from nova . tests . unit import matchers <EOL> from nova import utils <EOL> CONF = cfg . CONF <EOL> CONF . import_opt ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> project_id = str ( uuid_lib . uuid4 ( ) . hex ) <EOL> uuid = str ( uuid_lib . uuid4 ( ) ) <EOL> def fake_vpn_instance ( ) : <EOL> return objects . Instance ( <EOL> id = <NUM_LIT:7> , image_ref = CONF . vpn_image_id , vm_state = '<STR_LIT>' , <EOL> created_at = timeutils . parse_strtime ( '<STR_LIT>' ) , <EOL> uuid = uuid , project_id = project_id ) <EOL> def compute_api_get_all_empty ( context , search_opts = None , want_objects = True ) : <EOL> return [ ] <EOL> def compute_api_get_all ( context , search_opts = None , want_objects = True ) : <EOL> return [ fake_vpn_instance ( ) ] <EOL> def utils_vpn_ping ( addr , port , timoeout = <NUM_LIT> , session_id = None ) : <EOL> return True <EOL> class CloudpipeTestV21 ( test . NoDBTestCase ) : <EOL> cloudpipe = cloudpipe_v21 <EOL> url = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> super ( CloudpipeTestV21 , self ) . setUp ( ) <EOL> self . controller = self . cloudpipe . CloudpipeController ( ) <EOL> self . stubs . Set ( self . controller . compute_api , "<STR_LIT>" , <EOL> compute_api_get_all_empty ) <EOL> self . stubs . Set ( utils , '<STR_LIT>' , utils_vpn_ping ) <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> def test_cloudpipe_list_no_network ( self ) : <EOL> def fake_get_nw_info_for_instance ( instance ) : <EOL> return { } <EOL> self . stubs . Set ( compute_utils , "<STR_LIT>" , <EOL> fake_get_nw_info_for_instance ) <EOL> self . stubs . Set ( self . controller . compute_api , "<STR_LIT>" , <EOL> compute_api_get_all ) <EOL> res_dict = self . controller . index ( self . req ) <EOL> response = { '<STR_LIT>' : [ { '<STR_LIT>' : project_id , <EOL> '<STR_LIT>' : uuid , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } <EOL> self . assertEqual ( res_dict , response ) <EOL> def test_cloudpipe_list ( self ) : <EOL> def network_api_get ( context , network_id ) : <EOL> self . assertEqual ( context . project_id , project_id ) <EOL> return { '<STR_LIT>' : '<STR_LIT:127.0.0.1>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> def fake_get_nw_info_for_instance ( instance ) : <EOL> return fake_network . fake_get_instance_nw_info ( self ) <EOL> self . stubs . Set ( compute_utils , "<STR_LIT>" , <EOL> fake_get_nw_info_for_instance ) <EOL> self . stubs . Set ( self . controller . network_api , "<STR_LIT>" , <EOL> network_api_get ) <EOL> self . stubs . Set ( self . controller . compute_api , "<STR_LIT>" , <EOL> compute_api_get_all ) <EOL> res_dict = self . controller . index ( self . req ) <EOL> response = { '<STR_LIT>' : [ { '<STR_LIT>' : project_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:127.0.0.1>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:state>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : uuid , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } <EOL> self . assertThat ( res_dict , matchers . DictMatches ( response ) ) <EOL> def test_cloudpipe_create ( self ) : <EOL> def launch_vpn_instance ( context ) : <EOL> return ( [ fake_vpn_instance ( ) ] , '<STR_LIT>' ) <EOL> self . stubs . Set ( self . controller . cloudpipe , '<STR_LIT>' , <EOL> launch_vpn_instance ) <EOL> body = { '<STR_LIT>' : { '<STR_LIT>' : project_id } } <EOL> res_dict = self . controller . create ( self . req , body = body ) <EOL> response = { '<STR_LIT>' : uuid } <EOL> self . assertEqual ( res_dict , response ) <EOL> def test_cloudpipe_create_no_networks ( self ) : <EOL> def launch_vpn_instance ( context ) : <EOL> raise exception . NoMoreNetworks <EOL> self . stubs . Set ( self . controller . cloudpipe , '<STR_LIT>' , <EOL> launch_vpn_instance ) <EOL> body = { '<STR_LIT>' : { '<STR_LIT>' : project_id } } <EOL> req = fakes . HTTPRequest . blank ( self . url ) <EOL> self . assertRaises ( exc . HTTPBadRequest , <EOL> self . controller . create , req , body = body ) <EOL> def test_cloudpipe_create_already_running ( self ) : <EOL> def launch_vpn_instance ( * args , ** kwargs ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . stubs . Set ( self . controller . cloudpipe , '<STR_LIT>' , <EOL> launch_vpn_instance ) <EOL> self . stubs . Set ( self . controller . compute_api , "<STR_LIT>" , <EOL> compute_api_get_all ) <EOL> body = { '<STR_LIT>' : { '<STR_LIT>' : project_id } } <EOL> req = fakes . HTTPRequest . blank ( self . url ) <EOL> res_dict = self . controller . create ( req , body = body ) <EOL> response = { '<STR_LIT>' : uuid } <EOL> self . assertEqual ( res_dict , response ) <EOL> def test_cloudpipe_create_with_bad_project_id_failed ( self ) : <EOL> body = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> req = fakes . HTTPRequest . blank ( self . url ) <EOL> self . assertRaises ( exception . ValidationError , <EOL> self . controller . create , req , body = body ) <EOL> class CloudpipeTestV2 ( CloudpipeTestV21 ) : <EOL> cloudpipe = cloudpipe_v2 <EOL> def test_cloudpipe_create_with_bad_project_id_failed ( self ) : <EOL> pass <EOL> class CloudpipePolicyEnforcementV21 ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( CloudpipePolicyEnforcementV21 , self ) . setUp ( ) <EOL> self . controller = cloudpipe_v21 . CloudpipeController ( ) <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> def _common_policy_check ( self , func , * arg , ** kwarg ) : <EOL> rule_name = "<STR_LIT>" <EOL> rule = { rule_name : "<STR_LIT>" } <EOL> self . policy . set_rules ( rule ) <EOL> exc = self . assertRaises ( <EOL> exception . PolicyNotAuthorized , func , * arg , ** kwarg ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" % rule_name , <EOL> exc . format_message ( ) ) <EOL> def test_list_policy_failed ( self ) : <EOL> self . _common_policy_check ( self . controller . index , self . req ) <EOL> def test_create_policy_failed ( self ) : <EOL> body = { '<STR_LIT>' : { '<STR_LIT>' : uuid } } <EOL> self . _common_policy_check ( self . controller . create , self . req , body = body ) <EOL> def test_update_policy_failed ( self ) : <EOL> body = { "<STR_LIT>" : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } } <EOL> self . _common_policy_check ( <EOL> self . controller . update , self . req , uuid , body = body ) </s>
