# Copyright 2010 OpenStack Foundation # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import datetime from oslo_serialization import jsonutils from oslo_utils import timeutils from oslo_utils import uuidutils import routes import webob.dec from nova.api import auth as api_auth from nova.api import openstack as openstack_api from nova.api.openstack import api_version_request as api_version from nova.api.openstack import compute from nova.api.openstack.compute import versions from nova.api.openstack import urlmap from nova.api.openstack import wsgi as os_wsgi from nova.api import wsgi from nova.compute import flavors from nova.compute import vm_states import nova.conf from nova import context from nova.db.main import models from nova import exception as exc from nova import objects from nova.objects import base from nova import quota from nova.tests.unit import fake_block_device from nova.tests.unit.objects import test_keypair from nova import utils CONF = nova.conf.CONF QUOTAS = quota.QUOTAS FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' FAKE_PROJECT_ID = '6a6a9c9eee154e9cb8cec487b98d36ab' FAKE_USER_ID = '5fae60f5cf4642609ddd31f71748beac' FAKE_UUIDS = {} @webob.dec.wsgify def fake_wsgi(self, req): return self.application def wsgi_app_v21(fake_auth_context=None, v2_compatible=False, custom_routes=None): # NOTE(efried): Keep this (roughly) in sync with api-paste.ini def wrap(app, use_context=False): if v2_compatible: app = openstack_api.LegacyV2CompatibleWrapper(app) if use_context: if fake_auth_context is not None: ctxt = fake_auth_context else: ctxt = context.RequestContext( 'fake', FAKE_PROJECT_ID, auth_token=True) app = api_auth.InjectContext(ctxt, app) app = openstack_api.FaultWrapper(app) return app inner_app_v21 = compute.APIRouterV21(custom_routes=custom_routes) mapper = urlmap.URLMap() mapper['/'] = wrap(versions.Versions()) mapper['/v2'] = wrap(versions.VersionsV2()) mapper['/v2.1'] = wrap(versions.VersionsV2()) mapper['/v2/+'] = wrap(inner_app_v21, use_context=True) mapper['/v2.1/+'] = wrap(inner_app_v21, use_context=True) return mapper def stub_out_key_pair_funcs(testcase, have_key_pair=True, **kwargs): def key_pair(context, user_id): return [dict(test_keypair.fake_keypair, name='key', public_key='public_key', **kwargs)] def one_key_pair(context, user_id, name): if name in ['key', 'new-key']: return dict(test_keypair.fake_keypair, name=name, public_key='public_key', **kwargs) else: raise exc.KeypairNotFound(user_id=user_id, name=name) def no_key_pair(context, user_id): return [] if have_key_pair: testcase.stub_out( 'nova.db.main.api.key_pair_get_all_by_user', key_pair) testcase.stub_out('nova.db.main.api.key_pair_get', one_key_pair) else: testcase.stub_out( 'nova.db.main.api.key_pair_get_all_by_user', no_key_pair) def stub_out_instance_quota(test, allowed, quota, resource='instances'): def fake_reserve(context, **deltas): requested = deltas.pop(resource, 0) if requested > allowed: quotas = dict(instances=1, cores=1, ram=1) quotas[resource] = quota usages = dict(instances=dict(in_use=0, reserved=0), cores=dict(in_use=0, reserved=0), ram=dict(in_use=0, reserved=0)) usages[resource]['in_use'] = (quotas[resource] * 9 // 10 - allowed) usages[resource]['reserved'] = quotas[resource] // 10 raise exc.OverQuota(overs=[resource], quotas=quotas, usages=usages) test.stub_out('nova.quota.QUOTAS.reserve', fake_reserve) def stub_out_networking(test): def get_my_ip(): return '127.0.0.1' test.stub_out('oslo_utils.netutils.get_my_ipv4', get_my_ip) def stub_out_compute_api_snapshot(test): def snapshot(self, context, instance, name, extra_properties=None): # emulate glance rejecting image names which are too long if len(name) > 256: raise exc.Invalid return dict(id='123', status='ACTIVE', name=name, properties=extra_properties) test.stub_out('nova.compute.api.API.snapshot', snapshot) class stub_out_compute_api_backup(object): def __init__(self, test): self.extra_props_last_call = None test.stub_out('nova.compute.api.API.backup', self.backup) def backup(self, context, instance, name, backup_type, rotation, extra_properties=None): self.extra_props_last_call = extra_properties props = dict(backup_type=backup_type, rotation=rotation) props.update(extra_properties or {}) return dict(id='123', status='ACTIVE', name=name, properties=props) def stub_out_nw_api(test, cls=None, private=None, publics=None): if not private: private = '192.168.0.3' if not publics: publics = ['1.2.3.4'] class Fake(object): def __init__(self): pass def get_instance_nw_info(*args, **kwargs): pass def get_floating_ips_by_fixed_address(*args, **kwargs): return publics def validate_networks(self, context, networks, max_count): return max_count def create_resource_requests( self, context, requested_networks, pci_requests=None, affinity_policy=None): return None, [], objects.RequestLevelParams() if cls is None: cls = Fake test.stub_out('nova.network.neutron.API', cls) def stub_out_secgroup_api(test, security_groups=None): def get_instances_security_groups_bindings( context, servers, detailed=False): instances_security_group_bindings = {} if servers: # we don't get security group information for down cells instances_security_group_bindings = { server['id']: security_groups or [] for server in servers if server['status'] != 'UNKNOWN' } return instances_security_group_bindings def get_instance_security_groups(context, instance, detailed=False): return security_groups if security_groups is not None else [] test.stub_out( 'nova.network.security_group_api' '.get_instances_security_groups_bindings', get_instances_security_groups_bindings) test.stub_out( 'nova.network.security_group_api.get_instance_security_groups', get_instance_security_groups) class FakeToken(object): id_count = 0 def __getitem__(self, key): return getattr(self, key) def __init__(self, **kwargs): FakeToken.id_count += 1 self.id = FakeToken.id_count for k, v in kwargs.items(): setattr(self, k, v) class FakeRequestContext(context.RequestContext): def __init__(self, *args, **kwargs): kwargs['auth_token'] = kwargs.get('auth_token', 'fake_auth_token') super(FakeRequestContext, self).__init__(*args, **kwargs) class HTTPRequest(os_wsgi.Request): @classmethod def blank(cls, *args, **kwargs): defaults = {'base_url': 'http://localhost/v2'} use_admin_context = kwargs.pop('use_admin_context', False) project_id = kwargs.pop('project_id', FAKE_PROJECT_ID) version = kwargs.pop('version', os_wsgi.DEFAULT_API_VERSION) defaults.update(kwargs) out = super(HTTPRequest, cls).blank(*args, **defaults) out.environ['nova.context'] = FakeRequestContext( user_id='fake_user', project_id=project_id, is_admin=use_admin_context) out.api_version_request = api_version.APIVersionRequest(version) return out class HTTPRequestV21(HTTPRequest): pass class TestRouter(wsgi.Router): def __init__(self, controller, mapper=None): if not mapper: mapper = routes.Mapper() mapper.resource("test", "tests", controller=os_wsgi.Resource(controller)) super(TestRouter, self).__init__(mapper) class FakeAuthDatabase(object): data = {} @staticmethod def auth_token_get(context, token_hash): return FakeAuthDatabase.data.get(token_hash, None) @staticmethod def auth_token_create(context, token): fake_token = FakeToken(created_at=timeutils.utcnow(), **token) FakeAuthDatabase.data[fake_token.token_hash] = fake_token FakeAuthDatabase.data['id_%i' % fake_token.id] = fake_token return fake_token @staticmethod def auth_token_destroy(context, token_id): token = FakeAuthDatabase.data.get('id_%i' % token_id) if token and token.token_hash in FakeAuthDatabase.data: del FakeAuthDatabase.data[token.token_hash] del FakeAuthDatabase.data['id_%i' % token_id] def create_info_cache(nw_cache): if nw_cache is None: pub0 = ('192.168.1.100',) pub1 = ('2001:db8:0:1::1',) def _ip(ip): return {'address': ip, 'type': 'fixed'} nw_cache = [ {'address': 'aa:aa:aa:aa:aa:aa', 'id': 1, 'network': {'bridge': 'br0', 'id': 1, 'label': 'test1', 'subnets': [{'cidr': '192.168.1.0/24', 'ips': [_ip(ip) for ip in pub0]}, {'cidr': 'b33f::/64', 'ips': [_ip(ip) for ip in pub1]}]}}] if not isinstance(nw_cache, str): nw_cache = jsonutils.dumps(nw_cache) return { "info_cache": { "network_info": nw_cache, "deleted": False, "created_at": None, "deleted_at": None, "updated_at": None, } } def get_fake_uuid(token=0): if token not in FAKE_UUIDS: FAKE_UUIDS[token] = uuidutils.generate_uuid() return FAKE_UUIDS[token] def fake_instance_get(**kwargs): def _return_server(context, uuid, columns_to_join=None, use_slave=False): if 'project_id' not in kwargs: kwargs['project_id'] = 'fake' return stub_instance(1, **kwargs) return _return_server def fake_compute_get(**kwargs): def _return_server_obj(context, *a, **kw): return stub_instance_obj(context, **kwargs) return _return_server_obj def fake_actions_to_locked_server(self, context, instance, *args, **kwargs): raise exc.InstanceIsLocked(instance_uuid=instance['uuid']) def fake_instance_get_all_by_filters(num_servers=5, **kwargs): def _return_servers(context, *args, **kwargs): servers_list = [] marker = None limit = None found_marker = False if "marker" in kwargs: marker = kwargs["marker"] if "limit" in kwargs: limit = kwargs["limit"] if 'columns_to_join' in kwargs: kwargs.pop('columns_to_join') if 'use_slave' in kwargs: kwargs.pop('use_slave') if 'sort_keys' in kwargs: kwargs.pop('sort_keys') if 'sort_dirs' in kwargs: kwargs.pop('sort_dirs') if 'cell_mappings' in kwargs: kwargs.pop('cell_mappings') for i in range(num_servers): uuid = get_fake_uuid(i) server = stub_instance(id=i + 1, uuid=uuid, **kwargs) servers_list.append(server) if marker is not None and uuid == marker: found_marker = True servers_list = [] if marker is not None and not found_marker: raise exc.MarkerNotFound(marker=marker) if limit is not None: servers_list = servers_list[:limit] return servers_list return _return_servers def fake_compute_get_all(num_servers=5, **kwargs): def _return_servers_objs(context, search_opts=None, limit=None, marker=None, expected_attrs=None, sort_keys=None, sort_dirs=None, cell_down_support=False, all_tenants=False): db_insts = fake_instance_get_all_by_filters()(None, limit=limit, marker=marker) expected = ['metadata', 'system_metadata', 'flavor', 'info_cache', 'security_groups'] return base.obj_make_list(context, objects.InstanceList(), objects.Instance, db_insts, expected_attrs=expected) return _return_servers_objs def stub_instance(id=1, user_id=None, project_id=None, host=None, node=None, vm_state=None, task_state=None, reservation_id="", uuid=FAKE_UUID, image_ref=FAKE_UUID, flavor_id="1", name=None, key_name='', access_ipv4=None, access_ipv6=None, progress=0, auto_disk_config=False, display_name=None, display_description=None, include_fake_metadata=True, config_drive=None, power_state=None, nw_cache=None, metadata=None, security_groups=None, root_device_name=None, limit=None, marker=None, launched_at=timeutils.utcnow(), terminated_at=timeutils.utcnow(), availability_zone='', locked_by=None, cleaned=False, memory_mb=0, vcpus=0, root_gb=0, ephemeral_gb=0, flavor=None, launch_index=0, kernel_id="", ramdisk_id="", user_data=None, system_metadata=None, services=None, trusted_certs=None, hidden=False): if user_id is None: user_id = 'fake_user' if project_id is None: project_id = 'fake_project' if metadata: metadata = [{'key': k, 'value': v} for k, v in metadata.items()] elif include_fake_metadata: metadata = [models.InstanceMetadata(key='seq', value=str(id))] else: metadata = [] sys_meta = flavors.save_flavor_info( {}, flavors.get_flavor_by_flavor_id(int(flavor_id))) sys_meta.update(system_metadata or {}) if host is not None: host = str(host) if key_name: key_data = 'FAKE' else: key_data = '' if security_groups is None: security_groups = [{"id": 1, "name": "test", "description": "Foo:", "project_id": "project", "user_id": "user", "created_at": None, "updated_at": None, "deleted_at": None, "deleted": False}] # ReservationID isn't sent back, hack it in there. server_name = name or "server%s" % id if reservation_id != "": server_name = "reservation_%s" % (reservation_id, ) info_cache = create_info_cache(nw_cache) if flavor is None: flavor = objects.Flavor.get_by_name( context.get_admin_context(), 'm1.small') flavorinfo = jsonutils.dumps({ 'cur': flavor.obj_to_primitive(), 'old': None, 'new': None, }) instance = { "id": int(id), "created_at": datetime.datetime(2010, 10, 10, 12, 0, 0), "updated_at": datetime.datetime(2010, 11, 11, 11, 0, 0), "deleted_at": datetime.datetime(2010, 12, 12, 10, 0, 0), "deleted": None, "user_id": user_id, "project_id": project_id, "image_ref": image_ref, "kernel_id": kernel_id, "ramdisk_id": ramdisk_id, "hostname": display_name or server_name, "launch_index": launch_index, "key_name": key_name, "key_data": key_data, "power_state": power_state, "vm_state": vm_state or vm_states.ACTIVE, "task_state": task_state, "services": services, "memory_mb": memory_mb, "vcpus": vcpus, "root_gb": root_gb, "ephemeral_gb": ephemeral_gb, "ephemeral_key_uuid": None, "host": host, "node": node, "instance_type_id": flavor.id, "user_data": user_data, "reservation_id": reservation_id, "launched_at": launched_at, "terminated_at": terminated_at, "availability_zone": availability_zone, "display_name": display_name or server_name, "display_description": display_description, "launched_on": "", "locked": locked_by is not None, "locked_by": locked_by, "os_type": "", "architecture": "", "vm_mode": "", "uuid": uuid, "root_device_name": root_device_name, "default_ephemeral_device": "", "default_swap_device": "", "config_drive": config_drive, "access_ip_v4": access_ipv4, "access_ip_v6": access_ipv6, "auto_disk_config": auto_disk_config, "progress": progress, "shutdown_terminate": True, "disable_terminate": False, "cell_name": "", "metadata": metadata, "system_metadata": utils.dict_to_metadata(sys_meta), "security_groups": security_groups, "cleaned": cleaned, "pci_devices": [], "extra": {"numa_topology": None, "pci_requests": None, "flavor": flavorinfo, "trusted_certs": trusted_certs, }, "tags": [], "hidden": hidden, "name": "instance-%s" % id, } instance.update(info_cache) instance['info_cache']['instance_uuid'] = instance['uuid'] return instance def stub_instance_obj(ctxt, *args, **kwargs): db_inst = stub_instance(*args, **kwargs) expected = ['metadata', 'system_metadata', 'flavor', 'info_cache', 'security_groups', 'tags'] inst = objects.Instance._from_db_object(ctxt, objects.Instance(), db_inst, expected_attrs=expected) inst.fault = None if db_inst["services"] is not None: # This ensures services there if one wanted so inst.services = db_inst["services"] return inst def stub_volume(id, **kwargs): volume = { 'id': id, 'user_id': 'fakeuser', 'project_id': 'fakeproject', 'host': 'fakehost', 'size': 1, 'availability_zone': 'fakeaz', 'status': 'fakestatus', 'attach_status': 'attached', 'name': 'vol name', 'display_name': 'displayname', 'display_description': 'displaydesc', 'created_at': datetime.datetime(1999, 1, 1, 1, 1, 1), 'snapshot_id': None, 'volume_type_id': 'fakevoltype', 'volume_metadata': [], 'volume_type': {'name': 'vol_type_name'}, 'multiattach': False, 'attachments': {'fakeuuid': {'mountpoint': '/'}, 'fakeuuid2': {'mountpoint': '/dev/sdb'} } } volume.update(kwargs) return volume def stub_volume_create(self, context, size, name, description, snapshot, **param): vol = stub_volume('1') vol['size'] = size vol['display_name'] = name vol['display_description'] = description try: vol['snapshot_id'] = snapshot['id'] except (KeyError, TypeError): vol['snapshot_id'] = None vol['availability_zone'] = param.get('availability_zone', 'fakeaz') return vol def stub_volume_update(self, context, *args, **param): pass def stub_volume_get(self, context, volume_id): return stub_volume(volume_id) def stub_volume_get_all(context, search_opts=None): return [stub_volume(100, project_id='fake'), stub_volume(101, project_id='superfake'), stub_volume(102, project_id='superduperfake')] def stub_volume_check_attach(self, context, *args, **param): pass def stub_snapshot(id, **kwargs): snapshot = { 'id': id, 'volume_id': 12, 'status': 'available', 'volume_size': 100, 'created_at': timeutils.utcnow(), 'display_name': 'Default name', 'display_description': 'Default description', 'project_id': 'fake' } snapshot.update(kwargs) return snapshot def stub_snapshot_create(self, context, volume_id, name, description): return stub_snapshot(100, volume_id=volume_id, display_name=name, display_description=description) def stub_compute_volume_snapshot_create(self, context, volume_id, create_info): return {'snapshot': {'id': "421752a6-acf6-4b2d-bc7a-119f9148cd8c", 'volumeId': volume_id}} def stub_snapshot_delete(self, context, snapshot_id): if snapshot_id == '-1': raise exc.SnapshotNotFound(snapshot_id=snapshot_id) def stub_compute_volume_snapshot_delete(self, context, volume_id, snapshot_id, delete_info): pass def stub_snapshot_get(self, context, snapshot_id): if snapshot_id == '-1': raise exc.SnapshotNotFound(snapshot_id=snapshot_id) return stub_snapshot(snapshot_id) def stub_snapshot_get_all(self, context): return [stub_snapshot(100, project_id='fake'), stub_snapshot(101, project_id='superfake'), stub_snapshot(102, project_id='superduperfake')] def stub_bdm_get_all_by_instance_uuids(context, instance_uuids, use_slave=False): i = 1 result = [] for instance_uuid in instance_uuids: for x in range(2): # add two BDMs per instance result.append(fake_block_device.FakeDbBlockDeviceDict({ 'id': i, 'source_type': 'volume', 'destination_type': 'volume', 'volume_id': 'volume_id%d' % (i), 'instance_uuid': instance_uuid, })) i += 1 return result def fake_not_implemented(*args, **kwargs): raise NotImplementedError() FLAVORS = { '1': objects.Flavor( id=1, name='flavor 1', memory_mb=256, vcpus=1, root_gb=10, ephemeral_gb=20, flavorid='1', swap=10, rxtx_factor=1.0, vcpu_weight=None, disabled=False, is_public=True, description=None, extra_specs={"key1": "value1", "key2": "value2"} ), '2': objects.Flavor( id=2, name='flavor 2', memory_mb=512, vcpus=1, root_gb=20, ephemeral_gb=10, flavorid='2', swap=5, rxtx_factor=None, vcpu_weight=None, disabled=True, is_public=True, description='flavor 2 description', extra_specs={} ), } def stub_out_flavor_get_by_flavor_id(test): @staticmethod def fake_get_by_flavor_id(context, flavor_id, read_deleted=None): return FLAVORS[flavor_id] test.stub_out('nova.objects.Flavor.get_by_flavor_id', fake_get_by_flavor_id) def stub_out_flavor_get_all(test): @staticmethod def fake_get_all(context, inactive=False, filters=None, sort_key='flavorid', sort_dir='asc', limit=None, marker=None): if marker in ['99999']: raise exc.MarkerNotFound(marker) def reject_min(db_attr, filter_attr): return (filter_attr in filters and getattr(flavor, db_attr) < int(filters[filter_attr])) filters = filters or {} res = [] for flavor in FLAVORS.values(): if reject_min('memory_mb', 'min_memory_mb'): continue elif reject_min('root_gb', 'min_root_gb'): continue res.append(flavor) res = sorted(res, key=lambda item: getattr(item, sort_key)) output = [] marker_found = True if marker is None else False for flavor in res: if not marker_found and marker == flavor.flavorid: marker_found = True elif marker_found: if limit is None or len(output) < int(limit): output.append(flavor) return objects.FlavorList(objects=output) test.stub_out('nova.objects.FlavorList.get_all', fake_get_all)