<s> import datetime <EOL> import mock <EOL> from nova . conductor import manager as conductor_manager <EOL> from nova import context <EOL> from nova import db <EOL> from nova import objects <EOL> from nova . tests . functional . api_sample_tests import test_servers <EOL> from nova . tests . unit import fake_instance <EOL> class ServerMigrationsSampleJsonTest ( test_servers . ServersSampleBase ) : <EOL> extension_name = '<STR_LIT>' <EOL> scenarios = [ ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) ] <EOL> extra_extensions_to_load = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( ServerMigrationsSampleJsonTest , self ) . setUp ( ) <EOL> self . uuid = self . _post_server ( ) <EOL> self . api . microversion = '<STR_LIT>' <EOL> @ mock . patch . object ( conductor_manager . ComputeTaskManager , '<STR_LIT>' ) <EOL> @ mock . patch . object ( db , '<STR_LIT>' ) <EOL> @ mock . patch . object ( objects . Migration , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_live_migrate_force_complete ( self , live_migration_pause_instance , <EOL> get_by_id_and_instance , <EOL> service_get_by_compute_host , <EOL> _live_migrate ) : <EOL> migration = objects . Migration ( ) <EOL> migration . id = <NUM_LIT:1> <EOL> migration . status = '<STR_LIT>' <EOL> get_by_id_and_instance . return_value = migration <EOL> self . _do_post ( '<STR_LIT>' % self . uuid , '<STR_LIT>' , <EOL> { '<STR_LIT>' : self . compute . host } ) <EOL> response = self . _do_post ( '<STR_LIT>' <EOL> % ( self . uuid , '<STR_LIT:3>' ) , '<STR_LIT>' , { } ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> def test_get_migration ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> def test_list_migrations ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> class ServerMigrationsSamplesJsonTestV2_23 ( test_servers . ServersSampleBase ) : <EOL> ADMIN_API = True <EOL> extension_name = "<STR_LIT>" <EOL> microversion = '<STR_LIT>' <EOL> scenarios = [ ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) ] <EOL> UUID_1 = '<STR_LIT>' <EOL> UUID_2 = '<STR_LIT>' <EOL> fake_migrations = [ <EOL> { <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:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : UUID_1 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : datetime . datetime ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:2> ) , <EOL> '<STR_LIT>' : datetime . datetime ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:2> ) , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False <EOL> } , <EOL> { <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:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : UUID_2 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : datetime . datetime ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:2> ) , <EOL> '<STR_LIT>' : datetime . datetime ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:2> ) , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False <EOL> } <EOL> ] <EOL> def setUp ( self ) : <EOL> super ( ServerMigrationsSamplesJsonTestV2_23 , self ) . setUp ( ) <EOL> fake_context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . mig1 = objects . Migration ( <EOL> context = fake_context , ** self . fake_migrations [ <NUM_LIT:0> ] ) <EOL> self . mig1 . create ( ) <EOL> self . mig2 = objects . Migration ( <EOL> context = fake_context , ** self . fake_migrations [ <NUM_LIT:1> ] ) <EOL> self . mig2 . create ( ) <EOL> fake_ins = fake_instance . fake_db_instance ( uuid = self . UUID_1 ) <EOL> fake_ins . pop ( "<STR_LIT>" ) <EOL> fake_ins . pop ( "<STR_LIT>" ) <EOL> fake_ins . pop ( "<STR_LIT>" ) <EOL> fake_ins . pop ( "<STR_LIT>" ) <EOL> fake_ins . pop ( "<STR_LIT>" ) <EOL> fake_ins . pop ( "<STR_LIT:id>" ) <EOL> self . instance = objects . Instance ( <EOL> context = fake_context , <EOL> ** fake_ins ) <EOL> self . instance . create ( ) <EOL> def test_get_migration ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' % <EOL> ( self . fake_migrations [ <NUM_LIT:0> ] [ "<STR_LIT>" ] , <EOL> self . mig1 . id ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_code ) <EOL> self . _verify_response ( '<STR_LIT>' , <EOL> { "<STR_LIT>" : self . UUID_1 } , <EOL> response , <NUM_LIT:200> ) <EOL> def test_list_migrations ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' % <EOL> self . fake_migrations [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_code ) <EOL> self . _verify_response ( '<STR_LIT>' , <EOL> { "<STR_LIT>" : self . UUID_1 } , <EOL> response , <NUM_LIT:200> ) <EOL> class ServerMigrationsSampleJsonTestV2_24 ( test_servers . ServersSampleBase ) : <EOL> ADMIN_API = True <EOL> extension_name = "<STR_LIT>" <EOL> scenarios = [ ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) ] <EOL> extra_extensions_to_load = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( ServerMigrationsSampleJsonTestV2_24 , self ) . setUp ( ) <EOL> self . api . microversion = '<STR_LIT>' <EOL> self . uuid = self . _post_server ( ) <EOL> self . context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> fake_migration = { <EOL> '<STR_LIT>' : self . compute . host , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . uuid , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } <EOL> self . migration = objects . Migration ( context = self . context , <EOL> ** fake_migration ) <EOL> self . migration . create ( ) <EOL> @ mock . patch . object ( conductor_manager . ComputeTaskManager , '<STR_LIT>' ) <EOL> def test_live_migrate_abort ( self , _live_migrate ) : <EOL> self . _do_post ( '<STR_LIT>' % self . uuid , '<STR_LIT>' , <EOL> { '<STR_LIT>' : self . compute . host } ) <EOL> uri = '<STR_LIT>' % ( self . uuid , self . migration . id ) <EOL> response = self . _do_delete ( uri ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> @ mock . patch . object ( conductor_manager . ComputeTaskManager , '<STR_LIT>' ) <EOL> def test_live_migrate_abort_migration_not_found ( self , _live_migrate ) : <EOL> self . _do_post ( '<STR_LIT>' % self . uuid , '<STR_LIT>' , <EOL> { '<STR_LIT>' : self . compute . host } ) <EOL> uri = '<STR_LIT>' % ( self . uuid , '<STR_LIT>' ) <EOL> response = self . _do_delete ( uri ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> @ mock . patch . object ( conductor_manager . ComputeTaskManager , '<STR_LIT>' ) <EOL> def test_live_migrate_abort_migration_not_running ( self , _live_migrate ) : <EOL> self . migration . status = '<STR_LIT>' <EOL> self . migration . save ( ) <EOL> self . _do_post ( '<STR_LIT>' % self . uuid , '<STR_LIT>' , <EOL> { '<STR_LIT>' : self . compute . host } ) <EOL> uri = '<STR_LIT>' % ( self . uuid , self . migration . id ) <EOL> response = self . _do_delete ( uri ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) </s>
