<s> """<STR_LIT>""" <EOL> import copy <EOL> import string <EOL> import uuid <EOL> import mock <EOL> from oslo_serialization import jsonutils <EOL> from oslo_utils import importutils <EOL> import six <EOL> from nova . compute import flavors <EOL> from nova . compute import power_state <EOL> from nova . compute import task_states <EOL> from nova . compute import utils as compute_utils <EOL> import nova . conf <EOL> from nova import context <EOL> from nova import exception <EOL> from nova . image import glance <EOL> from nova . network import api as network_api <EOL> from nova . network import model <EOL> from nova import objects <EOL> from nova . objects import block_device as block_device_obj <EOL> from nova import rpc <EOL> from nova import test <EOL> from nova . tests . unit import fake_block_device <EOL> from nova . tests . unit import fake_instance <EOL> from nova . tests . unit import fake_network <EOL> from nova . tests . unit import fake_notifier <EOL> from nova . tests . unit import fake_server_actions <EOL> import nova . tests . unit . image . fake <EOL> from nova . tests . unit . objects import test_flavor <EOL> from nova . tests import uuidsentinel as uuids <EOL> CONF = nova . conf . CONF <EOL> def create_instance ( context , user_id = '<STR_LIT>' , project_id = '<STR_LIT>' , params = None ) : <EOL> """<STR_LIT>""" <EOL> flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> net_info = model . NetworkInfo ( [ ] ) <EOL> info_cache = objects . InstanceInfoCache ( network_info = net_info ) <EOL> inst = objects . Instance ( context = context , <EOL> image_ref = uuids . fake_image_ref , <EOL> reservation_id = '<STR_LIT>' , <EOL> user_id = user_id , <EOL> project_id = project_id , <EOL> instance_type_id = flavor . id , <EOL> flavor = flavor , <EOL> old_flavor = None , <EOL> new_flavor = None , <EOL> system_metadata = { } , <EOL> ami_launch_index = <NUM_LIT:0> , <EOL> root_gb = <NUM_LIT:0> , <EOL> ephemeral_gb = <NUM_LIT:0> , <EOL> info_cache = info_cache ) <EOL> if params : <EOL> inst . update ( params ) <EOL> inst . create ( ) <EOL> return inst <EOL> class ComputeValidateDeviceTestCase ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ComputeValidateDeviceTestCase , self ) . setUp ( ) <EOL> self . context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if '<STR_LIT>' in self . id ( ) : <EOL> self . flags ( compute_driver = '<STR_LIT>' ) <EOL> self . instance = objects . Instance ( uuid = uuid . uuid4 ( ) . hex , <EOL> root_device_name = None , <EOL> default_ephemeral_device = None ) <EOL> else : <EOL> self . instance = objects . Instance ( uuid = uuid . uuid4 ( ) . hex , <EOL> root_device_name = '<STR_LIT>' , <EOL> default_ephemeral_device = '<STR_LIT>' ) <EOL> flavor = objects . Flavor ( ** test_flavor . fake_flavor ) <EOL> self . instance . system_metadata = { } <EOL> self . instance . flavor = flavor <EOL> self . instance . default_swap_device = None <EOL> self . data = [ ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> lambda context , instance : self . data ) <EOL> def _validate_device ( self , device = None ) : <EOL> bdms = objects . BlockDeviceMappingList . get_by_instance_uuid ( <EOL> self . context , self . instance [ '<STR_LIT>' ] ) <EOL> return compute_utils . get_device_name_for_instance ( <EOL> self . instance , bdms , device ) <EOL> @ staticmethod <EOL> def _fake_bdm ( device ) : <EOL> return fake_block_device . FakeDbBlockDeviceDict ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : device , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None <EOL> } ) <EOL> def test_wrap ( self ) : <EOL> self . data = [ ] <EOL> for letter in string . ascii_lowercase [ <NUM_LIT:2> : ] : <EOL> self . data . append ( self . _fake_bdm ( '<STR_LIT>' + letter ) ) <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_wrap_plus_one ( self ) : <EOL> self . data = [ ] <EOL> for letter in string . ascii_lowercase [ <NUM_LIT:2> : ] : <EOL> self . data . append ( self . _fake_bdm ( '<STR_LIT>' + letter ) ) <EOL> self . data . append ( self . _fake_bdm ( '<STR_LIT>' ) ) <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_later ( self ) : <EOL> self . data = [ <EOL> self . _fake_bdm ( '<STR_LIT>' ) , <EOL> self . _fake_bdm ( '<STR_LIT>' ) , <EOL> self . _fake_bdm ( '<STR_LIT>' ) , <EOL> ] <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_gap ( self ) : <EOL> self . data = [ <EOL> self . _fake_bdm ( '<STR_LIT>' ) , <EOL> self . _fake_bdm ( '<STR_LIT>' ) , <EOL> ] <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_no_bdms ( self ) : <EOL> self . data = [ ] <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_lxc_names_work ( self ) : <EOL> self . instance [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . instance [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . data = [ ] <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_name_conversion ( self ) : <EOL> self . data = [ ] <EOL> device = self . _validate_device ( '<STR_LIT>' ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> device = self . _validate_device ( '<STR_LIT>' ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> device = self . _validate_device ( '<STR_LIT>' ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_invalid_device_prefix ( self ) : <EOL> self . assertRaises ( exception . InvalidDevicePath , <EOL> self . _validate_device , '<STR_LIT>' ) <EOL> def test_device_in_use ( self ) : <EOL> exc = self . assertRaises ( exception . DevicePathInUse , <EOL> self . _validate_device , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , six . text_type ( exc ) ) <EOL> def test_swap ( self ) : <EOL> self . instance [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_swap_no_ephemeral ( self ) : <EOL> self . instance . default_ephemeral_device = None <EOL> self . instance . default_swap_device = "<STR_LIT>" <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_ephemeral_xenapi ( self ) : <EOL> self . instance . flavor . ephemeral_gb = <NUM_LIT:10> <EOL> self . instance . flavor . swap = <NUM_LIT:0> <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_swap_xenapi ( self ) : <EOL> self . instance . flavor . ephemeral_gb = <NUM_LIT:0> <EOL> self . instance . flavor . swap = <NUM_LIT:10> <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_swap_and_ephemeral_xenapi ( self ) : <EOL> self . instance . flavor . ephemeral_gb = <NUM_LIT:10> <EOL> self . instance . flavor . swap = <NUM_LIT:10> <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_swap_and_one_attachment_xenapi ( self ) : <EOL> self . instance . flavor . ephemeral_gb = <NUM_LIT:0> <EOL> self . instance . flavor . swap = <NUM_LIT:10> <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> self . data . append ( self . _fake_bdm ( device ) ) <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( device , '<STR_LIT>' ) <EOL> def test_no_dev_root_device_name_get_next_name ( self ) : <EOL> self . instance [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> device = self . _validate_device ( ) <EOL> self . assertEqual ( '<STR_LIT>' , device ) <EOL> class DefaultDeviceNamesForInstanceTestCase ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( DefaultDeviceNamesForInstanceTestCase , self ) . setUp ( ) <EOL> self . context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . ephemerals = block_device_obj . block_device_make_list ( <EOL> self . context , <EOL> [ fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : uuids . block_device_instance , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:blank>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : - <NUM_LIT:1> } ) ] ) <EOL> self . swap = block_device_obj . block_device_make_list ( <EOL> self . context , <EOL> [ fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : uuids . block_device_instance , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:blank>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : - <NUM_LIT:1> } ) ] ) <EOL> self . block_device_mapping = block_device_obj . block_device_make_list ( <EOL> self . context , <EOL> [ fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:3> , <EOL> '<STR_LIT>' : uuids . block_device_instance , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } ) , <EOL> fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:4> , <EOL> '<STR_LIT>' : uuids . block_device_instance , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : - <NUM_LIT:1> } ) , <EOL> fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : uuids . block_device_instance , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:blank>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : - <NUM_LIT:1> } ) ] ) <EOL> self . instance = { '<STR_LIT>' : uuids . instance , <EOL> '<STR_LIT>' : <NUM_LIT:2> } <EOL> self . is_libvirt = False <EOL> self . root_device_name = '<STR_LIT>' <EOL> self . update_called = False <EOL> self . patchers = [ ] <EOL> self . patchers . append ( <EOL> mock . patch . object ( objects . BlockDeviceMapping , '<STR_LIT>' ) ) <EOL> for patcher in self . patchers : <EOL> patcher . start ( ) <EOL> def tearDown ( self ) : <EOL> super ( DefaultDeviceNamesForInstanceTestCase , self ) . tearDown ( ) <EOL> for patcher in self . patchers : <EOL> patcher . stop ( ) <EOL> def _test_default_device_names ( self , * block_device_lists ) : <EOL> compute_utils . default_device_names_for_instance ( self . instance , <EOL> self . root_device_name , <EOL> * block_device_lists ) <EOL> def test_only_block_device_mapping ( self ) : <EOL> original_bdm = copy . deepcopy ( self . block_device_mapping ) <EOL> self . _test_default_device_names ( [ ] , [ ] , self . block_device_mapping ) <EOL> for original , new in zip ( original_bdm , self . block_device_mapping ) : <EOL> self . assertEqual ( original . device_name , new . device_name ) <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = None <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( [ ] , [ ] , self . block_device_mapping ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] ) <EOL> def test_with_ephemerals ( self ) : <EOL> self . ephemerals [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( self . ephemerals , [ ] , <EOL> self . block_device_mapping ) <EOL> self . assertEqual ( self . ephemerals [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = None <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( self . ephemerals , [ ] , <EOL> self . block_device_mapping ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] ) <EOL> def test_with_swap ( self ) : <EOL> self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( [ ] , self . swap , [ ] ) <EOL> self . assertEqual ( self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = None <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( [ ] , self . swap , <EOL> self . block_device_mapping ) <EOL> self . assertEqual ( self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] ) <EOL> def test_all_together ( self ) : <EOL> self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( self . ephemerals , <EOL> self . swap , self . block_device_mapping ) <EOL> self . assertEqual ( self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . ephemerals [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( self . ephemerals , <EOL> self . swap , self . block_device_mapping ) <EOL> self . assertEqual ( self . ephemerals [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . ephemerals [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = None <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = None <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] = None <EOL> self . _test_default_device_names ( self . ephemerals , <EOL> self . swap , self . block_device_mapping ) <EOL> self . assertEqual ( self . ephemerals [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( self . swap [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . block_device_mapping [ <NUM_LIT:2> ] [ '<STR_LIT>' ] ) <EOL> class UsageInfoTestCase ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> def fake_get_nw_info ( cls , ctxt , instance ) : <EOL> self . assertTrue ( ctxt . is_admin ) <EOL> return fake_network . fake_get_instance_nw_info ( self , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> super ( UsageInfoTestCase , self ) . setUp ( ) <EOL> self . stubs . Set ( network_api . API , '<STR_LIT>' , <EOL> fake_get_nw_info ) <EOL> fake_notifier . stub_notifier ( self . stubs ) <EOL> self . addCleanup ( fake_notifier . reset ) <EOL> self . flags ( use_local = True , group = '<STR_LIT>' ) <EOL> self . flags ( compute_driver = '<STR_LIT>' , <EOL> network_manager = '<STR_LIT>' ) <EOL> self . compute = importutils . import_object ( CONF . compute_manager ) <EOL> self . user_id = '<STR_LIT>' <EOL> self . project_id = '<STR_LIT>' <EOL> self . context = context . RequestContext ( self . user_id , self . project_id ) <EOL> def fake_show ( meh , context , id , ** kwargs ) : <EOL> return { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> } } <EOL> self . flags ( group = '<STR_LIT>' , api_servers = [ '<STR_LIT>' ] ) <EOL> self . stubs . Set ( nova . tests . unit . image . fake . _FakeImageService , <EOL> '<STR_LIT>' , fake_show ) <EOL> fake_network . set_stub_network_methods ( self ) <EOL> fake_server_actions . stub_out_action_events ( self . stubs ) <EOL> def test_notify_usage_exists ( self ) : <EOL> instance = create_instance ( self . context ) <EOL> sys_metadata = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . system_metadata . update ( sys_metadata ) <EOL> instance . save ( ) <EOL> compute_utils . notify_usage_exists ( <EOL> rpc . get_notifier ( '<STR_LIT>' ) , self . context , instance ) <EOL> self . assertEqual ( len ( fake_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( msg . priority , '<STR_LIT>' ) <EOL> self . assertEqual ( msg . event_type , '<STR_LIT>' ) <EOL> payload = msg . payload <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , instance [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> flavor_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( flavor_id ) ) <EOL> for attr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:state>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . assertIn ( attr , payload , <EOL> "<STR_LIT>" % attr ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> image_ref_url = "<STR_LIT>" % ( glance . generate_glance_url ( ) , <EOL> uuids . fake_image_ref ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , image_ref_url ) <EOL> self . compute . terminate_instance ( self . context , instance , [ ] , [ ] ) <EOL> def test_notify_usage_exists_deleted_instance ( self ) : <EOL> instance = create_instance ( self . context ) <EOL> sys_metadata = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . system_metadata . update ( sys_metadata ) <EOL> instance . save ( ) <EOL> self . compute . terminate_instance ( self . context , instance , [ ] , [ ] ) <EOL> compute_utils . notify_usage_exists ( <EOL> rpc . get_notifier ( '<STR_LIT>' ) , self . context , instance ) <EOL> msg = fake_notifier . NOTIFICATIONS [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( msg . priority , '<STR_LIT>' ) <EOL> self . assertEqual ( msg . event_type , '<STR_LIT>' ) <EOL> payload = msg . payload <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , instance [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> flavor_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( flavor_id ) ) <EOL> for attr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:state>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . assertIn ( attr , payload , "<STR_LIT>" % attr ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> image_ref_url = "<STR_LIT>" % ( glance . generate_glance_url ( ) , <EOL> uuids . fake_image_ref ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , image_ref_url ) <EOL> def test_notify_usage_exists_instance_not_found ( self ) : <EOL> instance = create_instance ( self . context ) <EOL> self . compute . terminate_instance ( self . context , instance , [ ] , [ ] ) <EOL> compute_utils . notify_usage_exists ( <EOL> rpc . get_notifier ( '<STR_LIT>' ) , self . context , instance ) <EOL> msg = fake_notifier . NOTIFICATIONS [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( msg . priority , '<STR_LIT>' ) <EOL> self . assertEqual ( msg . event_type , '<STR_LIT>' ) <EOL> payload = msg . payload <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , instance [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> flavor_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( flavor_id ) ) <EOL> for attr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:state>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . assertIn ( attr , payload , "<STR_LIT>" % attr ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , { } ) <EOL> image_ref_url = "<STR_LIT>" % ( glance . generate_glance_url ( ) , <EOL> uuids . fake_image_ref ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , image_ref_url ) <EOL> def test_notify_about_instance_usage ( self ) : <EOL> instance = create_instance ( self . context ) <EOL> sys_metadata = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . system_metadata . update ( sys_metadata ) <EOL> instance . save ( ) <EOL> extra_usage_info = { '<STR_LIT>' : '<STR_LIT>' } <EOL> compute_utils . notify_about_instance_usage ( <EOL> rpc . get_notifier ( '<STR_LIT>' ) , <EOL> self . context , instance , '<STR_LIT>' , <EOL> extra_usage_info = extra_usage_info ) <EOL> self . assertEqual ( len ( fake_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( msg . priority , '<STR_LIT>' ) <EOL> self . assertEqual ( msg . event_type , '<STR_LIT>' ) <EOL> payload = msg . payload <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . project_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , self . user_id ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , instance [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> type_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( type_id ) ) <EOL> flavor_id = flavors . get_flavor_by_name ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( str ( payload [ '<STR_LIT>' ] ) , str ( flavor_id ) ) <EOL> for attr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:state>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . assertIn ( attr , payload , "<STR_LIT>" % attr ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> image_ref_url = "<STR_LIT>" % ( glance . generate_glance_url ( ) , <EOL> uuids . fake_image_ref ) <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , image_ref_url ) <EOL> self . compute . terminate_instance ( self . context , instance , [ ] , [ ] ) <EOL> def test_notify_about_aggregate_update_with_id ( self ) : <EOL> aggregate_payload = { '<STR_LIT>' : <NUM_LIT:1> } <EOL> compute_utils . notify_about_aggregate_update ( self . context , <EOL> "<STR_LIT>" , <EOL> aggregate_payload ) <EOL> self . assertEqual ( len ( fake_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( msg . priority , '<STR_LIT>' ) <EOL> self . assertEqual ( msg . event_type , '<STR_LIT>' ) <EOL> payload = msg . payload <EOL> self . assertEqual ( payload [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> def test_notify_about_aggregate_update_with_name ( self ) : <EOL> aggregate_payload = { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> compute_utils . notify_about_aggregate_update ( self . context , <EOL> "<STR_LIT>" , <EOL> aggregate_payload ) <EOL> self . assertEqual ( len ( fake_notifier . NOTIFICATIONS ) , <NUM_LIT:1> ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( msg . priority , '<STR_LIT>' ) <EOL> self . assertEqual ( msg . event_type , '<STR_LIT>' ) <EOL> payload = msg . payload <EOL> self . assertEqual ( payload [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> def test_notify_about_aggregate_update_without_name_id ( self ) : <EOL> aggregate_payload = { } <EOL> compute_utils . notify_about_aggregate_update ( self . context , <EOL> "<STR_LIT>" , <EOL> aggregate_payload ) <EOL> self . assertEqual ( len ( fake_notifier . NOTIFICATIONS ) , <NUM_LIT:0> ) <EOL> class ComputeUtilsGetValFromSysMetadata ( test . NoDBTestCase ) : <EOL> def test_get_value_from_system_metadata ( self ) : <EOL> instance = fake_instance . fake_instance_obj ( '<STR_LIT>' ) <EOL> system_meta = { '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT:2>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . system_metadata = system_meta <EOL> result = compute_utils . get_value_from_system_metadata ( <EOL> instance , '<STR_LIT>' , int , <NUM_LIT:0> ) <EOL> self . assertEqual ( <NUM_LIT:1> , result ) <EOL> result = compute_utils . get_value_from_system_metadata ( <EOL> instance , '<STR_LIT>' , int , <NUM_LIT:0> ) <EOL> self . assertEqual ( <NUM_LIT:2> , result ) <EOL> result = compute_utils . get_value_from_system_metadata ( <EOL> instance , '<STR_LIT>' , int , <NUM_LIT:0> ) <EOL> self . assertEqual ( <NUM_LIT:0> , result ) <EOL> class ComputeUtilsGetNWInfo ( test . NoDBTestCase ) : <EOL> def test_instance_object_none_info_cache ( self ) : <EOL> inst = fake_instance . fake_instance_obj ( '<STR_LIT>' , <EOL> expected_attrs = [ '<STR_LIT>' ] ) <EOL> self . assertIsNone ( inst . info_cache ) <EOL> result = compute_utils . get_nw_info_for_instance ( inst ) <EOL> self . assertEqual ( jsonutils . dumps ( [ ] ) , result . json ( ) ) <EOL> class ComputeUtilsGetRebootTypes ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ComputeUtilsGetRebootTypes , self ) . setUp ( ) <EOL> self . context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_get_reboot_type_started_soft ( self ) : <EOL> reboot_type = compute_utils . get_reboot_type ( task_states . REBOOT_STARTED , <EOL> power_state . RUNNING ) <EOL> self . assertEqual ( reboot_type , '<STR_LIT>' ) <EOL> def test_get_reboot_type_pending_soft ( self ) : <EOL> reboot_type = compute_utils . get_reboot_type ( task_states . REBOOT_PENDING , <EOL> power_state . RUNNING ) <EOL> self . assertEqual ( reboot_type , '<STR_LIT>' ) <EOL> def test_get_reboot_type_hard ( self ) : <EOL> reboot_type = compute_utils . get_reboot_type ( '<STR_LIT:foo>' , power_state . RUNNING ) <EOL> self . assertEqual ( reboot_type , '<STR_LIT>' ) <EOL> def test_get_reboot_not_running_hard ( self ) : <EOL> reboot_type = compute_utils . get_reboot_type ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> self . assertEqual ( reboot_type , '<STR_LIT>' ) <EOL> class ComputeUtilsTestCase ( test . NoDBTestCase ) : <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_machine_ips_value_error ( self , mock_interfaces ) : <EOL> iface = mock . sentinel <EOL> mock_interfaces . return_value = [ iface ] <EOL> with mock . patch ( '<STR_LIT>' , <EOL> side_effect = ValueError ) as mock_ifaddresses : <EOL> addresses = compute_utils . get_machine_ips ( ) <EOL> self . assertEqual ( [ ] , addresses ) <EOL> mock_ifaddresses . assert_called_once_with ( iface ) <EOL> class ComputeUtilsQuotaDeltaTestCase ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ComputeUtilsQuotaDeltaTestCase , self ) . setUp ( ) <EOL> self . context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_upsize_quota_delta ( self ) : <EOL> old_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> new_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> expected_deltas = { <EOL> '<STR_LIT>' : new_flavor [ '<STR_LIT>' ] - old_flavor [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : new_flavor [ '<STR_LIT>' ] - old_flavor [ '<STR_LIT>' ] <EOL> } <EOL> deltas = compute_utils . upsize_quota_delta ( self . context , new_flavor , <EOL> old_flavor ) <EOL> self . assertEqual ( expected_deltas , deltas ) <EOL> def test_downsize_quota_delta ( self ) : <EOL> inst = create_instance ( self . context , params = None ) <EOL> inst . old_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> inst . new_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> expected_deltas = { <EOL> '<STR_LIT>' : ( inst . new_flavor [ '<STR_LIT>' ] - <EOL> inst . old_flavor [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : ( inst . new_flavor [ '<STR_LIT>' ] - <EOL> inst . old_flavor [ '<STR_LIT>' ] ) <EOL> } <EOL> deltas = compute_utils . downsize_quota_delta ( self . context , inst ) <EOL> self . assertEqual ( expected_deltas , deltas ) <EOL> def test_reverse_quota_delta ( self ) : <EOL> inst = create_instance ( self . context , params = None ) <EOL> inst . old_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> inst . new_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> expected_deltas = { <EOL> '<STR_LIT>' : - <NUM_LIT:1> * ( inst . new_flavor [ '<STR_LIT>' ] - <EOL> inst . old_flavor [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : - <NUM_LIT:1> * ( inst . new_flavor [ '<STR_LIT>' ] - <EOL> inst . old_flavor [ '<STR_LIT>' ] ) <EOL> } <EOL> deltas = compute_utils . reverse_upsize_quota_delta ( self . context , inst ) <EOL> self . assertEqual ( expected_deltas , deltas ) <EOL> @ mock . patch . object ( objects . Quotas , '<STR_LIT>' ) <EOL> @ mock . patch . object ( objects . quotas , '<STR_LIT>' ) <EOL> def test_reserve_quota_delta ( self , mock_ids_from_instance , <EOL> mock_reserve ) : <EOL> quotas = objects . Quotas ( context = context ) <EOL> inst = create_instance ( self . context , params = None ) <EOL> inst . old_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> inst . new_flavor = flavors . get_flavor_by_name ( '<STR_LIT>' ) <EOL> mock_ids_from_instance . return_value = ( inst . project_id , inst . user_id ) <EOL> mock_reserve . return_value = quotas <EOL> deltas = compute_utils . upsize_quota_delta ( self . context , <EOL> inst . new_flavor , <EOL> inst . old_flavor ) <EOL> compute_utils . reserve_quota_delta ( self . context , deltas , inst ) <EOL> mock_reserve . assert_called_once_with ( project_id = inst . project_id , <EOL> user_id = inst . user_id , ** deltas ) </s>
