本文整理汇总了Python中nova.compute.utils.notify_about_instance_usage函数的典型用法代码示例。如果您正苦于以下问题：Python notify_about_instance_usage函数的具体用法？Python notify_about_instance_usage怎么用？Python notify_about_instance_usage使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了notify_about_instance_usage函数的12个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_notify_usage_exists
def test_notify_usage_exists(self):
info = {
"audit_period_beginning": "start",
"audit_period_ending": "end",
"bandwidth": "bw_usage",
"image_meta": {},
"extra": "info",
}
instance = {"system_metadata": []}
self.mox.StubOutWithMock(notifications, "audit_period_bounds")
self.mox.StubOutWithMock(notifications, "bandwidth_usage")
self.mox.StubOutWithMock(compute_utils, "notify_about_instance_usage")
notifications.audit_period_bounds(False).AndReturn(("start", "end"))
notifications.bandwidth_usage(instance, "start", True).AndReturn("bw_usage")
compute_utils.notify_about_instance_usage(
self.context, instance, "exists", system_metadata={}, extra_usage_info=info
)
self.mox.ReplayAll()
self.conductor.notify_usage_exists(
self.context, instance, system_metadata={}, extra_usage_info=dict(extra="info")
)
开发者ID:jdurgin，项目名称:nova，代码行数:25，代码来源:test_conductor.py
示例2: test_notify_usage_exists
def test_notify_usage_exists(self):
info = {
'audit_period_beginning': 'start',
'audit_period_ending': 'end',
'bandwidth': 'bw_usage',
'image_meta': {},
'extra': 'info',
}
instance = {
'system_metadata': [],
}
self.mox.StubOutWithMock(notifications, 'audit_period_bounds')
self.mox.StubOutWithMock(notifications, 'bandwidth_usage')
self.mox.StubOutWithMock(compute_utils, 'notify_about_instance_usage')
notifications.audit_period_bounds(False).AndReturn(('start', 'end'))
notifications.bandwidth_usage(instance, 'start', True).AndReturn(
'bw_usage')
compute_utils.notify_about_instance_usage(self.context, instance,
'exists',
system_metadata={},
extra_usage_info=info)
self.mox.ReplayAll()
self.conductor.notify_usage_exists(self.context, instance,
system_metadata={},
extra_usage_info=dict(extra='info'))
开发者ID:gminator，项目名称:nova，代码行数:29，代码来源:test_conductor.py
示例3: test_notify_about_instance_usage
def test_notify_about_instance_usage(self):
instance = create_instance(self.context)
# Set some system metadata
sys_metadata = {"image_md_key1": "val1", "image_md_key2": "val2", "other_data": "meow"}
instance.system_metadata.update(sys_metadata)
instance.save()
extra_usage_info = {"image_name": "fake_name"}
compute_utils.notify_about_instance_usage(
rpc.get_notifier("compute"), self.context, instance, "create.start", extra_usage_info=extra_usage_info
)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)
msg = fake_notifier.NOTIFICATIONS[0]
self.assertEqual(msg.priority, "INFO")
self.assertEqual(msg.event_type, "compute.instance.create.start")
payload = msg.payload
self.assertEqual(payload["tenant_id"], self.project_id)
self.assertEqual(payload["user_id"], self.user_id)
self.assertEqual(payload["instance_id"], instance["uuid"])
self.assertEqual(payload["instance_type"], "m1.tiny")
type_id = flavors.get_flavor_by_name("m1.tiny")["id"]
self.assertEqual(str(payload["instance_type_id"]), str(type_id))
flavor_id = flavors.get_flavor_by_name("m1.tiny")["flavorid"]
self.assertEqual(str(payload["instance_flavor_id"]), str(flavor_id))
for attr in ("display_name", "created_at", "launched_at", "state", "state_description", "image_meta"):
self.assertIn(attr, payload, "Key %s not in payload" % attr)
self.assertEqual(payload["image_meta"], {"md_key1": "val1", "md_key2": "val2"})
self.assertEqual(payload["image_name"], "fake_name")
image_ref_url = "%s/images/1" % glance.generate_glance_url()
self.assertEqual(payload["image_ref_url"], image_ref_url)
self.compute.terminate_instance(self.context, instance, [], [])
开发者ID:jorgevgut，项目名称:nova，代码行数:30，代码来源:test_compute_utils.py
示例4: test_notify_about_instance_usage
def test_notify_about_instance_usage(self):
instance = create_instance(self.context)
# Set some system metadata
sys_metadata = {'image_md_key1': 'val1',
'image_md_key2': 'val2',
'other_data': 'meow'}
instance.system_metadata.update(sys_metadata)
instance.save()
extra_usage_info = {'image_name': 'fake_name'}
compute_utils.notify_about_instance_usage(
rpc.get_notifier('compute'),
self.context, instance, 'create.start',
extra_usage_info=extra_usage_info)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)
msg = fake_notifier.NOTIFICATIONS[0]
self.assertEqual(msg.priority, 'INFO')
self.assertEqual(msg.event_type, 'compute.instance.create.start')
payload = msg.payload
self.assertEqual(payload['tenant_id'], self.project_id)
self.assertEqual(payload['user_id'], self.user_id)
self.assertEqual(payload['instance_id'], instance['uuid'])
self.assertEqual(payload['instance_type'], 'm1.tiny')
type_id = flavors.get_flavor_by_name('m1.tiny')['id']
self.assertEqual(str(payload['instance_type_id']), str(type_id))
flavor_id = flavors.get_flavor_by_name('m1.tiny')['flavorid']
self.assertEqual(str(payload['instance_flavor_id']), str(flavor_id))
for attr in ('display_name', 'created_at', 'launched_at',
'state', 'state_description', 'image_meta'):
self.assertIn(attr, payload, "Key %s not in payload" % attr)
self.assertEqual(payload['image_meta'],
{'md_key1': 'val1', 'md_key2': 'val2'})
self.assertEqual(payload['image_name'], 'fake_name')
image_ref_url = "%s/images/1" % glance.generate_glance_url()
self.assertEqual(payload['image_ref_url'], image_ref_url)
self.compute.terminate_instance(self.context, instance, [], [])
开发者ID:billhu422，项目名称:nova，代码行数:35，代码来源:test_compute_utils.py
示例5: test_notify_about_instance_usage
def test_notify_about_instance_usage(self):
instance_id = self._create_instance()
instance = db.instance_get(self.context, instance_id)
# Set some system metadata
sys_metadata = {'image_md_key1': 'val1',
'image_md_key2': 'val2',
'other_data': 'meow'}
extra_usage_info = {'image_name': 'fake_name'}
db.instance_system_metadata_update(self.context, instance['uuid'],
sys_metadata, False)
compute_utils.notify_about_instance_usage(self.context, instance,
'create.start', extra_usage_info=extra_usage_info)
self.assertEquals(len(test_notifier.NOTIFICATIONS), 1)
msg = test_notifier.NOTIFICATIONS[0]
self.assertEquals(msg['priority'], 'INFO')
self.assertEquals(msg['event_type'], 'compute.instance.create.start')
payload = msg['payload']
self.assertEquals(payload['tenant_id'], self.project_id)
self.assertEquals(payload['user_id'], self.user_id)
self.assertEquals(payload['instance_id'], instance.uuid)
self.assertEquals(payload['instance_type'], 'm1.tiny')
type_id = instance_types.get_instance_type_by_name('m1.tiny')['id']
self.assertEquals(str(payload['instance_type_id']), str(type_id))
for attr in ('display_name', 'created_at', 'launched_at',
'state', 'state_description', 'image_meta'):
self.assertTrue(attr in payload,
msg="Key %s not in payload" % attr)
self.assertEquals(payload['image_meta'],
{'md_key1': 'val1', 'md_key2': 'val2'})
self.assertEquals(payload['image_name'], 'fake_name')
image_ref_url = "%s/images/1" % utils.generate_glance_url()
self.assertEquals(payload['image_ref_url'], image_ref_url)
self.compute.terminate_instance(self.context, instance)
开发者ID:EE-NovRain，项目名称:nova，代码行数:33，代码来源:test_compute_utils.py
示例6: test_notify_about_instance_usage
def test_notify_about_instance_usage(self):
instance_id = self._create_instance()
instance = db.instance_get(self.context, instance_id)
# Set some system metadata
sys_metadata = {"image_md_key1": "val1", "image_md_key2": "val2", "other_data": "meow"}
extra_usage_info = {"image_name": "fake_name"}
db.instance_system_metadata_update(self.context, instance["uuid"], sys_metadata, False)
# NOTE(russellb) Make sure our instance has the latest system_metadata
# in it.
instance = db.instance_get(self.context, instance_id)
compute_utils.notify_about_instance_usage(
notify.get_notifier("compute"), self.context, instance, "create.start", extra_usage_info=extra_usage_info
)
self.assertEquals(len(fake_notifier.NOTIFICATIONS), 1)
msg = fake_notifier.NOTIFICATIONS[0]
self.assertEquals(msg.priority, "INFO")
self.assertEquals(msg.event_type, "compute.instance.create.start")
payload = msg.payload
self.assertEquals(payload["tenant_id"], self.project_id)
self.assertEquals(payload["user_id"], self.user_id)
self.assertEquals(payload["instance_id"], instance["uuid"])
self.assertEquals(payload["instance_type"], "m1.tiny")
type_id = flavors.get_flavor_by_name("m1.tiny")["id"]
self.assertEquals(str(payload["instance_type_id"]), str(type_id))
flavor_id = flavors.get_flavor_by_name("m1.tiny")["flavorid"]
self.assertEquals(str(payload["instance_flavor_id"]), str(flavor_id))
for attr in ("display_name", "created_at", "launched_at", "state", "state_description", "image_meta"):
self.assertTrue(attr in payload, msg="Key %s not in payload" % attr)
self.assertEquals(payload["image_meta"], {"md_key1": "val1", "md_key2": "val2"})
self.assertEquals(payload["image_name"], "fake_name")
image_ref_url = "%s/images/1" % glance.generate_glance_url()
self.assertEquals(payload["image_ref_url"], image_ref_url)
self.compute.terminate_instance(self.context, jsonutils.to_primitive(instance))
开发者ID:kraman，项目名称:nova，代码行数:33，代码来源:test_compute_utils.py
示例7: rebuild_instance
def rebuild_instance(self, context, instance, orig_image_ref, image_ref,
injected_files, new_pass, orig_sys_metadata,
bdms, recreate, on_shared_storage,
preserve_ephemeral=False, host=None):
with compute_utils.EventReporter(context, 'rebuild_server',
instance.uuid):
if not host:
# NOTE(lcostantino): Retrieve scheduler filters for the
# instance when the feature is available
filter_properties = {'ignore_hosts': [instance.host]}
request_spec = scheduler_utils.build_request_spec(context,
image_ref,
[instance])
try:
scheduler_utils.setup_instance_group(context, request_spec,
filter_properties)
hosts = self.scheduler_client.select_destinations(context,
request_spec,
filter_properties)
host = hosts.pop(0)['host']
except exception.NoValidHost as ex:
with excutils.save_and_reraise_exception():
self._set_vm_state_and_notify(context, instance.uuid,
'rebuild_server',
{'vm_state': instance.vm_state,
'task_state': None}, ex, request_spec)
LOG.warning(_LW("No valid host found for rebuild"),
instance=instance)
except exception.UnsupportedPolicyException as ex:
with excutils.save_and_reraise_exception():
self._set_vm_state_and_notify(context, instance.uuid,
'rebuild_server',
{'vm_state': instance.vm_state,
'task_state': None}, ex, request_spec)
LOG.warning(_LW("Server with unsupported policy "
"cannot be rebuilt"),
instance=instance)
compute_utils.notify_about_instance_usage(
self.notifier, context, instance, "rebuild.scheduled")
self.compute_rpcapi.rebuild_instance(context,
instance=instance,
new_pass=new_pass,
injected_files=injected_files,
image_ref=image_ref,
orig_image_ref=orig_image_ref,
orig_sys_metadata=orig_sys_metadata,
bdms=bdms,
recreate=recreate,
on_shared_storage=on_shared_storage,
preserve_ephemeral=preserve_ephemeral,
host=host)
开发者ID:tomatoXu，项目名称:nova，代码行数:54，代码来源:manager.py
示例8: test_notify_about_instance_usage
def test_notify_about_instance_usage(self):
instance_id = self._create_instance()
instance = db.instance_get(self.context, instance_id)
# Set some system metadata
sys_metadata = {'image_md_key1': 'val1',
'image_md_key2': 'val2',
'other_data': 'meow'}
extra_usage_info = {'image_name': 'fake_name'}
db.instance_system_metadata_update(self.context, instance['uuid'],
sys_metadata, False)
# NOTE(russellb) Make sure our instance has the latest system_metadata
# in it.
instance = db.instance_get(self.context, instance_id)
compute_utils.notify_about_instance_usage(
notify.get_notifier('compute'),
self.context, instance, 'create.start',
extra_usage_info=extra_usage_info)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)
msg = fake_notifier.NOTIFICATIONS[0]
self.assertEqual(msg.priority, 'INFO')
self.assertEqual(msg.event_type, 'compute.instance.create.start')
payload = msg.payload
self.assertEqual(payload['tenant_id'], self.project_id)
self.assertEqual(payload['user_id'], self.user_id)
self.assertEqual(payload['instance_id'], instance['uuid'])
self.assertEqual(payload['instance_type'], 'm1.tiny')
type_id = flavors.get_flavor_by_name('m1.tiny')['id']
self.assertEqual(str(payload['instance_type_id']), str(type_id))
flavor_id = flavors.get_flavor_by_name('m1.tiny')['flavorid']
self.assertEqual(str(payload['instance_flavor_id']), str(flavor_id))
for attr in ('display_name', 'created_at', 'launched_at',
'state', 'state_description', 'image_meta'):
self.assertTrue(attr in payload,
msg="Key %s not in payload" % attr)
self.assertEqual(payload['image_meta'],
{'md_key1': 'val1', 'md_key2': 'val2'})
self.assertEqual(payload['image_name'], 'fake_name')
image_ref_url = "%s/images/1" % glance.generate_glance_url()
self.assertEqual(payload['image_ref_url'], image_ref_url)
self.compute.terminate_instance(self.context,
jsonutils.to_primitive(instance))
开发者ID:674009287，项目名称:nova，代码行数:41，代码来源:test_compute_utils.py
示例9: rebuild_instance
def rebuild_instance(self, context, instance, orig_image_ref, image_ref,
injected_files, new_pass, orig_sys_metadata,
bdms, recreate, on_shared_storage,
preserve_ephemeral=False, host=None):
with compute_utils.EventReporter(context, 'rebuild_server',
instance.uuid):
node = limits = None
if not host:
# NOTE(lcostantino): Retrieve scheduler filters for the
# instance when the feature is available
filter_properties = {'ignore_hosts': [instance.host]}
try:
request_spec = scheduler_utils.build_request_spec(
context, image_ref, [instance])
hosts = self._schedule_instances(
context, request_spec, filter_properties)
host_dict = hosts.pop(0)
host, node, limits = (host_dict['host'],
host_dict['nodename'],
host_dict['limits'])
except exception.NoValidHost as ex:
with excutils.save_and_reraise_exception():
self._set_vm_state_and_notify(context, instance.uuid,
'rebuild_server',
{'vm_state': instance.vm_state,
'task_state': None}, ex, request_spec)
LOG.warning(_LW("No valid host found for rebuild"),
instance=instance)
except exception.UnsupportedPolicyException as ex:
with excutils.save_and_reraise_exception():
self._set_vm_state_and_notify(context, instance.uuid,
'rebuild_server',
{'vm_state': instance.vm_state,
'task_state': None}, ex, request_spec)
LOG.warning(_LW("Server with unsupported policy "
"cannot be rebuilt"),
instance=instance)
try:
migration = objects.Migration.get_by_instance_and_status(
context, instance.uuid, 'accepted')
except exception.MigrationNotFoundByStatus:
LOG.debug("No migration record for the rebuild/evacuate "
"request.", instance=instance)
migration = None
compute_utils.notify_about_instance_usage(
self.notifier, context, instance, "rebuild.scheduled")
self.compute_rpcapi.rebuild_instance(context,
instance=instance,
new_pass=new_pass,
injected_files=injected_files,
image_ref=image_ref,
orig_image_ref=orig_image_ref,
orig_sys_metadata=orig_sys_metadata,
bdms=bdms,
recreate=recreate,
on_shared_storage=on_shared_storage,
preserve_ephemeral=preserve_ephemeral,
migration=migration,
host=host, node=node, limits=limits)
开发者ID:mnaser，项目名称:nova-1，代码行数:63，代码来源:manager.py
示例10: rebuild_instance
def rebuild_instance(self, context, instance, orig_image_ref, image_ref,
injected_files, new_pass, orig_sys_metadata,
bdms, recreate, on_shared_storage,
preserve_ephemeral=False, host=None,
request_spec=None):
with compute_utils.EventReporter(context, 'rebuild_server',
instance.uuid):
node = limits = None
if not host:
if not request_spec:
# NOTE(sbauza): We were unable to find an original
# RequestSpec object - probably because the instance is old
# We need to mock that the old way
filter_properties = {'ignore_hosts': [instance.host]}
request_spec = scheduler_utils.build_request_spec(
context, image_ref, [instance])
else:
# NOTE(sbauza): Augment the RequestSpec object by excluding
# the source host for avoiding the scheduler to pick it
request_spec.ignore_hosts = request_spec.ignore_hosts or []
request_spec.ignore_hosts.append(instance.host)
# NOTE(sbauza): Force_hosts/nodes needs to be reset
# if we want to make sure that the next destination
# is not forced to be the original host
request_spec.reset_forced_destinations()
# TODO(sbauza): Provide directly the RequestSpec object
# when _schedule_instances() and _set_vm_state_and_notify()
# accept it
filter_properties = request_spec.\
to_legacy_filter_properties_dict()
request_spec = request_spec.to_legacy_request_spec_dict()
try:
hosts = self._schedule_instances(
context, request_spec, filter_properties)
host_dict = hosts.pop(0)
host, node, limits = (host_dict['host'],
host_dict['nodename'],
host_dict['limits'])
except exception.NoValidHost as ex:
with excutils.save_and_reraise_exception():
self._set_vm_state_and_notify(context, instance.uuid,
'rebuild_server',
{'vm_state': instance.vm_state,
'task_state': None}, ex, request_spec)
LOG.warning(_LW("No valid host found for rebuild"),
instance=instance)
except exception.UnsupportedPolicyException as ex:
with excutils.save_and_reraise_exception():
self._set_vm_state_and_notify(context, instance.uuid,
'rebuild_server',
{'vm_state': instance.vm_state,
'task_state': None}, ex, request_spec)
LOG.warning(_LW("Server with unsupported policy "
"cannot be rebuilt"),
instance=instance)
try:
migration = objects.Migration.get_by_instance_and_status(
context, instance.uuid, 'accepted')
except exception.MigrationNotFoundByStatus:
LOG.debug("No migration record for the rebuild/evacuate "
"request.", instance=instance)
migration = None
compute_utils.notify_about_instance_usage(
self.notifier, context, instance, "rebuild.scheduled")
self.compute_rpcapi.rebuild_instance(context,
instance=instance,
new_pass=new_pass,
injected_files=injected_files,
image_ref=image_ref,
orig_image_ref=orig_image_ref,
orig_sys_metadata=orig_sys_metadata,
bdms=bdms,
recreate=recreate,
on_shared_storage=on_shared_storage,
preserve_ephemeral=preserve_ephemeral,
migration=migration,
host=host, node=node, limits=limits)
开发者ID:JerryDog，项目名称:nova，代码行数:81，代码来源:manager.py
示例11: _test_delete
def _test_delete(self, delete_type, **attrs):
inst = self._create_instance_obj()
inst.update(attrs)
inst._context = self.context
delete_time = datetime.datetime(1955, 11, 5, 9, 30,
tzinfo=iso8601.iso8601.Utc())
timeutils.set_time_override(delete_time)
task_state = (delete_type == 'soft_delete' and
task_states.SOFT_DELETING or task_states.DELETING)
db_inst = obj_base.obj_to_primitive(inst)
updates = {'progress': 0, 'task_state': task_state}
if delete_type == 'soft_delete':
updates['deleted_at'] = delete_time
self.mox.StubOutWithMock(inst, 'save')
self.mox.StubOutWithMock(db,
'block_device_mapping_get_all_by_instance')
self.mox.StubOutWithMock(self.compute_api, '_create_reservations')
self.mox.StubOutWithMock(self.context, 'elevated')
self.mox.StubOutWithMock(db, 'service_get_by_compute_host')
self.mox.StubOutWithMock(self.compute_api.servicegroup_api,
'service_is_up')
self.mox.StubOutWithMock(db, 'migration_get_by_instance_and_status')
self.mox.StubOutWithMock(self.compute_api, '_downsize_quota_delta')
self.mox.StubOutWithMock(self.compute_api, '_reserve_quota_delta')
self.mox.StubOutWithMock(self.compute_api, '_record_action_start')
self.mox.StubOutWithMock(db, 'instance_update_and_get_original')
self.mox.StubOutWithMock(db, 'instance_info_cache_delete')
self.mox.StubOutWithMock(self.compute_api.network_api,
'deallocate_for_instance')
self.mox.StubOutWithMock(db, 'instance_system_metadata_get')
self.mox.StubOutWithMock(db, 'instance_destroy')
self.mox.StubOutWithMock(compute_utils,
'notify_about_instance_usage')
self.mox.StubOutWithMock(quota.QUOTAS, 'commit')
db.block_device_mapping_get_all_by_instance(
self.context, inst.uuid).AndReturn([])
inst.save()
self.compute_api._create_reservations(
self.context, inst, inst.instance_type_id, inst.project_id,
inst.user_id).AndReturn('fake-resv')
if inst.vm_state == vm_states.RESIZED:
self._test_delete_resized_part(inst)
self.context.elevated().MultipleTimes().AndReturn(self.context)
db.service_get_by_compute_host(self.context, inst.host).AndReturn(
'fake-service')
self.compute_api.servicegroup_api.service_is_up(
'fake-service').AndReturn(inst.host != 'down-host')
if self.is_cells:
rpcapi = self.compute_api.cells_rpcapi
else:
rpcapi = self.compute_api.compute_rpcapi
self.mox.StubOutWithMock(rpcapi, 'terminate_instance')
self.mox.StubOutWithMock(rpcapi, 'soft_delete_instance')
if inst.host == 'down-host':
db.instance_info_cache_delete(self.context, inst.uuid)
compute_utils.notify_about_instance_usage(self.context,
inst,
'%s.start' % delete_type)
self.compute_api.network_api.deallocate_for_instance(
self.context, inst)
db.instance_system_metadata_get(self.context, inst.uuid
).AndReturn('sys-meta')
state = ('soft' in delete_type and vm_states.SOFT_DELETED or
vm_states.DELETED)
updates.update({'vm_state': state,
'task_state': None,
'terminated_at': delete_time})
inst.save()
if self.is_cells:
if delete_type == 'soft_delete':
rpcapi.soft_delete_instance(self.context, inst,
reservations=None)
else:
rpcapi.terminate_instance(self.context, inst, [],
reservations=None)
db.instance_destroy(self.context, inst.uuid)
compute_utils.notify_about_instance_usage(
self.context, inst, '%s.end' % delete_type,
system_metadata='sys-meta')
if inst.host == 'down-host':
quota.QUOTAS.commit(self.context, 'fake-resv',
project_id=inst.project_id,
user_id=inst.user_id)
elif delete_type == 'soft_delete':
self.compute_api._record_action_start(self.context, inst,
instance_actions.DELETE)
rpcapi.soft_delete_instance(self.context, inst,
reservations='fake-resv')
elif delete_type in ['delete', 'force_delete']:
self.compute_api._record_action_start(self.context, inst,
instance_actions.DELETE)
rpcapi.terminate_instance(self.context, inst, [],
reservations='fake-resv')
#.........这里部分代码省略.........
开发者ID:raidwang，项目名称:nova，代码行数:101，代码来源:test_compute_api.py
示例12: _test_delete
def _test_delete(self, delete_type, **attrs):
inst = self._create_instance_obj()
inst.update(attrs)
delete_time = datetime.datetime(1955, 11, 5, 9, 30)
timeutils.set_time_override(delete_time)
task_state = (delete_type == 'soft_delete' and
task_states.SOFT_DELETING or task_states.DELETING)
db_inst = obj_base.obj_to_primitive(inst)
updates = {'progress': 0, 'task_state': task_state}
if delete_type == 'soft_delete':
updates['deleted_at'] = delete_time
new_inst = dict(db_inst, **updates)
self.mox.StubOutWithMock(db,
'block_device_mapping_get_all_by_instance')
self.mox.StubOutWithMock(self.compute_api, '_create_reservations')
self.mox.StubOutWithMock(self.context, 'elevated')
self.mox.StubOutWithMock(db, 'service_get_by_compute_host')
self.mox.StubOutWithMock(self.compute_api.servicegroup_api,
'service_is_up')
self.mox.StubOutWithMock(db, 'migration_get_by_instance_and_status')
self.mox.StubOutWithMock(self.compute_api, '_downsize_quota_delta')
self.mox.StubOutWithMock(self.compute_api, '_reserve_quota_delta')
self.mox.StubOutWithMock(self.compute_api, '_record_action_start')
self.mox.StubOutWithMock(db, 'instance_update_and_get_original')
self.mox.StubOutWithMock(db, 'instance_info_cache_delete')
self.mox.StubOutWithMock(self.compute_api.network_api,
'deallocate_for_instance')
self.mox.StubOutWithMock(db, 'instance_system_metadata_get')
self.mox.StubOutWithMock(db, 'instance_destroy')
self.mox.StubOutWithMock(compute_utils,
'notify_about_instance_usage')
self.mox.StubOutWithMock(quota.QUOTAS, 'commit')
self.mox.StubOutWithMock(self.compute_api.compute_rpcapi,
'terminate_instance')
self.mox.StubOutWithMock(self.compute_api.compute_rpcapi,
'soft_delete_instance')
db.block_device_mapping_get_all_by_instance(
self.context, inst.uuid).AndReturn([])
db.instance_update_and_get_original(
self.context, inst.uuid, updates).AndReturn((db_inst, new_inst))
self.compute_api._create_reservations(
self.context, db_inst, new_inst, inst.project_id, inst.user_id
).AndReturn('fake-resv')
if inst.vm_state == vm_states.RESIZED:
self._test_delete_resized_part(db_inst)
self.context.elevated().MultipleTimes().AndReturn(self.context)
db.service_get_by_compute_host(self.context, inst.host).AndReturn(
'fake-service')
self.compute_api.servicegroup_api.service_is_up(
'fake-service').AndReturn(inst.host != 'down-host')
if inst.host == 'down-host' and (
not self.is_cells or not inst.cell_name):
db.instance_info_cache_delete(self.context, inst.uuid)
compute_utils.notify_about_instance_usage(self.context,
db_inst, 'delete.start')
self.compute_api.network_api.deallocate_for_instance(
self.context, db_inst)
db.instance_system_metadata_get(self.context, inst.uuid
).AndReturn('sys-meta')
updates = {'vm_state': vm_states.DELETED,
'task_state': None,
'terminated_at': delete_time}
del_inst = dict(new_inst, **updates)
db.instance_update_and_get_original(
self.context, inst.uuid, updates
).AndReturn((db_inst, del_inst))
db.instance_destroy(self.context, inst.uuid)
compute_utils.notify_about_instance_usage(
self.context, del_inst, 'delete.end',
system_metadata='sys-meta')
if inst.host == 'down-host':
quota.QUOTAS.commit(self.context, 'fake-resv',
project_id=inst.project_id,
user_id=inst.user_id)
elif delete_type == 'soft_delete':
self.compute_api._record_action_start(self.context, db_inst,
instance_actions.DELETE)
self.compute_api.compute_rpcapi.soft_delete_instance(
self.context, db_inst, reservations='fake-resv')
elif delete_type in ['delete', 'force_delete']:
self.compute_api._record_action_start(self.context, db_inst,
instance_actions.DELETE)
self.compute_api.compute_rpcapi.terminate_instance(
self.context, db_inst, [], reservations='fake-resv')
if self.is_cells:
self.mox.StubOutWithMock(self.compute_api, '_cast_to_cells')
self.compute_api._cast_to_cells(
self.context, db_inst, delete_type)
self.mox.ReplayAll()
getattr(self.compute_api, delete_type)(self.context, db_inst)
开发者ID:xqueralt，项目名称:nova，代码行数:97，代码来源:test_compute_api.py
注：本文中的nova.compute.utils.notify_about_instance_usage函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论