<s> """<STR_LIT>""" <EOL> from nova import compute <EOL> from nova . compute import instance_types <EOL> from nova . compute import manager as compute_manager <EOL> from nova . compute import power_state <EOL> from nova . compute import task_states <EOL> from nova . compute import vm_states <EOL> from nova import context <EOL> from nova import db <EOL> from nova . db . sqlalchemy import models <EOL> from nova . db . sqlalchemy import api as sqlalchemy_api <EOL> from nova import exception <EOL> from nova import flags <EOL> import nova . image . fake <EOL> from nova import log as logging <EOL> from nova import rpc <EOL> from nova import test <EOL> from nova import utils <EOL> from nova . notifier import test_notifier <EOL> LOG = logging . getLogger ( '<STR_LIT>' ) <EOL> FLAGS = flags . FLAGS <EOL> flags . DECLARE ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> flags . DECLARE ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class FakeTime ( object ) : <EOL> def __init__ ( self ) : <EOL> self . counter = <NUM_LIT:0> <EOL> def sleep ( self , t ) : <EOL> self . counter += t <EOL> def nop_report_driver_status ( self ) : <EOL> pass <EOL> class ComputeTestCase ( test . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ComputeTestCase , self ) . setUp ( ) <EOL> self . flags ( connection_type = '<STR_LIT>' , <EOL> stub_network = True , <EOL> notification_driver = '<STR_LIT>' , <EOL> network_manager = '<STR_LIT>' ) <EOL> self . compute = utils . import_object ( FLAGS . compute_manager ) <EOL> self . compute_api = compute . API ( ) <EOL> self . user_id = '<STR_LIT>' <EOL> self . project_id = '<STR_LIT>' <EOL> self . context = context . RequestContext ( self . user_id , self . project_id ) <EOL> test_notifier . NOTIFICATIONS = [ ] <EOL> def fake_show ( meh , context , id ) : <EOL> return { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> } } <EOL> self . stubs . Set ( nova . image . fake . _FakeImageService , '<STR_LIT>' , fake_show ) <EOL> def _create_instance ( self , params = None ) : <EOL> """<STR_LIT>""" <EOL> if not params : <EOL> params = { } <EOL> inst = { } <EOL> inst [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> inst [ '<STR_LIT>' ] = self . user_id <EOL> inst [ '<STR_LIT>' ] = self . project_id <EOL> type_id = instance_types . get_instance_type_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> inst [ '<STR_LIT>' ] = type_id <EOL> inst [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> inst . update ( params ) <EOL> return db . instance_create ( self . context , inst ) [ '<STR_LIT:id>' ] <EOL> def _create_instance_type ( self , params = None ) : <EOL> """<STR_LIT>""" <EOL> if not params : <EOL> params = { } <EOL> context = self . context . elevated ( ) <EOL> inst = { } <EOL> inst [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT:1>' <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT:1>' <EOL> inst [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> inst [ '<STR_LIT>' ] = <NUM_LIT:100> <EOL> inst [ '<STR_LIT>' ] = <NUM_LIT:200> <EOL> inst . update ( params ) <EOL> return db . instance_type_create ( context , inst ) [ '<STR_LIT:id>' ] <EOL> def _create_group ( self ) : <EOL> values = { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . user_id , <EOL> '<STR_LIT>' : self . project_id } <EOL> return db . security_group_create ( self . context , values ) <EOL> def _get_dummy_instance ( self ) : <EOL> """<STR_LIT>""" <EOL> vol1 = models . Volume ( ) <EOL> vol1 [ '<STR_LIT:id>' ] = <NUM_LIT:1> <EOL> vol2 = models . Volume ( ) <EOL> vol2 [ '<STR_LIT:id>' ] = <NUM_LIT:2> <EOL> instance_ref = models . Instance ( ) <EOL> instance_ref [ '<STR_LIT:id>' ] = <NUM_LIT:1> <EOL> instance_ref [ '<STR_LIT>' ] = [ vol1 , vol2 ] <EOL> instance_ref [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> instance_ref [ '<STR_LIT:host>' ] = '<STR_LIT>' <EOL> return instance_ref <EOL> def test_create_instance_defaults_display_name ( self ) : <EOL> """<STR_LIT>""" <EOL> cases = [ dict ( ) , dict ( display_name = None ) ] <EOL> for instance in cases : <EOL> ref = self . compute_api . create ( self . context , <EOL> instance_types . get_default_instance_type ( ) , None , ** instance ) <EOL> try : <EOL> self . assertNotEqual ( ref [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , None ) <EOL> finally : <EOL> db . instance_destroy ( self . context , ref [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_instance_associates_security_groups ( self ) : <EOL> """<STR_LIT>""" <EOL> group = self . _create_group ( ) <EOL> ref = self . compute_api . create ( <EOL> self . context , <EOL> instance_type = instance_types . get_default_instance_type ( ) , <EOL> image_href = None , <EOL> security_group = [ '<STR_LIT>' ] ) <EOL> try : <EOL> self . assertEqual ( len ( db . security_group_get_by_instance ( <EOL> self . context , ref [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) ) , <NUM_LIT:1> ) <EOL> group = db . security_group_get ( self . context , group [ '<STR_LIT:id>' ] ) <EOL> self . assert_ ( len ( group . instances ) == <NUM_LIT:1> ) <EOL> finally : <EOL> db . security_group_destroy ( self . context , group [ '<STR_LIT:id>' ] ) <EOL> db . instance_destroy ( self . context , ref [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_instance_with_invalid_security_group_raises ( self ) : <EOL> instance_type = instance_types . get_default_instance_type ( ) <EOL> pre_build_len = len ( db . instance_get_all ( context . get_admin_context ( ) ) ) <EOL> self . assertRaises ( exception . SecurityGroupNotFoundForProject , <EOL> self . compute_api . create , <EOL> self . context , <EOL> instance_type = instance_type , <EOL> image_href = None , <EOL> security_group = [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( pre_build_len , <EOL> len ( db . instance_get_all ( context . get_admin_context ( ) ) ) ) <EOL> def test_create_instance_with_img_ref_associates_config_drive ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( params = { '<STR_LIT>' : '<STR_LIT>' , } ) <EOL> try : <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> instances = db . instance_get_all ( context . get_admin_context ( ) ) <EOL> instance = instances [ <NUM_LIT:0> ] <EOL> self . assertTrue ( instance . config_drive ) <EOL> finally : <EOL> db . instance_destroy ( self . context , instance_id ) <EOL> def test_create_instance_associates_config_drive ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( params = { '<STR_LIT>' : True , } ) <EOL> try : <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> instances = db . instance_get_all ( context . get_admin_context ( ) ) <EOL> instance = instances [ <NUM_LIT:0> ] <EOL> self . assertTrue ( instance . config_drive ) <EOL> finally : <EOL> db . instance_destroy ( self . context , instance_id ) <EOL> def test_default_hostname_generator ( self ) : <EOL> cases = [ ( None , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:hello>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> for display_name , hostname in cases : <EOL> ref = self . compute_api . create ( self . context , <EOL> instance_types . get_default_instance_type ( ) , None , <EOL> display_name = display_name ) <EOL> try : <EOL> self . assertEqual ( ref [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , hostname ) <EOL> finally : <EOL> db . instance_destroy ( self . context , ref [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> def test_destroy_instance_disassociates_security_groups ( self ) : <EOL> """<STR_LIT>""" <EOL> group = self . _create_group ( ) <EOL> ref = self . compute_api . create ( <EOL> self . context , <EOL> instance_type = instance_types . get_default_instance_type ( ) , <EOL> image_href = None , <EOL> security_group = [ '<STR_LIT>' ] ) <EOL> try : <EOL> db . instance_destroy ( self . context , ref [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> group = db . security_group_get ( self . context , group [ '<STR_LIT:id>' ] ) <EOL> self . assert_ ( len ( group . instances ) == <NUM_LIT:0> ) <EOL> finally : <EOL> db . security_group_destroy ( self . context , group [ '<STR_LIT:id>' ] ) <EOL> def test_destroy_security_group_disassociates_instances ( self ) : <EOL> """<STR_LIT>""" <EOL> group = self . _create_group ( ) <EOL> ref = self . compute_api . create ( <EOL> self . context , <EOL> instance_type = instance_types . get_default_instance_type ( ) , <EOL> image_href = None , <EOL> security_group = [ '<STR_LIT>' ] ) <EOL> try : <EOL> db . security_group_destroy ( self . context , group [ '<STR_LIT:id>' ] ) <EOL> group = db . security_group_get ( context . get_admin_context ( <EOL> read_deleted = True ) , group [ '<STR_LIT:id>' ] ) <EOL> self . assert_ ( len ( group . instances ) == <NUM_LIT:0> ) <EOL> finally : <EOL> db . instance_destroy ( self . context , ref [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> def test_run_terminate ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> instances = db . instance_get_all ( context . get_admin_context ( ) ) <EOL> LOG . info ( _ ( "<STR_LIT>" ) , instances ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> instances = db . instance_get_all ( context . get_admin_context ( ) ) <EOL> LOG . info ( _ ( "<STR_LIT>" ) , instances ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> def test_run_terminate_timestamps ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> instance_ref = db . instance_get ( self . context , instance_id ) <EOL> self . assertEqual ( instance_ref [ '<STR_LIT>' ] , None ) <EOL> self . assertEqual ( instance_ref [ '<STR_LIT>' ] , None ) <EOL> launch = utils . utcnow ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> instance_ref = db . instance_get ( self . context , instance_id ) <EOL> self . assert_ ( instance_ref [ '<STR_LIT>' ] > launch ) <EOL> self . assertEqual ( instance_ref [ '<STR_LIT>' ] , None ) <EOL> terminate = utils . utcnow ( ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> self . context = self . context . elevated ( True ) <EOL> instance_ref = db . instance_get ( self . context , instance_id ) <EOL> self . assert_ ( instance_ref [ '<STR_LIT>' ] < terminate ) <EOL> self . assert_ ( instance_ref [ '<STR_LIT>' ] > terminate ) <EOL> def test_stop ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . stop_instance ( self . context , instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_start ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . stop_instance ( self . context , instance_id ) <EOL> self . compute . start_instance ( self . context , instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_pause ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . pause_instance ( self . context , instance_id ) <EOL> self . compute . unpause_instance ( self . context , instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_suspend ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . suspend_instance ( self . context , instance_id ) <EOL> self . compute . resume_instance ( self . context , instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_reboot ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . reboot_instance ( self . context , instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_set_admin_password ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . set_admin_password ( self . context , instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_inject_file ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . inject_file ( self . context , instance_id , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_agent_update ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . agent_update ( self . context , instance_id , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_snapshot ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> name = "<STR_LIT>" <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute . snapshot_instance ( self . context , instance_id , name ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_snapshot_conflict_backup ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> instance_values = { '<STR_LIT>' : task_states . IMAGE_BACKUP } <EOL> db . instance_update ( self . context , instance_id , instance_values ) <EOL> self . assertRaises ( exception . InstanceBackingUp , <EOL> self . compute_api . backup , <EOL> self . context , <EOL> instance_id , <EOL> None , <EOL> None , <EOL> None ) <EOL> db . instance_destroy ( self . context , instance_id ) <EOL> def test_snapshot_conflict_snapshot ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> instance_values = { '<STR_LIT>' : task_states . IMAGE_SNAPSHOT } <EOL> db . instance_update ( self . context , instance_id , instance_values ) <EOL> self . assertRaises ( exception . InstanceSnapshotting , <EOL> self . compute_api . snapshot , <EOL> self . context , <EOL> instance_id , <EOL> None ) <EOL> db . instance_destroy ( self . context , instance_id ) <EOL> def test_console_output ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> console = self . compute . get_console_output ( self . context , <EOL> instance_id ) <EOL> self . assert_ ( console ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_ajax_console ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> console = self . compute . get_ajax_console ( self . context , <EOL> instance_id ) <EOL> self . assert_ ( set ( [ '<STR_LIT>' , '<STR_LIT:host>' , '<STR_LIT:port>' ] ) . issubset ( console . keys ( ) ) ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_vnc_console ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> console = self . compute . get_vnc_console ( self . context , <EOL> instance_id ) <EOL> self . assert_ ( console ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_run_instance_usage_notification ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . assertEquals ( len ( test_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = test_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> payload = msg [ '<STR_LIT>' ] <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , instance_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = instance_types . get_instance_type_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEquals ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_terminate_usage_notification ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> test_notifier . NOTIFICATIONS = [ ] <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> self . assertEquals ( len ( test_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = test_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> payload = msg [ '<STR_LIT>' ] <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , instance_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = instance_types . get_instance_type_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEquals ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> def test_run_instance_existing ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . assertRaises ( exception . Error , <EOL> self . compute . run_instance , <EOL> self . context , <EOL> instance_id ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_lock ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> non_admin_context = context . RequestContext ( None , None , False , False ) <EOL> self . compute . lock_instance ( self . context , instance_id ) <EOL> ret_val = self . compute . reboot_instance ( non_admin_context , instance_id ) <EOL> self . assertEqual ( ret_val , False ) <EOL> self . compute . unlock_instance ( self . context , instance_id ) <EOL> ret_val = self . compute . reboot_instance ( non_admin_context , instance_id ) <EOL> self . assertEqual ( ret_val , None ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_finish_resize ( self ) : <EOL> """<STR_LIT>""" <EOL> def fake ( * args , ** kwargs ) : <EOL> pass <EOL> self . stubs . Set ( self . compute . driver , '<STR_LIT>' , fake ) <EOL> self . stubs . Set ( self . compute . network_api , '<STR_LIT>' , fake ) <EOL> context = self . context . elevated ( ) <EOL> instance_id = self . _create_instance ( ) <EOL> instance_ref = db . instance_get ( context , instance_id ) <EOL> self . compute . prep_resize ( context , instance_ref [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> migration_ref = db . migration_get_by_instance_and_status ( context , <EOL> instance_ref [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> try : <EOL> self . compute . finish_resize ( context , instance_ref [ '<STR_LIT>' ] , <EOL> int ( migration_ref [ '<STR_LIT:id>' ] ) , { } ) <EOL> except KeyError , e : <EOL> self . fail ( ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_resize_instance_notification ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> context = self . context . elevated ( ) <EOL> inst_ref = db . instance_get ( context , instance_id ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> test_notifier . NOTIFICATIONS = [ ] <EOL> db . instance_update ( self . context , instance_id , { '<STR_LIT:host>' : '<STR_LIT:foo>' } ) <EOL> self . compute . prep_resize ( context , inst_ref [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> migration_ref = db . migration_get_by_instance_and_status ( context , <EOL> inst_ref [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( len ( test_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = test_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> payload = msg [ '<STR_LIT>' ] <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , instance_id ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = instance_types . get_instance_type_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEquals ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertEquals ( payload [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def test_resize_instance ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> context = self . context . elevated ( ) <EOL> inst_ref = db . instance_get ( context , instance_id ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> db . instance_update ( self . context , inst_ref [ '<STR_LIT>' ] , <EOL> { '<STR_LIT:host>' : '<STR_LIT:foo>' } ) <EOL> self . compute . prep_resize ( context , inst_ref [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> migration_ref = db . migration_get_by_instance_and_status ( context , <EOL> inst_ref [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . compute . resize_instance ( context , inst_ref [ '<STR_LIT>' ] , <EOL> migration_ref [ '<STR_LIT:id>' ] ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def test_resize_invalid_flavor_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> context = self . context . elevated ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . assertRaises ( exception . NotFound , self . compute_api . resize , <EOL> context , instance_id , <NUM_LIT:200> ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def test_resize_down_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> context = self . context . elevated ( ) <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> inst_type = instance_types . get_instance_type_by_name ( '<STR_LIT>' ) <EOL> db . instance_update ( self . context , instance_id , <EOL> { '<STR_LIT>' : inst_type [ '<STR_LIT:id>' ] } ) <EOL> self . assertRaises ( exception . CannotResizeToSmallerSize , <EOL> self . compute_api . resize , context , instance_id , <NUM_LIT:1> ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def test_resize_same_size_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> context = self . context . elevated ( ) <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . assertRaises ( exception . CannotResizeToSameSize , <EOL> self . compute_api . resize , context , instance_id , <NUM_LIT:1> ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def test_finish_revert_resize ( self ) : <EOL> """<STR_LIT>""" <EOL> context = self . context . elevated ( ) <EOL> instance_id = self . _create_instance ( ) <EOL> def fake ( * args , ** kwargs ) : <EOL> pass <EOL> self . stubs . Set ( self . compute . driver , '<STR_LIT>' , fake ) <EOL> self . stubs . Set ( self . compute . driver , '<STR_LIT>' , fake ) <EOL> self . stubs . Set ( self . compute . network_api , '<STR_LIT>' , fake ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> inst_ref = db . instance_get ( context , instance_id ) <EOL> instance_type_ref = db . instance_type_get ( context , <EOL> inst_ref [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( instance_type_ref [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> db . instance_update ( self . context , instance_id , { '<STR_LIT:host>' : '<STR_LIT:foo>' } ) <EOL> new_instance_type_ref = db . instance_type_get_by_flavor_id ( context , <NUM_LIT:3> ) <EOL> self . compute . prep_resize ( context , inst_ref [ '<STR_LIT>' ] , <EOL> new_instance_type_ref [ '<STR_LIT:id>' ] ) <EOL> migration_ref = db . migration_get_by_instance_and_status ( context , <EOL> inst_ref [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . compute . resize_instance ( context , inst_ref [ '<STR_LIT>' ] , <EOL> migration_ref [ '<STR_LIT:id>' ] ) <EOL> self . compute . finish_resize ( context , inst_ref [ '<STR_LIT>' ] , <EOL> int ( migration_ref [ '<STR_LIT:id>' ] ) , { } ) <EOL> inst_ref = db . instance_get ( context , instance_id ) <EOL> instance_type_ref = db . instance_type_get ( context , <EOL> inst_ref [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( instance_type_ref [ '<STR_LIT>' ] , <NUM_LIT:3> ) <EOL> self . compute . revert_resize ( context , inst_ref [ '<STR_LIT>' ] , <EOL> migration_ref [ '<STR_LIT:id>' ] ) <EOL> self . compute . finish_revert_resize ( context , inst_ref [ '<STR_LIT>' ] , <EOL> migration_ref [ '<STR_LIT:id>' ] ) <EOL> inst_ref = db . instance_get ( context , instance_id ) <EOL> instance_type_ref = db . instance_type_get ( context , <EOL> inst_ref [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( instance_type_ref [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def test_get_by_flavor_id ( self ) : <EOL> type = instance_types . get_instance_type_by_flavor_id ( <NUM_LIT:1> ) <EOL> self . assertEqual ( type [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> def test_resize_same_source_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> inst_ref = db . instance_get ( self . context , instance_id ) <EOL> self . assertRaises ( exception . Error , self . compute . prep_resize , <EOL> self . context , inst_ref [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_migrate ( self ) : <EOL> context = self . context . elevated ( ) <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> self . compute_api . resize ( context , instance_id , None ) <EOL> self . compute . terminate_instance ( context , instance_id ) <EOL> def _setup_other_managers ( self ) : <EOL> self . volume_manager = utils . import_object ( FLAGS . volume_manager ) <EOL> self . network_manager = utils . import_object ( FLAGS . network_manager ) <EOL> self . compute_driver = utils . import_object ( FLAGS . compute_driver ) <EOL> def test_pre_live_migration_instance_has_no_fixed_ip ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_ref = self . _get_dummy_instance ( ) <EOL> c = context . get_admin_context ( ) <EOL> i_id = instance_ref [ '<STR_LIT:id>' ] <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> dbmock . instance_get ( c , i_id ) . AndReturn ( instance_ref ) <EOL> dbmock . instance_get_fixed_addresses ( c , i_id ) . AndReturn ( None ) <EOL> self . compute . db = dbmock <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( exception . NotFound , <EOL> self . compute . pre_live_migration , <EOL> c , instance_ref [ '<STR_LIT:id>' ] , time = FakeTime ( ) ) <EOL> def test_pre_live_migration_instance_has_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> c = context . get_admin_context ( ) <EOL> self . _setup_other_managers ( ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> volmock = self . mox . CreateMock ( self . volume_manager ) <EOL> drivermock = self . mox . CreateMock ( self . compute_driver ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> dbmock . instance_get_fixed_addresses ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( '<STR_LIT>' ) <EOL> for i in range ( len ( i_ref [ '<STR_LIT>' ] ) ) : <EOL> vid = i_ref [ '<STR_LIT>' ] [ i ] [ '<STR_LIT:id>' ] <EOL> volmock . setup_compute_volume ( c , vid ) . InAnyOrder ( '<STR_LIT>' ) <EOL> drivermock . plug_vifs ( i_ref , [ ] ) <EOL> drivermock . ensure_filtering_rules_for_instance ( i_ref , [ ] ) <EOL> self . compute . db = dbmock <EOL> self . compute . volume_manager = volmock <EOL> self . compute . driver = drivermock <EOL> self . mox . ReplayAll ( ) <EOL> ret = self . compute . pre_live_migration ( c , i_ref [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( ret , None ) <EOL> def test_pre_live_migration_instance_has_no_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> i_ref [ '<STR_LIT>' ] = [ ] <EOL> c = context . get_admin_context ( ) <EOL> self . _setup_other_managers ( ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> drivermock = self . mox . CreateMock ( self . compute_driver ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> dbmock . instance_get_fixed_addresses ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( compute_manager . LOG , '<STR_LIT:info>' ) <EOL> compute_manager . LOG . info ( _ ( "<STR_LIT>" ) , i_ref [ '<STR_LIT>' ] ) <EOL> drivermock . plug_vifs ( i_ref , [ ] ) <EOL> drivermock . ensure_filtering_rules_for_instance ( i_ref , [ ] ) <EOL> self . compute . db = dbmock <EOL> self . compute . driver = drivermock <EOL> self . mox . ReplayAll ( ) <EOL> ret = self . compute . pre_live_migration ( c , i_ref [ '<STR_LIT:id>' ] , time = FakeTime ( ) ) <EOL> self . assertEqual ( ret , None ) <EOL> def test_pre_live_migration_setup_compute_node_fail ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> c = context . get_admin_context ( ) <EOL> self . _setup_other_managers ( ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> netmock = self . mox . CreateMock ( self . network_manager ) <EOL> volmock = self . mox . CreateMock ( self . volume_manager ) <EOL> drivermock = self . mox . CreateMock ( self . compute_driver ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> dbmock . instance_get_fixed_addresses ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( '<STR_LIT>' ) <EOL> for i in range ( len ( i_ref [ '<STR_LIT>' ] ) ) : <EOL> volmock . setup_compute_volume ( c , i_ref [ '<STR_LIT>' ] [ i ] [ '<STR_LIT:id>' ] ) <EOL> for i in range ( FLAGS . live_migration_retry_count ) : <EOL> drivermock . plug_vifs ( i_ref , [ ] ) . AndRaise ( exception . ProcessExecutionError ( ) ) <EOL> self . compute . db = dbmock <EOL> self . compute . network_manager = netmock <EOL> self . compute . volume_manager = volmock <EOL> self . compute . driver = drivermock <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( exception . ProcessExecutionError , <EOL> self . compute . pre_live_migration , <EOL> c , i_ref [ '<STR_LIT:id>' ] , time = FakeTime ( ) ) <EOL> def test_live_migration_works_correctly_with_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> c = context . get_admin_context ( ) <EOL> topic = db . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> self . mox . StubOutWithMock ( rpc , '<STR_LIT>' ) <EOL> rpc . call ( c , FLAGS . volume_topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] } } ) <EOL> dbmock . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) . AndReturn ( topic ) <EOL> rpc . call ( c , topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None } } ) <EOL> self . mox . StubOutWithMock ( self . compute . driver , '<STR_LIT>' ) <EOL> self . compute . driver . live_migration ( c , i_ref , i_ref [ '<STR_LIT:host>' ] , <EOL> self . compute . post_live_migration , <EOL> self . compute . rollback_live_migration , <EOL> False ) <EOL> self . compute . db = dbmock <EOL> self . mox . ReplayAll ( ) <EOL> ret = self . compute . live_migration ( c , i_ref [ '<STR_LIT:id>' ] , i_ref [ '<STR_LIT:host>' ] ) <EOL> self . assertEqual ( ret , None ) <EOL> def test_live_migration_dest_raises_exception ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> c = context . get_admin_context ( ) <EOL> topic = db . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> self . mox . StubOutWithMock ( rpc , '<STR_LIT>' ) <EOL> rpc . call ( c , FLAGS . volume_topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] } } ) <EOL> dbmock . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) . AndReturn ( topic ) <EOL> rpc . call ( c , topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None } } ) . AndRaise ( rpc . RemoteError ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> dbmock . instance_update ( c , i_ref [ '<STR_LIT:id>' ] , { '<STR_LIT>' : vm_states . ACTIVE , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT:host>' : i_ref [ '<STR_LIT:host>' ] } ) <EOL> for v in i_ref [ '<STR_LIT>' ] : <EOL> dbmock . volume_update ( c , v [ '<STR_LIT:id>' ] , { '<STR_LIT:status>' : '<STR_LIT>' } ) <EOL> rpc . call ( c , topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : v [ '<STR_LIT:id>' ] } } ) <EOL> self . compute . db = dbmock <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( rpc . RemoteError , <EOL> self . compute . live_migration , <EOL> c , i_ref [ '<STR_LIT:id>' ] , i_ref [ '<STR_LIT:host>' ] ) <EOL> def test_live_migration_dest_raises_exception_no_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> i_ref [ '<STR_LIT>' ] = [ ] <EOL> c = context . get_admin_context ( ) <EOL> topic = db . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> dbmock . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) . AndReturn ( topic ) <EOL> self . mox . StubOutWithMock ( rpc , '<STR_LIT>' ) <EOL> rpc . call ( c , topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None } } ) . AndRaise ( rpc . RemoteError ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> dbmock . instance_update ( c , i_ref [ '<STR_LIT:id>' ] , { '<STR_LIT>' : vm_states . ACTIVE , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT:host>' : i_ref [ '<STR_LIT:host>' ] } ) <EOL> self . compute . db = dbmock <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( rpc . RemoteError , <EOL> self . compute . live_migration , <EOL> c , i_ref [ '<STR_LIT:id>' ] , i_ref [ '<STR_LIT:host>' ] ) <EOL> def test_live_migration_works_correctly_no_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> i_ref = self . _get_dummy_instance ( ) <EOL> i_ref [ '<STR_LIT>' ] = [ ] <EOL> c = context . get_admin_context ( ) <EOL> topic = db . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) <EOL> dbmock = self . mox . CreateMock ( db ) <EOL> dbmock . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) . AndReturn ( i_ref ) <EOL> self . mox . StubOutWithMock ( rpc , '<STR_LIT>' ) <EOL> dbmock . queue_get_for ( c , FLAGS . compute_topic , i_ref [ '<STR_LIT:host>' ] ) . AndReturn ( topic ) <EOL> rpc . call ( c , topic , { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None } } ) <EOL> self . mox . StubOutWithMock ( self . compute . driver , '<STR_LIT>' ) <EOL> self . compute . driver . live_migration ( c , i_ref , i_ref [ '<STR_LIT:host>' ] , <EOL> self . compute . post_live_migration , <EOL> self . compute . rollback_live_migration , <EOL> False ) <EOL> self . compute . db = dbmock <EOL> self . mox . ReplayAll ( ) <EOL> ret = self . compute . live_migration ( c , i_ref [ '<STR_LIT:id>' ] , i_ref [ '<STR_LIT:host>' ] ) <EOL> self . assertEqual ( ret , None ) <EOL> def test_post_live_migration_working_correctly ( self ) : <EOL> """<STR_LIT>""" <EOL> dest = '<STR_LIT>' <EOL> flo_addr = '<STR_LIT>' <EOL> c = context . get_admin_context ( ) <EOL> instance_id = self . _create_instance ( ) <EOL> i_ref = db . instance_get ( c , instance_id ) <EOL> db . instance_update ( c , i_ref [ '<STR_LIT:id>' ] , { '<STR_LIT>' : vm_states . MIGRATING , <EOL> '<STR_LIT>' : power_state . PAUSED } ) <EOL> v_ref = db . volume_create ( c , { '<STR_LIT:size>' : <NUM_LIT:1> , '<STR_LIT>' : instance_id } ) <EOL> fix_addr = db . fixed_ip_create ( c , { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id } ) <EOL> fix_ref = db . fixed_ip_get_by_address ( c , fix_addr ) <EOL> flo_ref = db . floating_ip_create ( c , { '<STR_LIT:address>' : flo_addr , <EOL> '<STR_LIT>' : fix_ref [ '<STR_LIT:id>' ] } ) <EOL> i_ref = db . instance_get ( c , instance_id ) <EOL> self . mox . StubOutWithMock ( self . compute . volume_manager , <EOL> '<STR_LIT>' ) <EOL> for v in i_ref [ '<STR_LIT>' ] : <EOL> self . compute . volume_manager . remove_compute_volume ( c , v [ '<STR_LIT:id>' ] ) <EOL> self . mox . StubOutWithMock ( self . compute . driver , '<STR_LIT>' ) <EOL> self . compute . driver . unfilter_instance ( i_ref , [ ] ) <EOL> self . mox . StubOutWithMock ( rpc , '<STR_LIT>' ) <EOL> rpc . call ( c , db . queue_get_for ( c , FLAGS . compute_topic , dest ) , <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : { '<STR_LIT>' : i_ref [ '<STR_LIT:id>' ] , '<STR_LIT>' : False } } ) <EOL> self . mox . ReplayAll ( ) <EOL> ret = self . compute . post_live_migration ( c , i_ref , dest ) <EOL> i_ref = db . instance_get ( c , i_ref [ '<STR_LIT:id>' ] ) <EOL> c1 = ( i_ref [ '<STR_LIT:host>' ] == dest ) <EOL> flo_refs = db . floating_ip_get_all_by_host ( c , dest ) <EOL> c2 = ( len ( flo_refs ) != <NUM_LIT:0> and flo_refs [ <NUM_LIT:0> ] [ '<STR_LIT:address>' ] == flo_addr ) <EOL> self . assertTrue ( c1 and c2 ) <EOL> db . instance_destroy ( c , instance_id ) <EOL> db . volume_destroy ( c , v_ref [ '<STR_LIT:id>' ] ) <EOL> db . floating_ip_destroy ( c , flo_addr ) <EOL> def test_run_kill_vm ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stubs . Set ( compute_manager . ComputeManager , <EOL> '<STR_LIT>' , nop_report_driver_status ) <EOL> instance_id = self . _create_instance ( ) <EOL> self . compute . run_instance ( self . context , instance_id ) <EOL> instances = db . instance_get_all ( context . get_admin_context ( ) ) <EOL> LOG . info ( _ ( "<STR_LIT>" ) , instances ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> instance_name = instances [ <NUM_LIT:0> ] . name <EOL> self . compute . driver . test_remove_vm ( instance_name ) <EOL> error_list = self . compute . periodic_tasks ( context . get_admin_context ( ) ) <EOL> self . assertFalse ( error_list ) <EOL> instances = db . instance_get_all ( context . get_admin_context ( ) ) <EOL> LOG . info ( _ ( "<STR_LIT>" ) , instances ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( power_state . NOSTATE , instances [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> def test_get_all_by_name_regexp ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:20> } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:30> } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id1 in instance_ids ) <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( instance_id1 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id1 in instance_ids ) <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_instance_name_regexp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . flags ( instance_name_template = '<STR_LIT>' ) <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( ) <EOL> instance_id2 = self . _create_instance ( { '<STR_LIT:id>' : <NUM_LIT:2> } ) <EOL> instance_id3 = self . _create_instance ( { '<STR_LIT:id>' : <NUM_LIT:10> } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:3> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id1 in instance_ids ) <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_by_fixed_ip ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( ) <EOL> instance_id2 = self . _create_instance ( { '<STR_LIT:id>' : <NUM_LIT:20> } ) <EOL> instance_id3 = self . _create_instance ( { '<STR_LIT:id>' : <NUM_LIT:30> } ) <EOL> vif_ref1 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref2 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : vif_ref1 [ '<STR_LIT:id>' ] } ) <EOL> db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : vif_ref2 [ '<STR_LIT:id>' ] } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id1 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id2 ) <EOL> db . virtual_interface_delete ( c , vif_ref1 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref2 [ '<STR_LIT:id>' ] ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> def test_get_all_by_ip_regexp ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:20> } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:30> } ) <EOL> vif_ref1 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref2 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref3 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id3 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : vif_ref1 [ '<STR_LIT:id>' ] } ) <EOL> db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : vif_ref2 [ '<STR_LIT:id>' ] } ) <EOL> fix_addr = db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id3 , <EOL> '<STR_LIT>' : vif_ref3 [ '<STR_LIT:id>' ] } ) <EOL> fix_ref = db . fixed_ip_get_by_address ( c , fix_addr ) <EOL> flo_ref = db . floating_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : fix_ref [ '<STR_LIT:id>' ] } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:3> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:3> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id1 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id2 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id3 ) <EOL> db . virtual_interface_delete ( c , vif_ref1 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref2 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref3 [ '<STR_LIT:id>' ] ) <EOL> db . floating_ip_destroy ( c , '<STR_LIT>' ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_ipv6_regexp ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:20> } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:30> } ) <EOL> vif_ref1 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref2 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref3 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id3 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id1 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:3> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id1 in instance_ids ) <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> db . virtual_interface_delete ( c , vif_ref1 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref2 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref3 [ '<STR_LIT:id>' ] ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_multiple_options_at_once ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:20> } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:30> } ) <EOL> vif_ref1 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref2 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> vif_ref3 = db . virtual_interface_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id3 , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id1 , <EOL> '<STR_LIT>' : vif_ref1 [ '<STR_LIT:id>' ] } ) <EOL> db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id2 , <EOL> '<STR_LIT>' : vif_ref2 [ '<STR_LIT:id>' ] } ) <EOL> fix_addr = db . fixed_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_id3 , <EOL> '<STR_LIT>' : vif_ref3 [ '<STR_LIT:id>' ] } ) <EOL> fix_ref = db . fixed_ip_get_by_address ( c , fix_addr ) <EOL> flo_ref = db . floating_ip_create ( c , <EOL> { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : fix_ref [ '<STR_LIT:id>' ] } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id3 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id2 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id3 ) <EOL> db . virtual_interface_delete ( c , vif_ref1 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref2 [ '<STR_LIT:id>' ] ) <EOL> db . virtual_interface_delete ( c , vif_ref3 [ '<STR_LIT:id>' ] ) <EOL> db . floating_ip_destroy ( c , '<STR_LIT>' ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_image ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT:id>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT:id>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:image>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:image>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id1 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:image>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT:image>' : [ '<STR_LIT>' , '<STR_LIT>' ] } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:3> ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_flavor ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT:id>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT:id>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : <NUM_LIT:2> } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : <NUM_LIT:5> } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> self . assertRaises ( exception . FlavorNotFound , <EOL> self . compute_api . get_all , <EOL> c , search_opts = { '<STR_LIT>' : <NUM_LIT> } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : <NUM_LIT:3> } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id1 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_state ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id1 = self . _create_instance ( { <EOL> '<STR_LIT>' : power_state . SHUTDOWN , <EOL> } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT:id>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : power_state . RUNNING , <EOL> } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT:id>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : power_state . RUNNING , <EOL> } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : power_state . SUSPENDED } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : power_state . SHUTDOWN } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id1 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : power_state . RUNNING } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id2 in instance_ids ) <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : [ power_state . SHUTDOWN , <EOL> power_state . RUNNING ] } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:3> ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> def test_get_all_by_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> c = context . get_admin_context ( ) <EOL> instance_id0 = self . _create_instance ( ) <EOL> instance_id1 = self . _create_instance ( { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> instance_id2 = self . _create_instance ( { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> instance_id3 = self . _create_instance ( { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> instance_id4 = self . _create_instance ( { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : { } } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:5> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:0> ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id2 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:2> ) <EOL> instance_ids = [ instance . id for instance in instances ] <EOL> self . assertTrue ( instance_id3 in instance_ids ) <EOL> self . assertTrue ( instance_id4 in instance_ids ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id4 ) <EOL> instances = self . compute_api . get_all ( c , <EOL> search_opts = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ] } ) <EOL> self . assertEqual ( len ( instances ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( instances [ <NUM_LIT:0> ] . id , instance_id4 ) <EOL> db . instance_destroy ( c , instance_id0 ) <EOL> db . instance_destroy ( c , instance_id1 ) <EOL> db . instance_destroy ( c , instance_id2 ) <EOL> db . instance_destroy ( c , instance_id3 ) <EOL> db . instance_destroy ( c , instance_id4 ) <EOL> @ staticmethod <EOL> def _parse_db_block_device_mapping ( bdm_ref ) : <EOL> attr_list = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> bdm = { } <EOL> for attr in attr_list : <EOL> val = bdm_ref . get ( attr , None ) <EOL> if val : <EOL> bdm [ attr ] = val <EOL> return bdm <EOL> def test_update_block_device_mapping ( self ) : <EOL> swap_size = <NUM_LIT:1> <EOL> instance_type = { '<STR_LIT>' : swap_size } <EOL> instance_id = self . _create_instance ( ) <EOL> mappings = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT:root>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> block_device_mapping = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : False } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : False } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : False } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : False } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } ] <EOL> self . compute_api . _update_image_block_device_mapping ( <EOL> self . context , instance_type , instance_id , mappings ) <EOL> bdms = [ self . _parse_db_block_device_mapping ( bdm_ref ) <EOL> for bdm_ref in db . block_device_mapping_get_all_by_instance ( <EOL> self . context , instance_id ) ] <EOL> expected_result = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : swap_size } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> ] <EOL> bdms . sort ( ) <EOL> expected_result . sort ( ) <EOL> self . assertDictListMatch ( bdms , expected_result ) <EOL> self . compute_api . _update_block_device_mapping ( <EOL> self . context , instance_types . get_default_instance_type ( ) , <EOL> instance_id , block_device_mapping ) <EOL> bdms = [ self . _parse_db_block_device_mapping ( bdm_ref ) <EOL> for bdm_ref in db . block_device_mapping_get_all_by_instance ( <EOL> self . context , instance_id ) ] <EOL> expected_result = [ <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : swap_size } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> bdms . sort ( ) <EOL> expected_result . sort ( ) <EOL> self . assertDictListMatch ( bdms , expected_result ) <EOL> for bdm in db . block_device_mapping_get_all_by_instance ( <EOL> self . context , instance_id ) : <EOL> db . block_device_mapping_destroy ( self . context , bdm [ '<STR_LIT:id>' ] ) <EOL> self . compute . terminate_instance ( self . context , instance_id ) <EOL> def test_volume_size ( self ) : <EOL> local_size = <NUM_LIT:2> <EOL> swap_size = <NUM_LIT:3> <EOL> inst_type = { '<STR_LIT>' : local_size , '<STR_LIT>' : swap_size } <EOL> self . assertEqual ( self . compute_api . _volume_size ( inst_type , <EOL> '<STR_LIT>' ) , <EOL> local_size ) <EOL> self . assertEqual ( self . compute_api . _volume_size ( inst_type , <EOL> '<STR_LIT>' ) , <EOL> <NUM_LIT:0> ) <EOL> self . assertEqual ( self . compute_api . _volume_size ( inst_type , <EOL> '<STR_LIT>' ) , <EOL> swap_size ) </s>
