# Copyright 2010 United States Government as represented by the # Administrator of the National Aeronautics and Space Administration. # 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 ddt import mock from oslo_db.sqlalchemy import enginefacade from nova.compute import api as compute import nova.conf from nova import context from nova.db.sqlalchemy import models as sqa_models from nova import exception from nova import objects from nova import quota from nova import test from nova.tests import fixtures as nova_fixtures CONF = nova.conf.CONF def _get_fake_get_usages(updates=None): # These values are not realistic (they should all be 0) and are # only for testing that countable usages get included in the # results. usages = {'key_pairs': {'in_use': 2}, 'server_group_members': {'in_use': 3}, 'instances': {'in_use': 2}, 'cores': {'in_use': 4}, 'ram': {'in_use': 10 * 1024}} if updates: usages.update(updates) def fake_get_usages(*a, **k): return usages return fake_get_usages class QuotaIntegrationTestCase(test.TestCase): REQUIRES_LOCKING = True def setUp(self): super(QuotaIntegrationTestCase, self).setUp() self.flags(instances=2, cores=4, group='quota') self.user_id = 'admin' self.project_id = 'admin' self.context = context.RequestContext(self.user_id, self.project_id, is_admin=True) self.flavor = objects.Flavor.get_by_name(self.context, 'm1.small') self.useFixture(nova_fixtures.GlanceFixture(self)) self.compute_api = compute.API() def fake_validate_networks(context, requested_networks, num_instances): return num_instances # we aren't testing network quota in these tests when creating a server # so just mock that out and assume network (port) quota is OK self.compute_api.network_api.validate_networks = ( mock.Mock(side_effect=fake_validate_networks)) def _create_instance(self, flavor_name='m1.large'): """Create a test instance in cell1 with an instance mapping.""" cell1 = self.cell_mappings[test.CELL1_NAME] with context.target_cell(self.context, cell1) as cctxt: inst = objects.Instance(context=cctxt) inst.image_id = 'cedef40a-ed67-4d10-800e-17455edce175' inst.reservation_id = 'r-fakeres' inst.user_id = self.user_id inst.project_id = self.project_id inst.flavor = objects.Flavor.get_by_name(cctxt, flavor_name) # This is needed for instance quota counting until we have the # ability to count allocations in placement. inst.vcpus = inst.flavor.vcpus inst.memory_mb = inst.flavor.memory_mb inst.create() # Create the related instance mapping which will be used in # _instances_cores_ram_count(). inst_map = objects.InstanceMapping( self.context, instance_uuid=inst.uuid, project_id=inst.project_id, cell_mapping=cell1) inst_map.create() return inst def test_too_many_instances(self): for i in range(CONF.quota.instances): self._create_instance() image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175' try: self.compute_api.create( self.context, min_count=1, max_count=1, flavor=self.flavor, image_href=image_uuid) except exception.QuotaError as e: expected_kwargs = {'code': 413, 'req': '1, 1', 'used': '8, 2', 'allowed': '4, 2', 'overs': 'cores, instances'} self.assertEqual(expected_kwargs, e.kwargs) else: self.fail('Expected QuotaError exception') def test_too_many_cores(self): self._create_instance() image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175' try: self.compute_api.create( self.context, min_count=1, max_count=1, flavor=self.flavor, image_href=image_uuid) except exception.QuotaError as e: expected_kwargs = {'code': 413, 'req': '1', 'used': '4', 'allowed': '4', 'overs': 'cores'} self.assertEqual(expected_kwargs, e.kwargs) else: self.fail('Expected QuotaError exception') def test_many_cores_with_unlimited_quota(self): # Setting cores quota to unlimited: self.flags(cores=-1, group='quota') # Default is 20 cores, so create 3x m1.xlarge with # 8 cores each. for i in range(3): self._create_instance(flavor_name='m1.xlarge') def test_too_many_metadata_items(self): metadata = {} for i in range(CONF.quota.metadata_items + 1): metadata['key%s' % i] = 'value%s' % i image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175' self.assertRaises( exception.QuotaError, self.compute_api.create, self.context, min_count=1, max_count=1, flavor=self.flavor, image_href=image_uuid, metadata=metadata) def _create_with_injected_files(self, files): api = self.compute_api image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175' api.create( self.context, min_count=1, max_count=1, flavor=self.flavor, image_href=image_uuid, injected_files=files) def test_no_injected_files(self): api = self.compute_api image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175' api.create(self.context, flavor=self.flavor, image_href=image_uuid) def test_max_injected_files(self): files = [] for i in range(CONF.quota.injected_files): files.append(('/my/path%d' % i, 'config = test\n')) self._create_with_injected_files(files) # no QuotaError def test_too_many_injected_files(self): files = [] for i in range(CONF.quota.injected_files + 1): files.append(('/my/path%d' % i, 'my\ncontent%d\n' % i)) self.assertRaises(exception.QuotaError, self._create_with_injected_files, files) def test_max_injected_file_content_bytes(self): max = CONF.quota.injected_file_content_bytes content = ''.join(['a' for i in range(max)]) files = [('/test/path', content)] self._create_with_injected_files(files) # no QuotaError def test_too_many_injected_file_content_bytes(self): max = CONF.quota.injected_file_content_bytes content = ''.join(['a' for i in range(max + 1)]) files = [('/test/path', content)] self.assertRaises(exception.QuotaError, self._create_with_injected_files, files) def test_max_injected_file_path_bytes(self): max = CONF.quota.injected_file_path_length path = ''.join(['a' for i in range(max)]) files = [(path, 'config = quotatest')] self._create_with_injected_files(files) # no QuotaError def test_too_many_injected_file_path_bytes(self): max = CONF.quota.injected_file_path_length path = ''.join(['a' for i in range(max + 1)]) files = [(path, 'config = quotatest')] self.assertRaises(exception.QuotaError, self._create_with_injected_files, files) @enginefacade.transaction_context_provider class FakeContext(context.RequestContext): def __init__(self, project_id, quota_class): super(FakeContext, self).__init__(project_id=project_id, quota_class=quota_class) self.is_admin = False self.user_id = 'fake_user' self.project_id = project_id self.quota_class = quota_class self.read_deleted = 'no' def elevated(self): elevated = self.__class__(self.project_id, self.quota_class) elevated.is_admin = True return elevated class FakeDriver(object): def __init__(self, by_project=None, by_user=None, by_class=None, reservations=None): self.called = [] self.by_project = by_project or {} self.by_user = by_user or {} self.by_class = by_class or {} self.reservations = reservations or [] def get_defaults(self, context, resources): self.called.append(('get_defaults', context, resources)) return resources def get_class_quotas(self, context, resources, quota_class): self.called.append(('get_class_quotas', context, resources, quota_class)) return resources def get_user_quotas(self, context, resources, project_id, user_id, quota_class=None, usages=True): self.called.append(('get_user_quotas', context, resources, project_id, user_id, quota_class, usages)) return resources def get_project_quotas(self, context, resources, project_id, quota_class=None, usages=True, remains=False): self.called.append(('get_project_quotas', context, resources, project_id, quota_class, usages, remains)) return resources def limit_check(self, context, resources, values, project_id=None, user_id=None): self.called.append(('limit_check', context, resources, values, project_id, user_id)) def limit_check_project_and_user(self, context, resources, project_values=None, user_values=None, project_id=None, user_id=None): self.called.append(('limit_check_project_and_user', context, resources, project_values, user_values, project_id, user_id)) class BaseResourceTestCase(test.TestCase): def test_no_flag(self): resource = quota.BaseResource('test_resource') self.assertEqual(resource.name, 'test_resource') self.assertIsNone(resource.flag) self.assertEqual(resource.default, -1) def test_with_flag(self): # We know this flag exists, so use it... self.flags(instances=10, group='quota') resource = quota.BaseResource('test_resource', 'instances') self.assertEqual(resource.name, 'test_resource') self.assertEqual(resource.flag, 'instances') self.assertEqual(resource.default, 10) def test_with_flag_no_quota(self): self.flags(instances=-1, group='quota') resource = quota.BaseResource('test_resource', 'instances') self.assertEqual(resource.name, 'test_resource') self.assertEqual(resource.flag, 'instances') self.assertEqual(resource.default, -1) def test_valid_method_call_check_invalid_input(self): resources = {'dummy': 1} self.assertRaises(exception.InvalidQuotaMethodUsage, quota._valid_method_call_check_resources, resources, 'limit', quota.QUOTAS._resources) def test_valid_method_call_check_invalid_method(self): resources = {'key_pairs': 1} self.assertRaises(exception.InvalidQuotaMethodUsage, quota._valid_method_call_check_resources, resources, 'dummy', quota.QUOTAS._resources) def test_valid_method_call_check_multiple(self): resources = {'key_pairs': 1, 'dummy': 2} self.assertRaises(exception.InvalidQuotaMethodUsage, quota._valid_method_call_check_resources, resources, 'check', quota.QUOTAS._resources) resources = {'key_pairs': 1, 'instances': 2, 'dummy': 3} self.assertRaises(exception.InvalidQuotaMethodUsage, quota._valid_method_call_check_resources, resources, 'check', quota.QUOTAS._resources) def test_valid_method_call_check_wrong_method(self): resources = {'key_pairs': 1} engine_resources = {'key_pairs': quota.CountableResource('key_pairs', None, 'key_pairs')} self.assertRaises(exception.InvalidQuotaMethodUsage, quota._valid_method_call_check_resources, resources, 'bogus', engine_resources) class QuotaEngineTestCase(test.TestCase): def test_init(self): quota_obj = quota.QuotaEngine() self.assertIsInstance(quota_obj._driver, quota.DbQuotaDriver) def test_init_override_obj(self): quota_obj = quota.QuotaEngine(quota_driver=FakeDriver) self.assertEqual(quota_obj._driver, FakeDriver) def _get_quota_engine(self, driver, resources=None): resources = resources or [ quota.AbsoluteResource('test_resource4'), quota.AbsoluteResource('test_resource3'), quota.AbsoluteResource('test_resource2'), quota.AbsoluteResource('test_resource1'), ] return quota.QuotaEngine(quota_driver=driver, resources=resources) def test_get_defaults(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) result = quota_obj.get_defaults(context) self.assertEqual(driver.called, [ ('get_defaults', context, quota_obj._resources), ]) self.assertEqual(result, quota_obj._resources) def test_get_class_quotas(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) result1 = quota_obj.get_class_quotas(context, 'test_class') self.assertEqual(driver.called, [ ('get_class_quotas', context, quota_obj._resources, 'test_class'), ]) self.assertEqual(result1, quota_obj._resources) def test_get_user_quotas(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) result1 = quota_obj.get_user_quotas(context, 'test_project', 'fake_user') result2 = quota_obj.get_user_quotas(context, 'test_project', 'fake_user', quota_class='test_class', usages=False) self.assertEqual(driver.called, [ ('get_user_quotas', context, quota_obj._resources, 'test_project', 'fake_user', None, True), ('get_user_quotas', context, quota_obj._resources, 'test_project', 'fake_user', 'test_class', False), ]) self.assertEqual(result1, quota_obj._resources) self.assertEqual(result2, quota_obj._resources) def test_get_project_quotas(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) result1 = quota_obj.get_project_quotas(context, 'test_project') result2 = quota_obj.get_project_quotas(context, 'test_project', quota_class='test_class', usages=False) self.assertEqual(driver.called, [ ('get_project_quotas', context, quota_obj._resources, 'test_project', None, True, False), ('get_project_quotas', context, quota_obj._resources, 'test_project', 'test_class', False, False), ]) self.assertEqual(result1, quota_obj._resources) self.assertEqual(result2, quota_obj._resources) def test_count_as_dict_no_resource(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) self.assertRaises(exception.QuotaResourceUnknown, quota_obj.count_as_dict, context, 'test_resource5', True, foo='bar') def test_count_as_dict_wrong_resource(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) self.assertRaises(exception.QuotaResourceUnknown, quota_obj.count_as_dict, context, 'test_resource1', True, foo='bar') def test_count_as_dict(self): def fake_count_as_dict(context, *args, **kwargs): self.assertEqual(args, (True,)) self.assertEqual(kwargs, dict(foo='bar')) return {'project': {'test_resource5': 5}} context = FakeContext(None, None) driver = FakeDriver() resources = [ quota.CountableResource('test_resource5', fake_count_as_dict), ] quota_obj = self._get_quota_engine(driver, resources) result = quota_obj.count_as_dict(context, 'test_resource5', True, foo='bar') self.assertEqual({'project': {'test_resource5': 5}}, result) def test_limit_check(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) quota_obj.limit_check(context, test_resource1=4, test_resource2=3, test_resource3=2, test_resource4=1) self.assertEqual(driver.called, [ ('limit_check', context, quota_obj._resources, dict( test_resource1=4, test_resource2=3, test_resource3=2, test_resource4=1, ), None, None), ]) def test_limit_check_project_and_user(self): context = FakeContext(None, None) driver = FakeDriver() quota_obj = self._get_quota_engine(driver) project_values = dict(test_resource1=4, test_resource2=3) user_values = dict(test_resource3=2, test_resource4=1) quota_obj.limit_check_project_and_user(context, project_values=project_values, user_values=user_values) self.assertEqual([('limit_check_project_and_user', context, quota_obj._resources, dict(test_resource1=4, test_resource2=3), dict(test_resource3=2, test_resource4=1), None, None)], driver.called) def test_resources(self): quota_obj = self._get_quota_engine(None) self.assertEqual(quota_obj.resources, ['test_resource1', 'test_resource2', 'test_resource3', 'test_resource4']) class DbQuotaDriverTestCase(test.TestCase): def setUp(self): super(DbQuotaDriverTestCase, self).setUp() self.flags(instances=10, cores=20, ram=50 * 1024, metadata_items=128, injected_files=5, injected_file_content_bytes=10 * 1024, injected_file_path_length=255, server_groups=10, server_group_members=10, group='quota' ) self.driver = quota.DbQuotaDriver() self.calls = [] self.useFixture(test.TimeOverride()) def test_get_defaults(self): # Use our pre-defined resources self._stub_quota_class_get_default() result = self.driver.get_defaults(None, quota.QUOTAS._resources) self.assertEqual(result, dict( instances=5, cores=20, ram=25 * 1024, metadata_items=64, injected_files=5, injected_file_content_bytes=5 * 1024, injected_file_path_bytes=255, key_pairs=100, server_groups=10, server_group_members=10, security_groups=-1, security_group_rules=-1, fixed_ips=-1, floating_ips=-1, )) def _stub_quota_class_get_default(self): # Stub out quota_class_get_default def fake_qcgd(cls, context): self.calls.append('quota_class_get_default') return dict( instances=5, ram=25 * 1024, metadata_items=64, injected_file_content_bytes=5 * 1024, ) self.stub_out('nova.objects.Quotas.get_default_class', fake_qcgd) def _stub_quota_class_get_all_by_name(self): # Stub out quota_class_get_all_by_name def fake_qcgabn(cls, context, quota_class): self.calls.append('quota_class_get_all_by_name') self.assertEqual(quota_class, 'test_class') return dict( instances=5, ram=25 * 1024, metadata_items=64, injected_file_content_bytes=5 * 1024, ) self.stub_out('nova.objects.Quotas.get_all_class_by_name', fake_qcgabn) def test_get_class_quotas(self): self._stub_quota_class_get_all_by_name() result = self.driver.get_class_quotas(None, quota.QUOTAS._resources, 'test_class') self.assertEqual(self.calls, ['quota_class_get_all_by_name']) self.assertEqual(result, dict( instances=5, cores=20, ram=25 * 1024, metadata_items=64, injected_files=5, injected_file_content_bytes=5 * 1024, injected_file_path_bytes=255, key_pairs=100, server_groups=10, server_group_members=10, floating_ips=-1, fixed_ips=-1, security_groups=-1, security_group_rules=-1, )) def _stub_get_by_project_and_user(self): def fake_qgabpau(context, project_id, user_id): self.calls.append('quota_get_all_by_project_and_user') self.assertEqual(project_id, 'test_project') self.assertEqual(user_id, 'fake_user') return dict( cores=10, injected_files=2, injected_file_path_bytes=127, ) def fake_qgabp(context, project_id): self.calls.append('quota_get_all_by_project') self.assertEqual(project_id, 'test_project') return { 'cores': 10, 'injected_files': 2, 'injected_file_path_bytes': 127, } self.stub_out('nova.db.api.quota_get_all_by_project_and_user', fake_qgabpau) self.stub_out('nova.db.api.quota_get_all_by_project', fake_qgabp) self._stub_quota_class_get_all_by_name() def _get_fake_countable_resources(self): # Create several countable resources with fake count functions def fake_instances_cores_ram_count(*a, **k): return {'project': {'instances': 2, 'cores': 4, 'ram': 1024}, 'user': {'instances': 1, 'cores': 2, 'ram': 512}} def fake_server_group_count(*a, **k): return {'project': {'server_groups': 5}, 'user': {'server_groups': 3}} resources = {} resources['key_pairs'] = quota.CountableResource( 'key_pairs', lambda *a, **k: {'user': {'key_pairs': 1}}, 'key_pairs') resources['instances'] = quota.CountableResource( 'instances', fake_instances_cores_ram_count, 'instances') resources['cores'] = quota.CountableResource( 'cores', fake_instances_cores_ram_count, 'cores') resources['ram'] = quota.CountableResource( 'ram', fake_instances_cores_ram_count, 'ram') resources['server_groups'] = quota.CountableResource( 'server_groups', fake_server_group_count, 'server_groups') resources['server_group_members'] = quota.CountableResource( 'server_group_members', lambda *a, **k: {'user': {'server_group_members': 7}}, 'server_group_members') resources['floating_ips'] = quota.AbsoluteResource('floating_ips') resources['fixed_ips'] = quota.AbsoluteResource('fixed_ips') resources['security_groups'] = quota.AbsoluteResource( 'security_groups') resources['security_group_rules'] = quota.AbsoluteResource( 'security_group_rules') return resources def test_get_usages_for_project(self): resources = self._get_fake_countable_resources() actual = self.driver._get_usages( FakeContext('test_project', 'test_class'), resources, 'test_project') # key_pairs, server_group_members, and security_group_rules are never # counted as a usage. Their counts are only for quota limit checking. expected = {'key_pairs': {'in_use': 0}, 'instances': {'in_use': 2}, 'cores': {'in_use': 4}, 'ram': {'in_use': 1024}, 'server_groups': {'in_use': 5}, 'server_group_members': {'in_use': 0}} self.assertEqual(expected, actual) def test_get_usages_for_user(self): resources = self._get_fake_countable_resources() actual = self.driver._get_usages( FakeContext('test_project', 'test_class'), resources, 'test_project', user_id='fake_user') # key_pairs, server_group_members, and security_group_rules are never # counted as a usage. Their counts are only for quota limit checking. expected = {'key_pairs': {'in_use': 0}, 'instances': {'in_use': 1}, 'cores': {'in_use': 2}, 'ram': {'in_use': 512}, 'server_groups': {'in_use': 3}, 'server_group_members': {'in_use': 0}} self.assertEqual(expected, actual) @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_user_quotas(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project_and_user() ctxt = FakeContext('test_project', 'test_class') result = self.driver.get_user_quotas( ctxt, quota.QUOTAS._resources, 'test_project', 'fake_user') self.assertEqual(self.calls, [ 'quota_get_all_by_project_and_user', 'quota_get_all_by_project', 'quota_class_get_all_by_name', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project', user_id='fake_user') self.assertEqual(result, dict( instances=dict( limit=5, in_use=2, ), cores=dict( limit=10, in_use=4, ), ram=dict( limit=25 * 1024, in_use=10 * 1024, ), floating_ips=dict( limit=-1, in_use=0, ), fixed_ips=dict( limit=-1, in_use=0, ), metadata_items=dict( limit=64, in_use=0, ), injected_files=dict( limit=2, in_use=0, ), injected_file_content_bytes=dict( limit=5 * 1024, in_use=0, ), injected_file_path_bytes=dict( limit=127, in_use=0, ), security_groups=dict( limit=-1, in_use=0, ), security_group_rules=dict( limit=-1, in_use=0, ), key_pairs=dict( limit=100, in_use=2, ), server_groups=dict( limit=10, in_use=0, ), server_group_members=dict( limit=10, in_use=3, ), )) def _stub_get_by_project_and_user_specific(self): def fake_quota_get(context, project_id, resource, user_id=None): self.calls.append('quota_get') self.assertEqual(project_id, 'test_project') self.assertEqual(user_id, 'fake_user') self.assertEqual(resource, 'test_resource') return dict( test_resource=dict(in_use=20), ) self.stub_out('nova.db.api.quota_get', fake_quota_get) def _stub_get_by_project(self): def fake_qgabp(context, project_id): self.calls.append('quota_get_all_by_project') self.assertEqual(project_id, 'test_project') return dict( cores=10, injected_files=2, injected_file_path_bytes=127, ) def fake_quota_get_all(context, project_id): self.calls.append('quota_get_all') self.assertEqual(project_id, 'test_project') return [sqa_models.ProjectUserQuota(resource='instances', hard_limit=5), sqa_models.ProjectUserQuota(resource='cores', hard_limit=2)] self.stub_out('nova.db.api.quota_get_all_by_project', fake_qgabp) self.stub_out('nova.db.api.quota_get_all', fake_quota_get_all) self._stub_quota_class_get_all_by_name() self._stub_quota_class_get_default() @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_project_quotas(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project() ctxt = FakeContext('test_project', 'test_class') result = self.driver.get_project_quotas( ctxt, quota.QUOTAS._resources, 'test_project') self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'quota_class_get_all_by_name', 'quota_class_get_default', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project') self.assertEqual(result, dict( instances=dict( limit=5, in_use=2, ), cores=dict( limit=10, in_use=4, ), ram=dict( limit=25 * 1024, in_use=10 * 1024, ), floating_ips=dict( limit=-1, in_use=0, ), fixed_ips=dict( limit=-1, in_use=0, ), metadata_items=dict( limit=64, in_use=0, ), injected_files=dict( limit=2, in_use=0, ), injected_file_content_bytes=dict( limit=5 * 1024, in_use=0, ), injected_file_path_bytes=dict( limit=127, in_use=0, ), security_groups=dict( limit=-1, in_use=0, ), security_group_rules=dict( limit=-1, in_use=0, ), key_pairs=dict( limit=100, in_use=2, ), server_groups=dict( limit=10, in_use=0, ), server_group_members=dict( limit=10, in_use=3, ), )) @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_project_quotas_with_remains(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project() ctxt = FakeContext('test_project', 'test_class') result = self.driver.get_project_quotas( ctxt, quota.QUOTAS._resources, 'test_project', remains=True) self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'quota_class_get_all_by_name', 'quota_class_get_default', 'quota_get_all', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project') self.assertEqual(result, dict( instances=dict( limit=5, in_use=2, remains=0, ), cores=dict( limit=10, in_use=4, remains=8, ), ram=dict( limit=25 * 1024, in_use=10 * 1024, remains=25 * 1024, ), floating_ips=dict( limit=-1, in_use=0, remains=-1, ), fixed_ips=dict( limit=-1, in_use=0, remains=-1, ), metadata_items=dict( limit=64, in_use=0, remains=64, ), injected_files=dict( limit=2, in_use=0, remains=2, ), injected_file_content_bytes=dict( limit=5 * 1024, in_use=0, remains=5 * 1024, ), injected_file_path_bytes=dict( limit=127, in_use=0, remains=127, ), security_groups=dict( limit=-1, in_use=0, remains=-1, ), security_group_rules=dict( limit=-1, in_use=0, remains=-1, ), key_pairs=dict( limit=100, in_use=2, remains=100, ), server_groups=dict( limit=10, in_use=0, remains=10, ), server_group_members=dict( limit=10, in_use=3, remains=10, ), )) @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_user_quotas_alt_context_no_class(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project_and_user() ctxt = FakeContext('other_project', None) result = self.driver.get_user_quotas( ctxt, quota.QUOTAS._resources, 'test_project', 'fake_user') self.assertEqual(self.calls, [ 'quota_get_all_by_project_and_user', 'quota_get_all_by_project', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project', user_id='fake_user') self.assertEqual(result, dict( instances=dict( limit=10, in_use=2, ), cores=dict( limit=10, in_use=4, ), ram=dict( limit=50 * 1024, in_use=10 * 1024, ), floating_ips=dict( limit=-1, in_use=0, ), fixed_ips=dict( limit=-1, in_use=0, ), metadata_items=dict( limit=128, in_use=0, ), injected_files=dict( limit=2, in_use=0, ), injected_file_content_bytes=dict( limit=10 * 1024, in_use=0, ), injected_file_path_bytes=dict( limit=127, in_use=0, ), security_groups=dict( limit=-1, in_use=0, ), security_group_rules=dict( limit=-1, in_use=0, ), key_pairs=dict( limit=100, in_use=2, ), server_groups=dict( limit=10, in_use=0, ), server_group_members=dict( limit=10, in_use=3, ), )) @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_project_quotas_alt_context_no_class(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project() ctxt = FakeContext('other_project', None) result = self.driver.get_project_quotas( ctxt, quota.QUOTAS._resources, 'test_project') self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'quota_class_get_default', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project') self.assertEqual(result, dict( instances=dict( limit=5, in_use=2, ), cores=dict( limit=10, in_use=4, ), ram=dict( limit=25 * 1024, in_use=10 * 1024, ), floating_ips=dict( limit=-1, in_use=0, ), fixed_ips=dict( limit=-1, in_use=0, ), metadata_items=dict( limit=64, in_use=0, ), injected_files=dict( limit=2, in_use=0, ), injected_file_content_bytes=dict( limit=5 * 1024, in_use=0, ), injected_file_path_bytes=dict( limit=127, in_use=0, ), security_groups=dict( limit=-1, in_use=0, ), security_group_rules=dict( limit=-1, in_use=0, ), key_pairs=dict( limit=100, in_use=2, ), server_groups=dict( limit=10, in_use=0, ), server_group_members=dict( limit=10, in_use=3, ), )) @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_user_quotas_alt_context_with_class(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project_and_user() ctxt = FakeContext('other_project', 'other_class') result = self.driver.get_user_quotas( ctxt, quota.QUOTAS._resources, 'test_project', 'fake_user', quota_class='test_class') self.assertEqual(self.calls, [ 'quota_get_all_by_project_and_user', 'quota_get_all_by_project', 'quota_class_get_all_by_name', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project', user_id='fake_user') self.assertEqual(result, dict( instances=dict( limit=5, in_use=2, ), cores=dict( limit=10, in_use=4, ), ram=dict( limit=25 * 1024, in_use=10 * 1024, ), floating_ips=dict( limit=-1, in_use=0, ), fixed_ips=dict( limit=-1, in_use=0, ), metadata_items=dict( limit=64, in_use=0, ), injected_files=dict( limit=2, in_use=0, ), injected_file_content_bytes=dict( limit=5 * 1024, in_use=0, ), injected_file_path_bytes=dict( limit=127, in_use=0, ), security_groups=dict( limit=-1, in_use=0, ), security_group_rules=dict( limit=-1, in_use=0, ), key_pairs=dict( limit=100, in_use=2, ), server_groups=dict( limit=10, in_use=0, ), server_group_members=dict( limit=10, in_use=3, ), )) @mock.patch('nova.quota.DbQuotaDriver._get_usages', side_effect=_get_fake_get_usages()) def test_get_project_quotas_alt_context_with_class(self, mock_get_usages): self.maxDiff = None self._stub_get_by_project() ctxt = FakeContext('other_project', 'other_class') result = self.driver.get_project_quotas( ctxt, quota.QUOTAS._resources, 'test_project', quota_class='test_class') self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'quota_class_get_all_by_name', 'quota_class_get_default', ]) mock_get_usages.assert_called_once_with(ctxt, quota.QUOTAS._resources, 'test_project') self.assertEqual(result, dict( instances=dict( limit=5, in_use=2, ), cores=dict( limit=10, in_use=4, ), ram=dict( limit=25 * 1024, in_use=10 * 1024, ), floating_ips=dict( limit=-1, in_use=0, ), fixed_ips=dict( limit=-1, in_use=0, ), metadata_items=dict( limit=64, in_use=0, ), injected_files=dict( limit=2, in_use=0, ), injected_file_content_bytes=dict( limit=5 * 1024, in_use=0, ), injected_file_path_bytes=dict( limit=127, in_use=0, ), security_groups=dict( limit=-1, in_use=0, ), security_group_rules=dict( limit=-1, in_use=0, ), key_pairs=dict( limit=100, in_use=2, ), server_groups=dict( limit=10, in_use=0, ), server_group_members=dict( limit=10, in_use=3, ), )) def test_get_user_quotas_no_usages(self): self._stub_get_by_project_and_user() result = self.driver.get_user_quotas( FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, 'test_project', 'fake_user', usages=False) self.assertEqual(self.calls, [ 'quota_get_all_by_project_and_user', 'quota_get_all_by_project', 'quota_class_get_all_by_name', ]) self.assertEqual(result, dict( instances=dict( limit=5, ), cores=dict( limit=10, ), ram=dict( limit=25 * 1024, ), floating_ips=dict( limit=-1, ), fixed_ips=dict( limit=-1, ), metadata_items=dict( limit=64, ), injected_files=dict( limit=2, ), injected_file_content_bytes=dict( limit=5 * 1024, ), injected_file_path_bytes=dict( limit=127, ), security_groups=dict( limit=-1, ), security_group_rules=dict( limit=-1, ), key_pairs=dict( limit=100, ), server_groups=dict( limit=10, ), server_group_members=dict( limit=10, ), )) def test_get_project_quotas_no_usages(self): self._stub_get_by_project() result = self.driver.get_project_quotas( FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, 'test_project', usages=False) self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'quota_class_get_all_by_name', 'quota_class_get_default', ]) self.assertEqual(result, dict( instances=dict( limit=5, ), cores=dict( limit=10, ), ram=dict( limit=25 * 1024, ), floating_ips=dict( limit=-1, ), fixed_ips=dict( limit=-1, ), metadata_items=dict( limit=64, ), injected_files=dict( limit=2, ), injected_file_content_bytes=dict( limit=5 * 1024, ), injected_file_path_bytes=dict( limit=127, ), security_groups=dict( limit=-1, ), security_group_rules=dict( limit=-1, ), key_pairs=dict( limit=100, ), server_groups=dict( limit=10, ), server_group_members=dict( limit=10, ), )) def _stub_get_settable_quotas(self): def fake_quota_get_all_by_project(context, project_id): self.calls.append('quota_get_all_by_project') return {'floating_ips': -1} def fake_get_project_quotas(dbdrv, context, resources, project_id, quota_class=None, usages=True, remains=False, project_quotas=None): self.calls.append('get_project_quotas') result = {} for k, v in resources.items(): limit = v.default if k == 'instances': remains = v.default - 5 in_use = 1 elif k == 'cores': remains = -1 in_use = 5 limit = -1 elif k == 'floating_ips': remains = -1 in_use = 0 limit = -1 else: remains = v.default in_use = 0 result[k] = {'limit': limit, 'in_use': in_use, 'remains': remains} return result def fake_process_quotas_in_get_user_quotas(dbdrv, context, resources, project_id, quotas, quota_class=None, usages=None, remains=False): self.calls.append('_process_quotas') result = {} for k, v in resources.items(): if k == 'instances': in_use = 1 elif k == 'cores': in_use = 15 else: in_use = 0 result[k] = {'limit': v.default, 'in_use': in_use} return result def fake_qgabpau(context, project_id, user_id): self.calls.append('quota_get_all_by_project_and_user') return {'instances': 2, 'cores': -1} self.stub_out('nova.db.api.quota_get_all_by_project', fake_quota_get_all_by_project) self.stub_out('nova.quota.DbQuotaDriver.get_project_quotas', fake_get_project_quotas) self.stub_out('nova.quota.DbQuotaDriver._process_quotas', fake_process_quotas_in_get_user_quotas) self.stub_out('nova.db.api.quota_get_all_by_project_and_user', fake_qgabpau) def test_get_settable_quotas_with_user(self): self._stub_get_settable_quotas() result = self.driver.get_settable_quotas( FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, 'test_project', user_id='test_user') self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'get_project_quotas', 'quota_get_all_by_project_and_user', '_process_quotas', ]) self.assertEqual(result, { 'instances': { 'minimum': 1, 'maximum': 7, }, 'cores': { 'minimum': 15, 'maximum': -1, }, 'ram': { 'minimum': 0, 'maximum': 50 * 1024, }, 'floating_ips': { 'minimum': 0, 'maximum': -1, }, 'fixed_ips': { 'minimum': 0, 'maximum': -1, }, 'metadata_items': { 'minimum': 0, 'maximum': 128, }, 'injected_files': { 'minimum': 0, 'maximum': 5, }, 'injected_file_content_bytes': { 'minimum': 0, 'maximum': 10 * 1024, }, 'injected_file_path_bytes': { 'minimum': 0, 'maximum': 255, }, 'security_groups': { 'minimum': 0, 'maximum': -1, }, 'security_group_rules': { 'minimum': 0, 'maximum': -1, }, 'key_pairs': { 'minimum': 0, 'maximum': 100, }, 'server_groups': { 'minimum': 0, 'maximum': 10, }, 'server_group_members': { 'minimum': 0, 'maximum': 10, }, }) def test_get_settable_quotas_without_user(self): self._stub_get_settable_quotas() result = self.driver.get_settable_quotas( FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, 'test_project') self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'get_project_quotas', ]) self.assertEqual(result, { 'instances': { 'minimum': 5, 'maximum': -1, }, 'cores': { 'minimum': 5, 'maximum': -1, }, 'ram': { 'minimum': 0, 'maximum': -1, }, 'floating_ips': { 'minimum': 0, 'maximum': -1, }, 'fixed_ips': { 'minimum': 0, 'maximum': -1, }, 'metadata_items': { 'minimum': 0, 'maximum': -1, }, 'injected_files': { 'minimum': 0, 'maximum': -1, }, 'injected_file_content_bytes': { 'minimum': 0, 'maximum': -1, }, 'injected_file_path_bytes': { 'minimum': 0, 'maximum': -1, }, 'security_groups': { 'minimum': 0, 'maximum': -1, }, 'security_group_rules': { 'minimum': 0, 'maximum': -1, }, 'key_pairs': { 'minimum': 0, 'maximum': -1, }, 'server_groups': { 'minimum': 0, 'maximum': -1, }, 'server_group_members': { 'minimum': 0, 'maximum': -1, }, }) def test_get_settable_quotas_by_user_with_unlimited_value(self): self._stub_get_settable_quotas() result = self.driver.get_settable_quotas( FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, 'test_project', user_id='test_user') self.assertEqual(self.calls, [ 'quota_get_all_by_project', 'get_project_quotas', 'quota_get_all_by_project_and_user', '_process_quotas', ]) self.assertEqual(result, { 'instances': { 'minimum': 1, 'maximum': 7, }, 'cores': { 'minimum': 15, 'maximum': -1, }, 'ram': { 'minimum': 0, 'maximum': 50 * 1024, }, 'floating_ips': { 'minimum': 0, 'maximum': -1, }, 'fixed_ips': { 'minimum': 0, 'maximum': -1, }, 'metadata_items': { 'minimum': 0, 'maximum': 128, }, 'injected_files': { 'minimum': 0, 'maximum': 5, }, 'injected_file_content_bytes': { 'minimum': 0, 'maximum': 10 * 1024, }, 'injected_file_path_bytes': { 'minimum': 0, 'maximum': 255, }, 'security_groups': { 'minimum': 0, 'maximum': -1, }, 'security_group_rules': { 'minimum': 0, 'maximum': -1, }, 'key_pairs': { 'minimum': 0, 'maximum': 100, }, 'server_groups': { 'minimum': 0, 'maximum': 10, }, 'server_group_members': { 'minimum': 0, 'maximum': 10, }, }) def _stub_get_project_quotas(self): def fake_get_project_quotas(dbdrv, context, resources, project_id, quota_class=None, usages=True, remains=False, project_quotas=None): self.calls.append('get_project_quotas') return {k: dict(limit=v.default) for k, v in resources.items()} self.stub_out('nova.quota.DbQuotaDriver.get_project_quotas', fake_get_project_quotas) def test_get_quotas_unknown(self): self._stub_get_project_quotas() self.assertRaises(exception.QuotaResourceUnknown, self.driver._get_quotas, None, quota.QUOTAS._resources, ['unknown']) self.assertEqual(self.calls, []) def test_limit_check_under(self): self._stub_get_project_quotas() self.assertRaises(exception.InvalidQuotaValue, self.driver.limit_check, FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, dict(metadata_items=-1)) def test_limit_check_over(self): self._stub_get_project_quotas() self.assertRaises(exception.OverQuota, self.driver.limit_check, FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, dict(metadata_items=129)) def test_limit_check_project_overs(self): self._stub_get_project_quotas() self.assertRaises(exception.OverQuota, self.driver.limit_check, FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, dict(injected_file_content_bytes=10241, injected_file_path_bytes=256)) def test_limit_check_unlimited(self): self.flags(metadata_items=-1, group='quota') self._stub_get_project_quotas() self.driver.limit_check(FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, dict(metadata_items=32767)) def test_limit_check(self): self._stub_get_project_quotas() self.driver.limit_check(FakeContext('test_project', 'test_class'), quota.QUOTAS._resources, dict(metadata_items=128)) def test_limit_check_project_and_user_no_values(self): self.assertRaises(exception.Invalid, self.driver.limit_check_project_and_user, FakeContext('test_project', 'test_class'), quota.QUOTAS._resources) def test_limit_check_project_and_user_under(self): self._stub_get_project_quotas() ctxt = FakeContext('test_project', 'test_class') resources = self._get_fake_countable_resources() # Check: only project_values, only user_values, and then both. kwargs = [{'project_values': {'fixed_ips': -1}}, {'user_values': {'key_pairs': -1}}, {'project_values': {'instances': -1}, 'user_values': {'instances': -1}}] for kwarg in kwargs: self.assertRaises(exception.InvalidQuotaValue, self.driver.limit_check_project_and_user, ctxt, resources, **kwarg) def test_limit_check_project_and_user_over_project(self): # Check the case where user_values pass user quota but project_values # exceed project quota. self.flags(instances=5, group='quota') self._stub_get_project_quotas() resources = self._get_fake_countable_resources() self.assertRaises(exception.OverQuota, self.driver.limit_check_project_and_user, FakeContext('test_project', 'test_class'), resources, project_values=dict(instances=6), user_values=dict(instances=5)) def test_limit_check_project_and_user_over_user(self): self.flags(instances=5, group='quota') self._stub_get_project_quotas() resources = self._get_fake_countable_resources() # It's not realistic for user_values to be higher than project_values, # but this is just for testing the fictional case where project_values # pass project quota but user_values exceed user quota. self.assertRaises(exception.OverQuota, self.driver.limit_check_project_and_user, FakeContext('test_project', 'test_class'), resources, project_values=dict(instances=5), user_values=dict(instances=6)) def test_limit_check_project_and_user_overs(self): self._stub_get_project_quotas() ctxt = FakeContext('test_project', 'test_class') resources = self._get_fake_countable_resources() # Check: only project_values, only user_values, and then both. kwargs = [{'project_values': {'instances': 512}}, {'user_values': {'key_pairs': 256}}, {'project_values': {'instances': 512}, 'user_values': {'instances': 256}}] for kwarg in kwargs: self.assertRaises(exception.OverQuota, self.driver.limit_check_project_and_user, ctxt, resources, **kwarg) def test_limit_check_project_and_user_unlimited(self): self.flags(key_pairs=-1, group='quota') self.flags(instances=-1, group='quota') self._stub_get_project_quotas() ctxt = FakeContext('test_project', 'test_class') resources = self._get_fake_countable_resources() # Check: only project_values, only user_values, and then both. kwargs = [{'project_values': {'fixed_ips': 32767}}, {'user_values': {'key_pairs': 32767}}, {'project_values': {'instances': 32767}, 'user_values': {'instances': 32767}}] for kwarg in kwargs: self.driver.limit_check_project_and_user(ctxt, resources, **kwarg) def test_limit_check_project_and_user(self): self._stub_get_project_quotas() ctxt = FakeContext('test_project', 'test_class') resources = self._get_fake_countable_resources() # Check: only project_values, only user_values, and then both. kwargs = [{'project_values': {'fixed_ips': 5}}, {'user_values': {'key_pairs': 5}}, {'project_values': {'instances': 5}, 'user_values': {'instances': 5}}] for kwarg in kwargs: self.driver.limit_check_project_and_user(ctxt, resources, **kwarg) def test_limit_check_project_and_user_zero_values(self): """Tests to make sure that we don't compare 0 to None and fail with a TypeError in python 3 when calculating merged_values between project_values and user_values. """ self._stub_get_project_quotas() ctxt = FakeContext('test_project', 'test_class') resources = self._get_fake_countable_resources() # Check: only project_values, only user_values, and then both. kwargs = [{'project_values': {'fixed_ips': 0}}, {'user_values': {'key_pairs': 0}}, {'project_values': {'instances': 0}, 'user_values': {'instances': 0}}] for kwarg in kwargs: self.driver.limit_check_project_and_user(ctxt, resources, **kwarg) class NoopQuotaDriverTestCase(test.TestCase): def setUp(self): super(NoopQuotaDriverTestCase, self).setUp() self.flags(instances=10, cores=20, ram=50 * 1024, metadata_items=128, injected_files=5, injected_file_content_bytes=10 * 1024, injected_file_path_length=255, group='quota' ) self.expected_with_usages = {} self.expected_without_usages = {} self.expected_without_dict = {} self.expected_settable_quotas = {} for r in quota.QUOTAS._resources: self.expected_with_usages[r] = dict(limit=-1, in_use=-1) self.expected_without_usages[r] = dict(limit=-1) self.expected_without_dict[r] = -1 self.expected_settable_quotas[r] = dict(minimum=0, maximum=-1) self.driver = quota.NoopQuotaDriver() def test_get_defaults(self): # Use our pre-defined resources result = self.driver.get_defaults(None, quota.QUOTAS._resources) self.assertEqual(self.expected_without_dict, result) def test_get_class_quotas(self): result = self.driver.get_class_quotas(None, quota.QUOTAS._resources, 'test_class') self.assertEqual(self.expected_without_dict, result) def test_get_project_quotas(self): result = self.driver.get_project_quotas(None, quota.QUOTAS._resources, 'test_project') self.assertEqual(self.expected_with_usages, result) def test_get_user_quotas(self): result = self.driver.get_user_quotas(None, quota.QUOTAS._resources, 'test_project', 'fake_user') self.assertEqual(self.expected_with_usages, result) def test_get_project_quotas_no_usages(self): result = self.driver.get_project_quotas(None, quota.QUOTAS._resources, 'test_project', usages=False) self.assertEqual(self.expected_without_usages, result) def test_get_user_quotas_no_usages(self): result = self.driver.get_user_quotas(None, quota.QUOTAS._resources, 'test_project', 'fake_user', usages=False) self.assertEqual(self.expected_without_usages, result) def test_get_settable_quotas_with_user(self): result = self.driver.get_settable_quotas(None, quota.QUOTAS._resources, 'test_project', 'fake_user') self.assertEqual(self.expected_settable_quotas, result) def test_get_settable_quotas_without_user(self): result = self.driver.get_settable_quotas(None, quota.QUOTAS._resources, 'test_project') self.assertEqual(self.expected_settable_quotas, result) @ddt.ddt class QuotaCountTestCase(test.NoDBTestCase): @mock.patch('nova.scheduler.client.report.SchedulerReportClient.' 'get_usages_counts_for_quota') def test_cores_ram_count_placement(self, mock_get_usages): usages = quota._cores_ram_count_placement( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) mock_get_usages.assert_called_once_with( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) self.assertEqual(mock_get_usages.return_value, usages) @mock.patch('nova.objects.InstanceMappingList.get_counts') @mock.patch('nova.quota._cores_ram_count_placement') def test_instances_cores_ram_count_api_db_placement( self, mock_placement_count, mock_get_im_count): # Fake response from placement with project and user usages of cores # and ram. mock_placement_count.return_value = {'project': {'cores': 2, 'ram': 4}, 'user': {'cores': 1, 'ram': 2}} # Fake count of instances based on instance mappings in the API DB. mock_get_im_count.return_value = {'project': {'instances': 2}, 'user': {'instances': 1}} counts = quota._instances_cores_ram_count_api_db_placement( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) mock_get_im_count.assert_called_once_with( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) mock_placement_count.assert_called_once_with( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) expected = {'project': {'instances': 2, 'cores': 2, 'ram': 4}, 'user': {'instances': 1, 'cores': 1, 'ram': 2}} self.assertDictEqual(expected, counts) @ddt.data((True, True), (True, False), (False, True), (False, False)) @ddt.unpack @mock.patch('nova.quota.LOG.warning') @mock.patch('nova.quota._user_id_queued_for_delete_populated') @mock.patch('nova.quota._instances_cores_ram_count_legacy') @mock.patch('nova.quota._instances_cores_ram_count_api_db_placement') def test_instances_cores_ram_count(self, quota_from_placement, uid_qfd_populated, mock_api_db_placement_count, mock_legacy_count, mock_uid_qfd_populated, mock_warn_log): # Check that all the combinations of # [quota]count_usage_from_placement (True/False) and # user_id_queued_for_delete_populated (True/False) do the right things. # Fake count of instances, cores, and ram. expected = {'project': {'instances': 2, 'cores': 2, 'ram': 4}, 'user': {'instances': 1, 'cores': 1, 'ram': 2}} mock_api_db_placement_count.return_value = expected mock_legacy_count.return_value = expected # user_id and queued_for_delete populated/migrated (True/False) mock_uid_qfd_populated.return_value = uid_qfd_populated # Counting quota usage from placement enabled (True/False) self.flags(count_usage_from_placement=quota_from_placement, group='quota') counts = quota._instances_cores_ram_count( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) if quota_from_placement and uid_qfd_populated: # If we are counting quota usage from placement and user_id and # queued_for_delete data has all been migrated, we should count # instances from the API DB using instance mappings and count # cores and ram from placement. mock_api_db_placement_count.assert_called_once_with( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) # We should not have called the legacy counting method. mock_legacy_count.assert_not_called() # We should not have logged a warn message saying we were falling # back to the legacy counting method. mock_warn_log.assert_not_called() else: # If counting quota usage from placement is not enabled or if # user_id or queued_for_delete data has not all been migrated yet, # we should use the legacy counting method. mock_legacy_count.assert_called_once_with( mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) # We should have logged a warn message saying we were falling back # to the legacy counting method. if quota_from_placement: # We only log the message if someone has opted in to counting # from placement. mock_warn_log.assert_called_once() else: mock_warn_log.assert_not_called() # We should not have called the API DB and placement counting # method. mock_api_db_placement_count.assert_not_called() self.assertDictEqual(expected, counts) @mock.patch('nova.quota._user_id_queued_for_delete_populated') @mock.patch('nova.quota._instances_cores_ram_count_legacy') @mock.patch('nova.quota._instances_cores_ram_count_api_db_placement') def test_user_id_queued_for_delete_populated_cache_by_project( self, mock_api_db_placement_count, mock_legacy_count, mock_uid_qfd_populated): # We need quota usage from placement enabled to test this. For legacy # counting, the cache is not used. self.flags(count_usage_from_placement=True, group='quota') # Fake count of instances, cores, and ram. fake_counts = {'project': {'instances': 2, 'cores': 2, 'ram': 4}, 'user': {'instances': 1, 'cores': 1, 'ram': 2}} mock_api_db_placement_count.return_value = fake_counts mock_legacy_count.return_value = fake_counts # First, check the case where user_id and queued_for_delete are found # not to be migrated. mock_uid_qfd_populated.return_value = False quota._instances_cores_ram_count(mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) mock_uid_qfd_populated.assert_called_once() # The second call should check for unmigrated records again, since the # project was found not to be completely migrated last time. quota._instances_cores_ram_count(mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) self.assertEqual(2, mock_uid_qfd_populated.call_count) # Now check the case where the data migration was found to be complete. mock_uid_qfd_populated.reset_mock() mock_uid_qfd_populated.return_value = True # The first call will check whether there are any unmigrated records. quota._instances_cores_ram_count(mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) mock_uid_qfd_populated.assert_called_once() # Second call should skip the check for user_id and queued_for_delete # migrated because the result was cached. mock_uid_qfd_populated.reset_mock() quota._instances_cores_ram_count(mock.sentinel.context, mock.sentinel.project_id, user_id=mock.sentinel.user_id) mock_uid_qfd_populated.assert_not_called() @mock.patch('nova.quota._user_id_queued_for_delete_populated') @mock.patch('nova.quota._server_group_count_members_by_user_legacy') @mock.patch('nova.objects.InstanceMappingList.get_count_by_uuids_and_user') @mock.patch('nova.quota._instances_cores_ram_count_legacy') @mock.patch('nova.quota._instances_cores_ram_count_api_db_placement') def test_user_id_queued_for_delete_populated_cache_all( self, mock_api_db_placement_count, mock_legacy_icr_count, mock_api_db_sgm_count, mock_legacy_sgm_count, mock_uid_qfd_populated): # Check the case where the data migration was found to be complete by a # server group members count not scoped to a project. mock_uid_qfd_populated.return_value = True # Server group members call will check whether there are any unmigrated # records. fake_group = mock.Mock() quota._server_group_count_members_by_user(mock.sentinel.context, fake_group, mock.sentinel.user_id) mock_uid_qfd_populated.assert_called_once() # Second server group members call should skip the check for user_id # and queued_for_delete migrated because the result was cached. mock_uid_qfd_populated.reset_mock() quota._server_group_count_members_by_user(mock.sentinel.context, fake_group, mock.sentinel.user_id) mock_uid_qfd_populated.assert_not_called() # A call to count instances, cores, and ram should skip the check for # user_id and queued_for_delete migrated because the result was cached # during the call to count server group members. mock_uid_qfd_populated.reset_mock() quota._instances_cores_ram_count(mock.sentinel.context, mock.sentinel.project_id) mock_uid_qfd_populated.assert_not_called()