<s> from mox import IsA <EOL> from oslo . config import cfg <EOL> from rack . api . v1 . views . processes import ViewBuilder <EOL> from rack import context <EOL> from rack import db <EOL> from rack import exception <EOL> from rack . openstack . common import jsonutils <EOL> from rack . openstack . common import log as logging <EOL> from rack . resourceoperator import manager <EOL> from rack import test <EOL> from rack . tests . api import fakes <EOL> import json <EOL> import uuid <EOL> import webob <EOL> LOG = logging . getLogger ( __name__ ) <EOL> GID = unicode ( uuid . uuid4 ( ) ) <EOL> PPID1 = unicode ( uuid . uuid4 ( ) ) <EOL> PPID2 = unicode ( uuid . uuid4 ( ) ) <EOL> PID1 = unicode ( uuid . uuid4 ( ) ) <EOL> PID2 = unicode ( uuid . uuid4 ( ) ) <EOL> PIDX = unicode ( uuid . uuid4 ( ) ) <EOL> KEYPAIR_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> KEYPAIR_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> NOVA_KEYPAIR_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> SECURITYGROUP_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> SECURITYGROUP_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> SECURITYGROUP_ID3 = unicode ( uuid . uuid4 ( ) ) <EOL> NEUTRON_SECURITYGROUP_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> NEUTRON_SECURITYGROUP_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> NEUTRON_SECURITYGROUP_ID3 = unicode ( uuid . uuid4 ( ) ) <EOL> NETWORK_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> NETWORK_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> NEUTRON_NETWORK_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> NEUTRON_NETWORK_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> GLANCE_IMAGE_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> GLANCE_IMAGE_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> METADATA1 = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> USER_DATA_B64_ENC = "<STR_LIT>" <EOL> USER_DATA_B64_DEC = "<STR_LIT>" <EOL> NOVA_INSTANCE_ID1 = unicode ( uuid . uuid4 ( ) ) <EOL> NOVA_INSTANCE_ID2 = unicode ( uuid . uuid4 ( ) ) <EOL> NOVA_INSTANCE_ID3 = unicode ( uuid . uuid4 ( ) ) <EOL> def _base ( context ) : <EOL> return { <EOL> "<STR_LIT>" : context . user_id , <EOL> "<STR_LIT>" : context . project_id <EOL> } <EOL> def _base_keypair ( keypair_id , nova_keypair_id ) : <EOL> return { <EOL> "<STR_LIT>" : keypair_id , <EOL> "<STR_LIT>" : nova_keypair_id <EOL> } <EOL> def _base_securitygroup ( securitygroup_id , neutron_securitygroup_id ) : <EOL> return { <EOL> "<STR_LIT>" : securitygroup_id , <EOL> "<STR_LIT>" : neutron_securitygroup_id <EOL> } <EOL> def _base_securitygroups1 ( ) : <EOL> return [ <EOL> _base_securitygroup ( SECURITYGROUP_ID1 , NEUTRON_SECURITYGROUP_ID1 ) , <EOL> _base_securitygroup ( SECURITYGROUP_ID2 , NEUTRON_SECURITYGROUP_ID2 ) , <EOL> ] <EOL> def _base_securitygroups2 ( ) : <EOL> return [ <EOL> _base_securitygroup ( SECURITYGROUP_ID3 , NEUTRON_SECURITYGROUP_ID3 ) , <EOL> ] <EOL> def _base_network ( network_id , neutron_network_id ) : <EOL> return { <EOL> "<STR_LIT>" : network_id , <EOL> "<STR_LIT>" : neutron_network_id <EOL> } <EOL> def _base_networks ( ) : <EOL> return [ <EOL> _base_network ( NETWORK_ID1 , NEUTRON_NETWORK_ID1 ) , <EOL> _base_network ( NETWORK_ID2 , NEUTRON_NETWORK_ID2 ) , <EOL> ] <EOL> def _base_process1 ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : PPID1 , <EOL> "<STR_LIT>" : NOVA_INSTANCE_ID1 , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : GLANCE_IMAGE_ID1 , <EOL> "<STR_LIT>" : KEYPAIR_ID1 , <EOL> "<STR_LIT>" : _base_securitygroups1 ( ) , <EOL> "<STR_LIT>" : _base_networks ( ) , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT:status>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : USER_DATA_B64_ENC , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : '<STR_LIT>' + gid + '<STR_LIT>' + pid + '<STR_LIT>' } <EOL> def _base_process2 ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : PPID2 , <EOL> "<STR_LIT>" : NOVA_INSTANCE_ID2 , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : GLANCE_IMAGE_ID2 , <EOL> "<STR_LIT>" : KEYPAIR_ID2 , <EOL> "<STR_LIT>" : _base_securitygroups2 ( ) , <EOL> "<STR_LIT>" : _base_networks ( ) , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT:status>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : USER_DATA_B64_ENC , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : '<STR_LIT>' } <EOL> def _base_process3 ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : PPID1 , <EOL> "<STR_LIT>" : NOVA_INSTANCE_ID3 , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : GLANCE_IMAGE_ID1 , <EOL> "<STR_LIT>" : KEYPAIR_ID1 , <EOL> "<STR_LIT>" : _base_securitygroups1 ( ) , <EOL> "<STR_LIT>" : _base_networks ( ) , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT:status>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : USER_DATA_B64_ENC , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : '<STR_LIT>' } <EOL> def _base_processes ( gid ) : <EOL> return [ <EOL> _base_process1 ( gid , PPID1 ) , <EOL> _base_process2 ( gid , PPID2 ) , <EOL> _base_process1 ( gid , PID1 ) , <EOL> _base_process2 ( gid , PID2 ) , <EOL> ] <EOL> def fake_keypair_get_by_keypair_id ( context , gid , keypair_id ) : <EOL> return _base_keypair ( keypair_id , NOVA_KEYPAIR_ID1 ) <EOL> def fake_keypair_get_by_keypair_id_raise_not_found ( context , gid , keypair_id ) : <EOL> raise exception . KeypairNotFound ( keypair_id = keypair_id ) <EOL> def fake_network_get_all ( context , gid , filters = None ) : <EOL> return _base_networks ( ) <EOL> def fake_network_get_all_not_found ( context , gid , filters = None ) : <EOL> return [ ] <EOL> def fake_process_get_all ( context , gid , filters = None ) : <EOL> processes = _base_processes ( gid ) <EOL> for process in processes : <EOL> process . update ( _base ( context ) ) <EOL> return processes <EOL> def fake_process_get_all_for_proxy ( context , gid , filters = None ) : <EOL> process = _base_process3 ( gid , PID1 ) <EOL> return [ process ] <EOL> def fake_process_get_by_pid ( context , gid , pid ) : <EOL> processes = _base_processes ( gid ) <EOL> for process in processes : <EOL> if process [ "<STR_LIT>" ] == pid : <EOL> process . update ( _base ( context ) ) <EOL> return process <EOL> raise exception . ProcessNotFound ( pid = pid ) <EOL> def fake_pid1 ( ) : <EOL> return PID1 <EOL> def fake_create ( context , kwargs , network_ids , securitygroup_ids ) : <EOL> process = _base ( context ) <EOL> process . update ( kwargs ) <EOL> process [ "<STR_LIT>" ] = fake_network_get_all ( context , GID ) <EOL> process [ "<STR_LIT>" ] = _base_securitygroups1 ( ) <EOL> return process <EOL> def fake_delete ( context , gid , pid ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( nova_instance_id = NOVA_INSTANCE_ID1 ) <EOL> return process <EOL> def get_request ( url , method , body = None ) : <EOL> req = webob . Request . blank ( url ) <EOL> req . headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT:application/json>' <EOL> req . method = method <EOL> if body is not None : <EOL> req . body = jsonutils . dumps ( body ) <EOL> return req <EOL> def get_base_url ( gid ) : <EOL> return "<STR_LIT>" + gid + "<STR_LIT>" <EOL> def get_base_body ( process ) : <EOL> return { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ securitygroup [ "<STR_LIT>" ] <EOL> for securitygroup in process [ "<STR_LIT>" ] ] , <EOL> "<STR_LIT>" : METADATA1 , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] <EOL> } <EOL> def get_base_request_body1 ( process ) : <EOL> return { "<STR_LIT>" : get_base_body ( process ) } <EOL> def get_proxy_request_body1 ( process ) : <EOL> body = get_base_body ( process ) <EOL> body . update ( ipc_endpoint = "<STR_LIT>" ) <EOL> body . update ( shm_endpoint = "<STR_LIT>" ) <EOL> body . update ( fs_endpoint = "<STR_LIT>" ) <EOL> return { "<STR_LIT>" : body } <EOL> def get_base_process_body ( process ) : <EOL> process_body = get_base_body ( process ) <EOL> process_body . update ( gid = GID ) <EOL> process_body . update ( pid = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( status = process [ "<STR_LIT:status>" ] ) <EOL> process_body . update ( networks = [ <EOL> { "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : NETWORK_ID1 } , <EOL> { "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : NETWORK_ID2 } ] ) <EOL> process_body . update ( app_status = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( userdata = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( args = json . loads ( process [ "<STR_LIT:args>" ] ) ) <EOL> process_body . pop ( "<STR_LIT>" ) <EOL> return process_body <EOL> def get_base_prxoy_body ( process ) : <EOL> process_body = get_base_body ( process ) <EOL> process_body . update ( gid = GID ) <EOL> process_body . update ( pid = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( status = process [ "<STR_LIT:status>" ] ) <EOL> process_body . update ( networks = [ <EOL> { "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : NETWORK_ID1 } , <EOL> { "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : NETWORK_ID2 } ] ) <EOL> process_body . update ( app_status = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( userdata = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( args = json . loads ( process [ "<STR_LIT:args>" ] ) ) <EOL> process_body . update ( ipc_endpoint = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( shm_endpoint = process [ "<STR_LIT>" ] ) <EOL> process_body . update ( fs_endpoint = process [ "<STR_LIT>" ] ) <EOL> process_body . pop ( "<STR_LIT>" ) <EOL> return process_body <EOL> def get_base_process_response_body ( process ) : <EOL> process_body = get_base_process_body ( process ) <EOL> return { "<STR_LIT>" : process_body } <EOL> def get_base_processes_response_body ( processes ) : <EOL> processes_body = [ ] <EOL> for process in processes : <EOL> process_body = get_base_process_body ( process ) <EOL> processes_body . append ( process_body ) <EOL> return { "<STR_LIT>" : processes_body } <EOL> def get_proxy_response_body ( process ) : <EOL> process_body = get_base_prxoy_body ( process ) <EOL> return { "<STR_LIT>" : process_body } <EOL> class ProcessesTest ( test . NoDBTestCase ) : <EOL> def _set_mox_db_process_update_on_error ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_update ( IsA ( context . RequestContext ) , IsA ( <EOL> unicode ) , IsA ( unicode ) , { "<STR_LIT:status>" : "<STR_LIT>" } ) <EOL> def setUp ( self ) : <EOL> super ( ProcessesTest , self ) . setUp ( ) <EOL> self . stubs . Set ( uuid , "<STR_LIT>" , fake_pid1 ) <EOL> self . stubs . Set ( <EOL> db , "<STR_LIT>" , fake_keypair_get_by_keypair_id ) <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_network_get_all ) <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_process_get_all ) <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_process_get_by_pid ) <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_create ) <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_delete ) <EOL> self . app = fakes . wsgi_app ( ) <EOL> self . view = ViewBuilder ( ) <EOL> def test_index ( self ) : <EOL> processes = _base_processes ( GID ) <EOL> expect = get_base_processes_response_body ( processes ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_list ( IsA ( context . RequestContext ) , <EOL> IsA ( list ) ) . AndReturn ( processes ) <EOL> self . mox . ReplayAll ( ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> for i in range ( len ( body [ "<STR_LIT>" ] ) ) : <EOL> for key in body [ "<STR_LIT>" ] [ i ] : <EOL> self . assertEqual ( <EOL> expect [ "<STR_LIT>" ] [ i ] [ key ] , body [ "<STR_LIT>" ] [ i ] [ key ] ) <EOL> self . assertEqual ( <NUM_LIT:200> , res . status_code ) <EOL> def test_index_invalid_format_gid ( self ) : <EOL> url = get_base_url ( "<STR_LIT>" ) <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_show ( self ) : <EOL> process = _base_process1 ( GID , PID1 ) <EOL> expect = get_base_process_response_body ( process ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_show ( IsA ( context . RequestContext ) , <EOL> IsA ( dict ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( body , expect ) <EOL> def test_show_invalid_format_gid ( self ) : <EOL> url = get_base_url ( "<STR_LIT>" ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_show_invalid_format_pid ( self ) : <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_show_process_not_found ( self ) : <EOL> self . stubs . Set ( db , "<STR_LIT>" , <EOL> fake_keypair_get_by_keypair_id_raise_not_found ) <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PIDX <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_show_proxy ( self ) : <EOL> process = _base_process3 ( GID , PID1 ) <EOL> expect = get_proxy_response_body ( process ) <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_process_get_all_for_proxy ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_show ( IsA ( context . RequestContext ) , <EOL> IsA ( dict ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( body , expect ) <EOL> def test_show_proxy_not_found_exception ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndRaise ( exception . NotFound ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_show_proxy_bad_request ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> self . mox . ReplayAll ( ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:GET>' ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_proxy ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> cfg . CONF . my_ip = "<STR_LIT>" <EOL> cfg . CONF . os_username = "<STR_LIT>" <EOL> cfg . CONF . os_password = "<STR_LIT>" <EOL> cfg . CONF . os_tenant_name = "<STR_LIT>" <EOL> cfg . CONF . os_auth_url = "<STR_LIT>" <EOL> cfg . CONF . os_region_name = "<STR_LIT>" <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> expect = get_proxy_response_body ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( rackapi_ip = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_username = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_password = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_tenant_name = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_auth_url = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_region_name = "<STR_LIT>" ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_proxy_without_proxy_name ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> cfg . CONF . my_ip = "<STR_LIT>" <EOL> cfg . CONF . os_username = "<STR_LIT>" <EOL> cfg . CONF . os_password = "<STR_LIT>" <EOL> cfg . CONF . os_tenant_name = "<STR_LIT>" <EOL> cfg . CONF . os_auth_url = "<STR_LIT>" <EOL> cfg . CONF . os_region_name = "<STR_LIT>" <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT:name>" ) <EOL> expect = get_proxy_response_body ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( rackapi_ip = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_username = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_password = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_tenant_name = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_auth_url = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( os_region_name = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] . update ( name = "<STR_LIT>" + PID1 ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_proxy_ipc_endpoint_invalid_max_length ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> cfg . CONF . my_ip = "<STR_LIT>" <EOL> cfg . CONF . os_username = "<STR_LIT>" <EOL> cfg . CONF . os_password = "<STR_LIT>" <EOL> cfg . CONF . os_tenant_name = "<STR_LIT>" <EOL> cfg . CONF . os_auth_url = "<STR_LIT>" <EOL> cfg . CONF . os_region_name = "<STR_LIT>" <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( ipc_endpoint = "<STR_LIT:a>" * ( <NUM_LIT> ) ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_proxy_shm_endpoint_invalid_max_length ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> cfg . CONF . my_ip = "<STR_LIT>" <EOL> cfg . CONF . os_username = "<STR_LIT>" <EOL> cfg . CONF . os_password = "<STR_LIT>" <EOL> cfg . CONF . os_tenant_name = "<STR_LIT>" <EOL> cfg . CONF . os_auth_url = "<STR_LIT>" <EOL> cfg . CONF . os_region_name = "<STR_LIT>" <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( shm_endpoint = "<STR_LIT:a>" * ( <NUM_LIT> ) ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_proxy_fs_endpoint_invalid_max_length ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( cfg . CONF , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> cfg . CONF . my_ip = "<STR_LIT>" <EOL> cfg . CONF . os_username = "<STR_LIT>" <EOL> cfg . CONF . os_password = "<STR_LIT>" <EOL> cfg . CONF . os_tenant_name = "<STR_LIT>" <EOL> cfg . CONF . os_auth_url = "<STR_LIT>" <EOL> cfg . CONF . os_region_name = "<STR_LIT>" <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( fs_endpoint = "<STR_LIT:a>" * ( <NUM_LIT> ) ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_proxy_already_exist ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( fs_endpoint = "<STR_LIT:a>" * ( <NUM_LIT> ) ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_proxy_invalid_dict_key ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> expect = get_base_process_response_body ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_args_value_integer ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( args = { <EOL> "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : <NUM_LIT> } ) <EOL> expect = get_base_process_response_body ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( test_key1 = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( test_key2 = "<STR_LIT>" ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_default_securitygroup ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> db . securitygroup_get_all ( <EOL> IsA ( context . RequestContext ) , GID , <EOL> filters = IsA ( dict ) ) . AndReturn ( [ <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> expect = get_base_process_response_body ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] . update ( ppid = None ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_parent_securitygroup_and_image_and_flavor ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> request_body [ "<STR_LIT>" ] . update ( args = { "<STR_LIT:key>" : "<STR_LIT:value>" } ) <EOL> expect = get_base_process_response_body ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( key = "<STR_LIT:value>" ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_without_keypair_id ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> expect = get_base_process_response_body ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = KEYPAIR_ID1 <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_without_keypair_id_and_ppid ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . keypair_get_all ( IsA ( context . RequestContext ) , GID , <EOL> filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( str ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> expect = get_base_process_response_body ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = KEYPAIR_ID1 <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] . update ( keypair_id = "<STR_LIT>" ) <EOL> expect [ "<STR_LIT>" ] . update ( ppid = None ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_without_process_name ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( <EOL> manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> manager . ResourceOperator . get_process_address ( <EOL> IsA ( context . RequestContext ) , <EOL> IsA ( str ) ) . AndReturn ( "<STR_LIT>" ) <EOL> manager . ResourceOperator . process_create ( <EOL> IsA ( context . RequestContext ) , <EOL> name = IsA ( unicode ) , <EOL> key_name = IsA ( unicode ) , <EOL> security_groups = IsA ( list ) , <EOL> image = IsA ( unicode ) , <EOL> flavor = IsA ( int ) , <EOL> userdata = IsA ( unicode ) , <EOL> meta = IsA ( dict ) , <EOL> nics = IsA ( list ) ) . AndReturn ( ( NOVA_INSTANCE_ID1 , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT:name>" ) <EOL> expect = get_base_process_response_body ( process ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:name>" ] = "<STR_LIT>" + PID1 <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT>" ] = USER_DATA_B64_ENC <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( ppid = PPID1 ) <EOL> expect [ "<STR_LIT>" ] [ "<STR_LIT:args>" ] . update ( proxy_ip = "<STR_LIT>" ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( body [ "<STR_LIT>" ] [ key ] , expect [ "<STR_LIT>" ] [ key ] ) <EOL> def test_create_process_proxy_not_exits ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_process_invalid_dict_key ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_proxy_request_body1 ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_format_gid ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , IsA ( unicode ) , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> gid = "<STR_LIT>" <EOL> process = _base_process1 ( gid , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> url = get_base_url ( gid ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_securitygroup_ids ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , IsA ( unicode ) , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( securitygroup_ids = { "<STR_LIT:key>" : "<STR_LIT:value>" } ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_format_securitygroup_ids ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , IsA ( unicode ) , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( securitygroup_ids = [ "<STR_LIT>" ] ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_args ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( args = [ { "<STR_LIT:key>" : "<STR_LIT:value>" } ] ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_format_keypair_id ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , IsA ( unicode ) , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( keypair_id = "<STR_LIT>" ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_format_ppid ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , IsA ( unicode ) , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( ppid = "<STR_LIT>" ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_process_name_is_whitespace ( self ) : <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( name = "<STR_LIT:U+0020>" ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_default_security_group_not_found ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_all ( IsA ( context . RequestContext ) , GID , <EOL> filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> request_body [ "<STR_LIT>" ] . pop ( "<STR_LIT>" ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_userdata ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> request_body [ "<STR_LIT>" ] . update ( userdata = "<STR_LIT:/>" ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_empty_body ( self ) : <EOL> request_body = { "<STR_LIT>" : { } } <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_no_body ( self ) : <EOL> request_body = { } <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_invalid_format_body ( self ) : <EOL> request_body = [ ] <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_create_notfound_networks ( self ) : <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_network_get_all_not_found ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . securitygroup_get_by_securitygroup_id ( <EOL> IsA ( context . RequestContext ) , GID , IsA ( unicode ) ) . AndReturn ( <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> self . mox . ReplayAll ( ) <EOL> process = _base_process1 ( GID , PID1 ) <EOL> request_body = get_base_request_body1 ( process ) <EOL> url = get_base_url ( GID ) <EOL> req = get_request ( url , '<STR_LIT:POST>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def delete_invalid_format_gid ( self ) : <EOL> url = get_base_url ( "<STR_LIT>" ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , "<STR_LIT>" ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_delete_invalid_format_pid ( self ) : <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + "<STR_LIT>" <EOL> req = get_request ( url , "<STR_LIT>" ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update ( self ) : <EOL> def _fake_update ( context , gid , pid , kwargs ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( kwargs ) <EOL> process . update ( args = '<STR_LIT>' ) <EOL> process . update ( securitygroups = [ { "<STR_LIT>" : None } ] ) <EOL> process . update ( networks = [ { "<STR_LIT>" : None } ] ) <EOL> return process <EOL> def _update_process_mockdata ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:status>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:args>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] } <EOL> def _get_update_response_body ( process ) : <EOL> return { "<STR_LIT>" : { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:status>" : process [ "<STR_LIT:status>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:args>" : json . loads ( process [ "<STR_LIT:args>" ] ) , <EOL> "<STR_LIT>" : [ None ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : None , "<STR_LIT>" : None , "<STR_LIT>" : None } ] } } <EOL> self . stubs . Set ( db , "<STR_LIT>" , _fake_update ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> process = _update_process_mockdata ( GID , PID1 ) <EOL> expect = _get_update_response_body ( process ) <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( expect [ "<STR_LIT>" ] [ key ] , body [ "<STR_LIT>" ] [ key ] ) <EOL> def test_update_invalid_request_body ( self ) : <EOL> request_body = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update_app_status_required ( self ) : <EOL> request_body = { "<STR_LIT>" : { } } <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update_not_found ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_update ( <EOL> IsA ( context . RequestContext ) , GID , PID1 , IsA ( dict ) ) . AndRaise ( exception . ProcessNotFound ( pid = PID1 ) ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update_proxy_all ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : PID1 } ] ) <EOL> def _fake_update ( context , gid , pid , kwargs ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( kwargs ) <EOL> process . update ( args = '<STR_LIT>' ) <EOL> process . update ( is_proxy = True ) <EOL> process . update ( securitygroups = [ { "<STR_LIT>" : None } ] ) <EOL> process . update ( networks = [ { "<STR_LIT>" : None } ] ) <EOL> return process <EOL> def _update_process_mockdata ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:status>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT:args>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] } <EOL> def _get_update_response_body ( process ) : <EOL> return { "<STR_LIT>" : { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:status>" : process [ "<STR_LIT:status>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:args>" : json . loads ( process [ "<STR_LIT:args>" ] ) , <EOL> "<STR_LIT>" : [ None ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : None , "<STR_LIT>" : None , "<STR_LIT>" : None } ] } } <EOL> self . stubs . Set ( db , "<STR_LIT>" , _fake_update ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> process = _update_process_mockdata ( GID , PID1 ) <EOL> expect = _get_update_response_body ( process ) <EOL> url = url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( expect [ "<STR_LIT>" ] [ key ] , body [ "<STR_LIT>" ] [ key ] ) <EOL> def test_update_proxy_ipc_endpoint ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : PID1 } ] ) <EOL> def _fake_update ( context , gid , pid , kwargs ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( kwargs ) <EOL> process . update ( is_proxy = True ) <EOL> process . update ( args = '<STR_LIT>' ) <EOL> process . update ( securitygroups = [ { "<STR_LIT>" : None } ] ) <EOL> process . update ( networks = [ { "<STR_LIT>" : None } ] ) <EOL> return process <EOL> def _update_process_mockdata ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:status>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:args>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] } <EOL> def _get_update_response_body ( process ) : <EOL> return { "<STR_LIT>" : { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:status>" : process [ "<STR_LIT:status>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:args>" : json . loads ( process [ "<STR_LIT:args>" ] ) , <EOL> "<STR_LIT>" : [ None ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : None , "<STR_LIT>" : None , "<STR_LIT>" : None } ] } } <EOL> self . stubs . Set ( db , "<STR_LIT>" , _fake_update ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> process = _update_process_mockdata ( GID , PID1 ) <EOL> expect = _get_update_response_body ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( expect [ "<STR_LIT>" ] [ key ] , body [ "<STR_LIT>" ] [ key ] ) <EOL> def test_update_proxy_shm_endpoint ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : PID1 } ] ) <EOL> def _fake_update ( context , gid , pid , kwargs ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( kwargs ) <EOL> process . update ( is_proxy = True ) <EOL> process . update ( args = '<STR_LIT>' ) <EOL> process . update ( securitygroups = [ { "<STR_LIT>" : None } ] ) <EOL> process . update ( networks = [ { "<STR_LIT>" : None } ] ) <EOL> return process <EOL> def _update_process_mockdata ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:status>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:args>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] } <EOL> def _get_update_response_body ( process ) : <EOL> return { "<STR_LIT>" : { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:status>" : process [ "<STR_LIT:status>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:args>" : json . loads ( process [ "<STR_LIT:args>" ] ) , <EOL> "<STR_LIT>" : [ None ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : None , "<STR_LIT>" : None , "<STR_LIT>" : None } ] } } <EOL> self . stubs . Set ( db , "<STR_LIT>" , _fake_update ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> process = _update_process_mockdata ( GID , PID1 ) <EOL> expect = _get_update_response_body ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( expect [ "<STR_LIT>" ] [ key ] , body [ "<STR_LIT>" ] [ key ] ) <EOL> def test_update_proxy_fs_endpoint ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : PID1 } ] ) <EOL> def _fake_update ( context , gid , pid , kwargs ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( kwargs ) <EOL> process . update ( is_proxy = True ) <EOL> process . update ( args = '<STR_LIT>' ) <EOL> process . update ( securitygroups = [ { "<STR_LIT>" : None } ] ) <EOL> process . update ( networks = [ { "<STR_LIT>" : None } ] ) <EOL> return process <EOL> def _update_process_mockdata ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:status>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:args>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] } <EOL> def _get_update_response_body ( process ) : <EOL> return { "<STR_LIT>" : { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:status>" : process [ "<STR_LIT:status>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:args>" : json . loads ( process [ "<STR_LIT:args>" ] ) , <EOL> "<STR_LIT>" : [ None ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : None , "<STR_LIT>" : None , "<STR_LIT>" : None } ] } } <EOL> self . stubs . Set ( db , "<STR_LIT>" , _fake_update ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> process = _update_process_mockdata ( GID , PID1 ) <EOL> expect = _get_update_response_body ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( expect [ "<STR_LIT>" ] [ key ] , body [ "<STR_LIT>" ] [ key ] ) <EOL> def test_update_proxy_app_status ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : PID1 } ] ) <EOL> def _fake_update ( context , gid , pid , kwargs ) : <EOL> process = _base ( context ) <EOL> process . update ( gid = gid ) <EOL> process . update ( pid = pid ) <EOL> process . update ( kwargs ) <EOL> process . update ( is_proxy = True ) <EOL> process . update ( args = '<STR_LIT>' ) <EOL> process . update ( securitygroups = [ { "<STR_LIT>" : None } ] ) <EOL> process . update ( networks = [ { "<STR_LIT>" : None } ] ) <EOL> return process <EOL> def _update_process_mockdata ( gid , pid ) : <EOL> return { <EOL> "<STR_LIT>" : pid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : gid , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:status>" : None , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:args>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : None } ] } <EOL> def _get_update_response_body ( process ) : <EOL> return { "<STR_LIT>" : { <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:status>" : process [ "<STR_LIT:status>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : process . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : process [ "<STR_LIT>" ] , <EOL> "<STR_LIT:args>" : json . loads ( process [ "<STR_LIT:args>" ] ) , <EOL> "<STR_LIT>" : [ None ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : None , "<STR_LIT>" : None , "<STR_LIT>" : None } ] } } <EOL> self . stubs . Set ( db , "<STR_LIT>" , _fake_update ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> process = _update_process_mockdata ( GID , PID1 ) <EOL> expect = _get_update_response_body ( process ) <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> body = jsonutils . loads ( res . body ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> for key in body [ "<STR_LIT>" ] : <EOL> self . assertEqual ( expect [ "<STR_LIT>" ] [ key ] , body [ "<STR_LIT>" ] [ key ] ) <EOL> def test_update_proxy_does_not_exist ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ ] ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update_proxy_invalid_request_body ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update_prxoy_no_keyword ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { } ] ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { } } <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_update_proxy_not_found ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , filters = IsA ( dict ) ) . AndReturn ( [ { "<STR_LIT>" : PID1 } ] ) <EOL> db . process_update ( <EOL> IsA ( context . RequestContext ) , GID , PID1 , IsA ( dict ) ) . AndRaise ( exception . ProcessNotFound ( pid = PID1 ) ) <EOL> self . mox . ReplayAll ( ) <EOL> request_body = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> url = "<STR_LIT>" + GID + "<STR_LIT>" <EOL> req = get_request ( url , '<STR_LIT>' , request_body ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_delete_non_parent ( self ) : <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_delete ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> db . process_get_by_pid ( <EOL> IsA ( context . RequestContext ) , GID , PID1 ) . AndReturn ( <EOL> { "<STR_LIT>" : PID1 , "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : PID1 } ) . AndReturn ( [ ] ) <EOL> manager . ResourceOperator . process_delete ( <EOL> IsA ( context . RequestContext ) , IsA ( str ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , "<STR_LIT>" ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_delete_parent_child_relation ( self ) : <EOL> self . stubs . Set ( db , "<STR_LIT>" , fake_delete ) <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> self . mox . StubOutWithMock ( db , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( manager . ResourceOperator , "<STR_LIT>" ) <EOL> ppid = unicode ( uuid . uuid4 ( ) ) <EOL> pid_1 = unicode ( uuid . uuid4 ( ) ) <EOL> pid_1_1 = unicode ( uuid . uuid4 ( ) ) <EOL> pid_2 = unicode ( uuid . uuid4 ( ) ) <EOL> pid_2_1 = unicode ( uuid . uuid4 ( ) ) <EOL> pid_2_1_1 = unicode ( uuid . uuid4 ( ) ) <EOL> pid_2_1_2 = unicode ( uuid . uuid4 ( ) ) <EOL> pid_3 = unicode ( uuid . uuid4 ( ) ) <EOL> db . process_get_by_pid ( IsA ( context . RequestContext ) , GID , ppid ) . AndReturn ( <EOL> { "<STR_LIT>" : ppid , "<STR_LIT>" : "<STR_LIT>" } ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : ppid } ) . AndReturn ( <EOL> [ { "<STR_LIT>" : pid_1 , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : pid_2 , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : pid_3 , "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_1 } ) . AndReturn ( <EOL> [ { "<STR_LIT>" : pid_1_1 , "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_1_1 } ) . AndReturn ( [ ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_2 } ) . AndReturn ( <EOL> [ { "<STR_LIT>" : pid_2_1 , "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_2_1 } ) . AndReturn ( <EOL> [ { "<STR_LIT>" : pid_2_1_1 , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : pid_2_1_2 , "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_2_1_1 } ) . AndReturn ( [ ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_2_1_2 } ) . AndReturn ( [ ] ) <EOL> db . process_get_all ( <EOL> IsA ( context . RequestContext ) , GID , { "<STR_LIT>" : pid_3 } ) . AndReturn ( [ ] ) <EOL> for i in range ( <NUM_LIT:8> ) : <EOL> manager . ResourceOperator . process_delete ( <EOL> IsA ( context . RequestContext ) , IsA ( str ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + ppid <EOL> req = get_request ( url , "<STR_LIT>" ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) <EOL> def test_delete_not_found_exception ( self ) : <EOL> self . mox . StubOutWithMock ( db , "<STR_LIT>" ) <EOL> db . process_get_by_pid ( IsA ( context . RequestContext ) , GID , PID1 ) . AndRaise ( exception . NotFound ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = get_base_url ( GID ) + "<STR_LIT:/>" + PID1 <EOL> req = get_request ( url , "<STR_LIT>" ) <EOL> res = req . get_response ( self . app ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT> ) </s>
