<s> import base64 <EOL> import datetime <EOL> import json <EOL> import unittest <EOL> from xml . dom import minidom <EOL> import stubout <EOL> import webob <EOL> from nova import db <EOL> from nova import exception <EOL> from nova import flags <EOL> from nova import test <EOL> import nova . api . openstack <EOL> from nova . tests . api . openstack import fakes <EOL> FLAGS = flags . FLAGS <EOL> FLAGS . verbose = True <EOL> FAKE_UUID = '<STR_LIT>' <EOL> FAKE_NETWORKS = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> DUPLICATE_NETWORKS = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> INVALID_NETWORKS = [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> INSTANCE = { <EOL> "<STR_LIT:id>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : FAKE_UUID , <EOL> "<STR_LIT>" : datetime . datetime ( <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT:10> , <NUM_LIT:12> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" : datetime . datetime ( <NUM_LIT> , <NUM_LIT:11> , <NUM_LIT:11> , <NUM_LIT:11> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" : [ { "<STR_LIT:id>" : <NUM_LIT:1> , "<STR_LIT:name>" : "<STR_LIT:test>" } ] <EOL> } <EOL> def return_server_by_id ( context , id , session = None ) : <EOL> INSTANCE [ '<STR_LIT:id>' ] = id <EOL> return INSTANCE <EOL> def return_security_group_non_existing ( context , project_id , group_name ) : <EOL> raise exception . SecurityGroupNotFoundForProject ( project_id = project_id , <EOL> security_group_id = group_name ) <EOL> def return_security_group_get_by_name ( context , project_id , group_name ) : <EOL> return { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : group_name } <EOL> def return_security_group_get ( context , security_group_id , session ) : <EOL> return { '<STR_LIT:id>' : security_group_id } <EOL> def return_instance_add_security_group ( context , instance_id , <EOL> security_group_id ) : <EOL> pass <EOL> class CreateserverextTest ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( CreateserverextTest , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> super ( CreateserverextTest , self ) . tearDown ( ) <EOL> def _setup_mock_compute_api ( self ) : <EOL> class MockComputeAPI ( nova . compute . API ) : <EOL> def __init__ ( self ) : <EOL> self . injected_files = None <EOL> self . networks = None <EOL> self . user_data = None <EOL> self . db = db <EOL> def create ( self , * args , ** kwargs ) : <EOL> if '<STR_LIT>' in kwargs : <EOL> self . injected_files = kwargs [ '<STR_LIT>' ] <EOL> else : <EOL> self . injected_files = None <EOL> if '<STR_LIT>' in kwargs : <EOL> self . networks = kwargs [ '<STR_LIT>' ] <EOL> else : <EOL> self . networks = None <EOL> if '<STR_LIT>' in kwargs : <EOL> self . user_data = kwargs [ '<STR_LIT>' ] <EOL> return [ { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : FAKE_UUID , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" } ] <EOL> def set_admin_password ( self , * args , ** kwargs ) : <EOL> pass <EOL> def make_stub_method ( canned_return ) : <EOL> def stub_method ( * args , ** kwargs ) : <EOL> return canned_return <EOL> return stub_method <EOL> compute_api = MockComputeAPI ( ) <EOL> self . stubs . Set ( nova . compute , '<STR_LIT>' , make_stub_method ( compute_api ) ) <EOL> self . stubs . Set ( <EOL> nova . api . openstack . create_instance_helper . CreateInstanceHelper , <EOL> '<STR_LIT>' , make_stub_method ( ( <NUM_LIT:1> , <NUM_LIT:1> ) ) ) <EOL> return compute_api <EOL> def _create_security_group_request_dict ( self , security_groups ) : <EOL> server = { } <EOL> server [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> server [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> server [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> if security_groups is not None : <EOL> sg_list = [ ] <EOL> for name in security_groups : <EOL> sg_list . append ( { '<STR_LIT:name>' : name } ) <EOL> server [ '<STR_LIT>' ] = sg_list <EOL> return { '<STR_LIT>' : server } <EOL> def _create_networks_request_dict ( self , networks ) : <EOL> server = { } <EOL> server [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> server [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> server [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> if networks is not None : <EOL> network_list = [ ] <EOL> for uuid , fixed_ip in networks : <EOL> network_list . append ( { '<STR_LIT>' : uuid , '<STR_LIT>' : fixed_ip } ) <EOL> server [ '<STR_LIT>' ] = network_list <EOL> return { '<STR_LIT>' : server } <EOL> def _create_user_data_request_dict ( self , user_data ) : <EOL> server = { } <EOL> server [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> server [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> server [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> server [ '<STR_LIT>' ] = user_data <EOL> return { '<STR_LIT>' : server } <EOL> def _get_create_request_json ( self , body_dict ) : <EOL> req = webob . Request . blank ( '<STR_LIT>' ) <EOL> req . headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT:application/json>' <EOL> req . method = '<STR_LIT:POST>' <EOL> req . body = json . dumps ( body_dict ) <EOL> return req <EOL> def _run_create_instance_with_mock_compute_api ( self , request ) : <EOL> compute_api = self . _setup_mock_compute_api ( ) <EOL> response = request . get_response ( fakes . wsgi_app ( ) ) <EOL> return compute_api , response <EOL> def _format_xml_request_body ( self , body_dict ) : <EOL> server = body_dict [ '<STR_LIT>' ] <EOL> body_parts = [ ] <EOL> body_parts . extend ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' % ( <EOL> server [ '<STR_LIT:name>' ] , server [ '<STR_LIT>' ] , server [ '<STR_LIT>' ] ) ] ) <EOL> if '<STR_LIT>' in server : <EOL> metadata = server [ '<STR_LIT>' ] <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> for item in metadata . iteritems ( ) : <EOL> body_parts . append ( '<STR_LIT>' % item ) <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in server : <EOL> personalities = server [ '<STR_LIT>' ] <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> for file in personalities : <EOL> item = ( file [ '<STR_LIT:path>' ] , file [ '<STR_LIT>' ] ) <EOL> body_parts . append ( '<STR_LIT>' % item ) <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in server : <EOL> networks = server [ '<STR_LIT>' ] <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> for network in networks : <EOL> item = ( network [ '<STR_LIT>' ] , network [ '<STR_LIT>' ] ) <EOL> body_parts . append ( '<STR_LIT>' <EOL> % item ) <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> body_parts . append ( '<STR_LIT>' ) <EOL> return '<STR_LIT>' . join ( body_parts ) <EOL> def _get_create_request_xml ( self , body_dict ) : <EOL> req = webob . Request . blank ( '<STR_LIT>' ) <EOL> req . content_type = '<STR_LIT>' <EOL> req . accept = '<STR_LIT>' <EOL> req . method = '<STR_LIT:POST>' <EOL> req . body = self . _format_xml_request_body ( body_dict ) <EOL> return req <EOL> def _create_instance_with_networks_json ( self , networks ) : <EOL> body_dict = self . _create_networks_request_dict ( networks ) <EOL> request = self . _get_create_request_json ( body_dict ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> return request , response , compute_api . networks <EOL> def _create_instance_with_user_data_json ( self , networks ) : <EOL> body_dict = self . _create_user_data_request_dict ( networks ) <EOL> request = self . _get_create_request_json ( body_dict ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> return request , response , compute_api . user_data <EOL> def _create_instance_with_networks_xml ( self , networks ) : <EOL> body_dict = self . _create_networks_request_dict ( networks ) <EOL> request = self . _get_create_request_xml ( body_dict ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> return request , response , compute_api . networks <EOL> def test_create_instance_with_no_networks ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_json ( networks = None ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_no_networks_xml ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_xml ( networks = None ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_one_network ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_json ( [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> def test_create_instance_with_one_network_xml ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_xml ( [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> def test_create_instance_with_two_networks ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_json ( FAKE_NETWORKS ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , FAKE_NETWORKS ) <EOL> def test_create_instance_with_two_networks_xml ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_xml ( FAKE_NETWORKS ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , FAKE_NETWORKS ) <EOL> def test_create_instance_with_duplicate_networks ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_json ( DUPLICATE_NETWORKS ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_duplicate_networks_xml ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_xml ( DUPLICATE_NETWORKS ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_network_no_id ( self ) : <EOL> body_dict = self . _create_networks_request_dict ( [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> del body_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> request = self . _get_create_request_json ( body_dict ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( compute_api . networks , None ) <EOL> def test_create_instance_with_network_no_id_xml ( self ) : <EOL> body_dict = self . _create_networks_request_dict ( [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> request = self . _get_create_request_xml ( body_dict ) <EOL> uuid = '<STR_LIT>' <EOL> request . body = request . body . replace ( uuid , '<STR_LIT>' ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( compute_api . networks , None ) <EOL> def test_create_instance_with_network_invalid_id ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_json ( INVALID_NETWORKS ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_network_invalid_id_xml ( self ) : <EOL> request , response , networks = self . _create_instance_with_networks_xml ( INVALID_NETWORKS ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_network_empty_fixed_ip ( self ) : <EOL> networks = [ ( '<STR_LIT:1>' , '<STR_LIT>' ) ] <EOL> request , response , networks = self . _create_instance_with_networks_json ( networks ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_network_non_string_fixed_ip ( self ) : <EOL> networks = [ ( '<STR_LIT:1>' , <NUM_LIT> ) ] <EOL> request , response , networks = self . _create_instance_with_networks_json ( networks ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_network_empty_fixed_ip_xml ( self ) : <EOL> networks = [ ( '<STR_LIT:1>' , '<STR_LIT>' ) ] <EOL> request , response , networks = self . _create_instance_with_networks_xml ( networks ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( networks , None ) <EOL> def test_create_instance_with_network_no_fixed_ip ( self ) : <EOL> body_dict = self . _create_networks_request_dict ( [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> del body_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> request = self . _get_create_request_json ( body_dict ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( compute_api . networks , <EOL> [ ( '<STR_LIT>' , None ) ] ) <EOL> def test_create_instance_with_network_no_fixed_ip_xml ( self ) : <EOL> body_dict = self . _create_networks_request_dict ( [ FAKE_NETWORKS [ <NUM_LIT:0> ] ] ) <EOL> request = self . _get_create_request_xml ( body_dict ) <EOL> request . body = request . body . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> compute_api , response = self . _run_create_instance_with_mock_compute_api ( request ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( compute_api . networks , <EOL> [ ( '<STR_LIT>' , None ) ] ) <EOL> def test_create_instance_with_userdata ( self ) : <EOL> user_data_contents = '<STR_LIT>' <EOL> user_data_contents = base64 . b64encode ( user_data_contents ) <EOL> request , response , user_data = self . _create_instance_with_user_data_json ( user_data_contents ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( user_data , user_data_contents ) <EOL> def test_create_instance_with_userdata_none ( self ) : <EOL> user_data_contents = None <EOL> request , response , user_data = self . _create_instance_with_user_data_json ( user_data_contents ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( user_data , user_data_contents ) <EOL> def test_create_instance_with_userdata_with_non_b64_content ( self ) : <EOL> user_data_contents = '<STR_LIT>' <EOL> request , response , user_data = self . _create_instance_with_user_data_json ( user_data_contents ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> self . assertEquals ( user_data , None ) <EOL> def test_create_instance_with_security_group_json ( self ) : <EOL> security_groups = [ '<STR_LIT:test>' , '<STR_LIT>' ] <EOL> self . stubs . Set ( nova . db . api , '<STR_LIT>' , <EOL> return_security_group_get_by_name ) <EOL> self . stubs . Set ( nova . db . api , '<STR_LIT>' , <EOL> return_instance_add_security_group ) <EOL> body_dict = self . _create_security_group_request_dict ( security_groups ) <EOL> request = self . _get_create_request_json ( body_dict ) <EOL> response = request . get_response ( fakes . wsgi_app ( ) ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT> ) <EOL> def test_get_server_by_id_verify_security_groups_json ( self ) : <EOL> self . stubs . Set ( nova . db . api , '<STR_LIT>' , return_server_by_id ) <EOL> req = webob . Request . blank ( '<STR_LIT>' ) <EOL> req . headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT:application/json>' <EOL> response = req . get_response ( fakes . wsgi_app ( ) ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT:200> ) <EOL> res_dict = json . loads ( response . body ) <EOL> expected_security_group = [ { "<STR_LIT:name>" : "<STR_LIT:test>" } ] <EOL> self . assertEquals ( res_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> expected_security_group ) <EOL> def test_get_server_by_id_verify_security_groups_xml ( self ) : <EOL> self . stubs . Set ( nova . db . api , '<STR_LIT>' , return_server_by_id ) <EOL> req = webob . Request . blank ( '<STR_LIT>' ) <EOL> req . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response = req . get_response ( fakes . wsgi_app ( ) ) <EOL> self . assertEquals ( response . status_int , <NUM_LIT:200> ) <EOL> dom = minidom . parseString ( response . body ) <EOL> server = dom . childNodes [ <NUM_LIT:0> ] <EOL> sec_groups = server . getElementsByTagName ( '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> sec_group = sec_groups . getElementsByTagName ( '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( INSTANCE [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] , <EOL> sec_group . getAttribute ( "<STR_LIT:name>" ) ) </s>
