nova/nova/tests/test_quota.py
Dan Prince 1f644d2105 Implement security group quotas.
Fixes LP Bug #969545

Change-Id: I60d4300aa04597e2d8b0eea31ab0303b0b3c48f9
2012-04-19 01:18:14 -07:00

658 lines
27 KiB
Python

# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
from nova import compute
from nova import context
from nova import db
from nova import flags
from nova import quota
from nova import exception
from nova import rpc
from nova import test
from nova import volume
from nova.compute import instance_types
from nova.scheduler import driver as scheduler_driver
FLAGS = flags.FLAGS
class GetQuotaTestCase(test.TestCase):
def setUp(self):
super(GetQuotaTestCase, self).setUp()
self.flags(quota_instances=10,
quota_cores=20,
quota_ram=50 * 1024,
quota_volumes=10,
quota_gigabytes=1000,
quota_floating_ips=10,
quota_security_groups=10,
quota_security_group_rules=20,
quota_metadata_items=128,
quota_injected_files=5,
quota_injected_file_content_bytes=10 * 1024)
self.context = context.RequestContext('admin', 'admin', is_admin=True)
def _stub_class(self):
def fake_quota_class_get_all_by_name(context, quota_class):
result = dict(class_name=quota_class)
if quota_class == 'test_class':
result.update(
instances=5,
cores=10,
ram=25 * 1024,
volumes=5,
gigabytes=500,
floating_ips=5,
quota_security_groups=10,
quota_security_group_rules=20,
metadata_items=64,
injected_files=2,
injected_file_content_bytes=5 * 1024,
invalid_quota=100,
)
return result
self.stubs.Set(db, 'quota_class_get_all_by_name',
fake_quota_class_get_all_by_name)
def _stub_project(self, override=False):
def fake_quota_get_all_by_project(context, project_id):
result = dict(project_id=project_id)
if override:
result.update(
instances=2,
cores=5,
ram=12 * 1024,
volumes=2,
gigabytes=250,
floating_ips=2,
security_groups=5,
security_group_rules=10,
metadata_items=32,
injected_files=1,
injected_file_content_bytes=2 * 1024,
invalid_quota=50,
)
return result
self.stubs.Set(db, 'quota_get_all_by_project',
fake_quota_get_all_by_project)
def test_default_quotas(self):
result = quota._get_default_quotas()
self.assertEqual(result, dict(
instances=10,
cores=20,
ram=50 * 1024,
volumes=10,
gigabytes=1000,
floating_ips=10,
security_groups=10,
security_group_rules=20,
metadata_items=128,
injected_files=5,
injected_file_content_bytes=10 * 1024,
))
def test_default_quotas_unlimited(self):
self.flags(quota_instances=-1,
quota_cores=-1,
quota_ram=-1,
quota_volumes=-1,
quota_gigabytes=-1,
quota_floating_ips=-1,
quota_security_groups=-1,
quota_security_group_rules=-1,
quota_metadata_items=-1,
quota_injected_files=-1,
quota_injected_file_content_bytes=-1)
result = quota._get_default_quotas()
self.assertEqual(result, dict(
instances=-1,
cores=-1,
ram=-1,
volumes=-1,
gigabytes=-1,
floating_ips=-1,
security_groups=-1,
security_group_rules=-1,
metadata_items=-1,
injected_files=-1,
injected_file_content_bytes=-1,
))
def test_class_quotas_noclass(self):
self._stub_class()
result = quota.get_class_quotas(self.context, 'noclass')
self.assertEqual(result, dict(
instances=10,
cores=20,
ram=50 * 1024,
volumes=10,
gigabytes=1000,
floating_ips=10,
security_groups=10,
security_group_rules=20,
metadata_items=128,
injected_files=5,
injected_file_content_bytes=10 * 1024,
))
def test_class_quotas(self):
self._stub_class()
result = quota.get_class_quotas(self.context, 'test_class')
self.assertEqual(result, dict(
instances=5,
cores=10,
ram=25 * 1024,
volumes=5,
gigabytes=500,
floating_ips=5,
security_groups=10,
security_group_rules=20,
metadata_items=64,
injected_files=2,
injected_file_content_bytes=5 * 1024,
))
def test_project_quotas_defaults_noclass(self):
self._stub_class()
self._stub_project()
result = quota.get_project_quotas(self.context, 'admin')
self.assertEqual(result, dict(
instances=10,
cores=20,
ram=50 * 1024,
volumes=10,
gigabytes=1000,
floating_ips=10,
security_groups=10,
security_group_rules=20,
metadata_items=128,
injected_files=5,
injected_file_content_bytes=10 * 1024,
))
def test_project_quotas_overrides_noclass(self):
self._stub_class()
self._stub_project(True)
result = quota.get_project_quotas(self.context, 'admin')
self.assertEqual(result, dict(
instances=2,
cores=5,
ram=12 * 1024,
volumes=2,
gigabytes=250,
floating_ips=2,
security_groups=5,
security_group_rules=10,
metadata_items=32,
injected_files=1,
injected_file_content_bytes=2 * 1024,
))
def test_project_quotas_defaults_withclass(self):
self._stub_class()
self._stub_project()
self.context.quota_class = 'test_class'
result = quota.get_project_quotas(self.context, 'admin')
self.assertEqual(result, dict(
instances=5,
cores=10,
ram=25 * 1024,
volumes=5,
gigabytes=500,
floating_ips=5,
security_groups=10,
security_group_rules=20,
metadata_items=64,
injected_files=2,
injected_file_content_bytes=5 * 1024,
))
def test_project_quotas_overrides_withclass(self):
self._stub_class()
self._stub_project(True)
self.context.quota_class = 'test_class'
result = quota.get_project_quotas(self.context, 'admin')
self.assertEqual(result, dict(
instances=2,
cores=5,
ram=12 * 1024,
volumes=2,
gigabytes=250,
floating_ips=2,
security_groups=5,
security_group_rules=10,
metadata_items=32,
injected_files=1,
injected_file_content_bytes=2 * 1024,
))
class QuotaTestCase(test.TestCase):
class StubImageService(object):
def show(self, *args, **kwargs):
return {"properties": {}}
def setUp(self):
super(QuotaTestCase, self).setUp()
self.flags(connection_type='fake',
quota_instances=2,
quota_cores=4,
quota_volumes=2,
quota_gigabytes=20,
quota_floating_ips=1,
network_manager='nova.network.manager.FlatDHCPManager')
self.network = self.network = self.start_service('network')
self.user_id = 'admin'
self.project_id = 'admin'
self.context = context.RequestContext(self.user_id,
self.project_id,
is_admin=True)
orig_rpc_call = rpc.call
def rpc_call_wrapper(context, topic, msg):
"""Stub out the scheduler creating the instance entry"""
if (topic == FLAGS.scheduler_topic and
msg['method'] == 'run_instance'):
scheduler = scheduler_driver.Scheduler
instance = scheduler().create_instance_db_entry(
context,
msg['args']['request_spec'])
return [scheduler_driver.encode_instance(instance)]
else:
return orig_rpc_call(context, topic, msg)
self.stubs.Set(rpc, 'call', rpc_call_wrapper)
def _create_instance(self, cores=2):
"""Create a test instance"""
inst = {}
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['instance_type_id'] = '3' # m1.large
inst['vcpus'] = cores
return db.instance_create(self.context, inst)['id']
def _create_volume(self, size=10):
"""Create a test volume"""
vol = {}
vol['user_id'] = self.user_id
vol['project_id'] = self.project_id
vol['size'] = size
return db.volume_create(self.context, vol)['id']
def _get_instance_type(self, name):
instance_types = {
'm1.tiny': dict(memory_mb=512, vcpus=1, root_gb=0, flavorid=1),
'm1.small': dict(memory_mb=2048, vcpus=1, root_gb=20, flavorid=2),
'm1.medium':
dict(memory_mb=4096, vcpus=2, root_gb=40, flavorid=3),
'm1.large': dict(memory_mb=8192, vcpus=4, root_gb=80, flavorid=4),
'm1.xlarge':
dict(memory_mb=16384, vcpus=8, root_gb=160, flavorid=5),
'm1.nocpu': dict(memory_mb=512, vcpus=0, root_gb=0, flavorid=6),
'm1.nomem': dict(memory_mb=0, vcpus=1, root_gb=0, flavorid=7)}
return instance_types[name]
def test_quota_no_mem_no_cpu(self):
num_instances = quota.allowed_instances(self.context, 100,
self._get_instance_type('m1.nocpu'))
self.assertEqual(num_instances, 2)
num_instances = quota.allowed_instances(self.context, 100,
self._get_instance_type('m1.nomem'))
self.assertEqual(num_instances, 2)
def test_quota_overrides(self):
"""Make sure overriding a projects quotas works"""
num_instances = quota.allowed_instances(self.context, 100,
self._get_instance_type('m1.small'))
self.assertEqual(num_instances, 2)
db.quota_create(self.context, self.project_id, 'instances', 10)
num_instances = quota.allowed_instances(self.context, 100,
self._get_instance_type('m1.small'))
self.assertEqual(num_instances, 4)
db.quota_create(self.context, self.project_id, 'cores', 100)
num_instances = quota.allowed_instances(self.context, 100,
self._get_instance_type('m1.small'))
self.assertEqual(num_instances, 10)
db.quota_create(self.context, self.project_id, 'ram', 3 * 2048)
num_instances = quota.allowed_instances(self.context, 100,
self._get_instance_type('m1.small'))
self.assertEqual(num_instances, 3)
# metadata_items
too_many_items = FLAGS.quota_metadata_items + 1000
num_metadata_items = quota.allowed_metadata_items(self.context,
too_many_items)
self.assertEqual(num_metadata_items, FLAGS.quota_metadata_items)
db.quota_create(self.context, self.project_id, 'metadata_items', 5)
num_metadata_items = quota.allowed_metadata_items(self.context,
too_many_items)
self.assertEqual(num_metadata_items, 5)
# Cleanup
db.quota_destroy_all_by_project(self.context, self.project_id)
def test_unlimited_instances(self):
self.flags(quota_instances=2, quota_ram=-1, quota_cores=-1)
instance_type = self._get_instance_type('m1.small')
num_instances = quota.allowed_instances(self.context, 100,
instance_type)
self.assertEqual(num_instances, 2)
db.quota_create(self.context, self.project_id, 'instances', -1)
num_instances = quota.allowed_instances(self.context, 100,
instance_type)
self.assertEqual(num_instances, 100)
num_instances = quota.allowed_instances(self.context, 101,
instance_type)
self.assertEqual(num_instances, 101)
def test_unlimited_ram(self):
self.flags(quota_instances=-1, quota_ram=2 * 2048, quota_cores=-1)
instance_type = self._get_instance_type('m1.small')
num_instances = quota.allowed_instances(self.context, 100,
instance_type)
self.assertEqual(num_instances, 2)
db.quota_create(self.context, self.project_id, 'ram', -1)
num_instances = quota.allowed_instances(self.context, 100,
instance_type)
self.assertEqual(num_instances, 100)
num_instances = quota.allowed_instances(self.context, 101,
instance_type)
self.assertEqual(num_instances, 101)
def test_unlimited_cores(self):
self.flags(quota_instances=-1, quota_ram=-1, quota_cores=2)
instance_type = self._get_instance_type('m1.small')
num_instances = quota.allowed_instances(self.context, 100,
instance_type)
self.assertEqual(num_instances, 2)
db.quota_create(self.context, self.project_id, 'cores', -1)
num_instances = quota.allowed_instances(self.context, 100,
instance_type)
self.assertEqual(num_instances, 100)
num_instances = quota.allowed_instances(self.context, 101,
instance_type)
self.assertEqual(num_instances, 101)
def test_unlimited_volumes(self):
self.flags(quota_volumes=10, quota_gigabytes=-1)
volumes = quota.allowed_volumes(self.context, 100, 1)
self.assertEqual(volumes, 10)
db.quota_create(self.context, self.project_id, 'volumes', -1)
volumes = quota.allowed_volumes(self.context, 100, 1)
self.assertEqual(volumes, 100)
volumes = quota.allowed_volumes(self.context, 101, 1)
self.assertEqual(volumes, 101)
def test_unlimited_gigabytes(self):
self.flags(quota_volumes=-1, quota_gigabytes=10)
volumes = quota.allowed_volumes(self.context, 100, 1)
self.assertEqual(volumes, 10)
db.quota_create(self.context, self.project_id, 'gigabytes', -1)
volumes = quota.allowed_volumes(self.context, 100, 1)
self.assertEqual(volumes, 100)
volumes = quota.allowed_volumes(self.context, 101, 1)
self.assertEqual(volumes, 101)
def test_unlimited_floating_ips(self):
self.flags(quota_floating_ips=10)
floating_ips = quota.allowed_floating_ips(self.context, 100)
self.assertEqual(floating_ips, 10)
db.quota_create(self.context, self.project_id, 'floating_ips', -1)
floating_ips = quota.allowed_floating_ips(self.context, 100)
self.assertEqual(floating_ips, 100)
floating_ips = quota.allowed_floating_ips(self.context, 101)
self.assertEqual(floating_ips, 101)
def test_unlimited_security_groups(self):
self.flags(quota_security_groups=10)
security_groups = quota.allowed_security_groups(self.context, 100)
self.assertEqual(security_groups, 10)
db.quota_create(self.context, self.project_id, 'security_groups', -1)
security_groups = quota.allowed_security_groups(self.context, 100)
self.assertEqual(security_groups, 100)
security_groups = quota.allowed_security_groups(self.context, 101)
self.assertEqual(security_groups, 101)
def test_unlimited_security_group_rules(self):
def fake_security_group_rule_count_by_group(context, sec_group_id):
return 0
self.stubs.Set(db, 'security_group_rule_count_by_group',
fake_security_group_rule_count_by_group)
self.flags(quota_security_group_rules=20)
rules = quota.allowed_security_group_rules(self.context, 1234, 100)
self.assertEqual(rules, 20)
db.quota_create(self.context, self.project_id, 'security_group_rules',
-1)
rules = quota.allowed_security_group_rules(self.context, 1234, 100)
self.assertEqual(rules, 100)
rules = quota.allowed_security_group_rules(self.context, 1234, 101)
self.assertEqual(rules, 101)
def test_unlimited_metadata_items(self):
self.flags(quota_metadata_items=10)
items = quota.allowed_metadata_items(self.context, 100)
self.assertEqual(items, 10)
db.quota_create(self.context, self.project_id, 'metadata_items', -1)
items = quota.allowed_metadata_items(self.context, 100)
self.assertEqual(items, 100)
items = quota.allowed_metadata_items(self.context, 101)
self.assertEqual(items, 101)
def test_too_many_instances(self):
instance_ids = []
for i in range(FLAGS.quota_instances):
instance_id = self._create_instance()
instance_ids.append(instance_id)
inst_type = instance_types.get_instance_type_by_name('m1.small')
image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175'
self.assertRaises(exception.QuotaError, compute.API().create,
self.context,
min_count=1,
max_count=1,
instance_type=inst_type,
image_href=image_uuid)
for instance_id in instance_ids:
db.instance_destroy(self.context, instance_id)
def test_too_many_cores(self):
instance_ids = []
instance_id = self._create_instance(cores=4)
instance_ids.append(instance_id)
inst_type = instance_types.get_instance_type_by_name('m1.small')
image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175'
self.assertRaises(exception.QuotaError, compute.API().create,
self.context,
min_count=1,
max_count=1,
instance_type=inst_type,
image_href=image_uuid)
for instance_id in instance_ids:
db.instance_destroy(self.context, instance_id)
def test_too_many_volumes(self):
volume_ids = []
for i in range(FLAGS.quota_volumes):
volume_id = self._create_volume()
volume_ids.append(volume_id)
self.assertRaises(exception.QuotaError,
volume.API().create,
self.context, 10, '', '', None)
for volume_id in volume_ids:
db.volume_destroy(self.context, volume_id)
def test_too_many_gigabytes(self):
volume_ids = []
volume_id = self._create_volume(size=20)
volume_ids.append(volume_id)
self.assertRaises(exception.QuotaError,
volume.API().create,
self.context, 10, '', '', None)
for volume_id in volume_ids:
db.volume_destroy(self.context, volume_id)
def test_too_many_addresses(self):
address = '192.168.0.100'
db.floating_ip_create(context.get_admin_context(),
{'address': address,
'project_id': self.project_id})
self.assertRaises(exception.QuotaError,
self.network.allocate_floating_ip,
self.context,
self.project_id)
db.floating_ip_destroy(context.get_admin_context(), address)
def test_too_many_metadata_items(self):
metadata = {}
for i in range(FLAGS.quota_metadata_items + 1):
metadata['key%s' % i] = 'value%s' % i
inst_type = instance_types.get_instance_type_by_name('m1.small')
image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175'
self.assertRaises(exception.QuotaError, compute.API().create,
self.context,
min_count=1,
max_count=1,
instance_type=inst_type,
image_href=image_uuid,
metadata=metadata)
def test_default_allowed_injected_files(self):
self.flags(quota_injected_files=55)
self.assertEqual(quota.allowed_injected_files(self.context, 100), 55)
def test_overridden_allowed_injected_files(self):
self.flags(quota_injected_files=5)
db.quota_create(self.context, self.project_id, 'injected_files', 77)
self.assertEqual(quota.allowed_injected_files(self.context, 100), 77)
def test_unlimited_default_allowed_injected_files(self):
self.flags(quota_injected_files=-1)
self.assertEqual(quota.allowed_injected_files(self.context, 100), 100)
def test_unlimited_db_allowed_injected_files(self):
self.flags(quota_injected_files=5)
db.quota_create(self.context, self.project_id, 'injected_files', -1)
self.assertEqual(quota.allowed_injected_files(self.context, 100), 100)
def test_default_allowed_injected_file_content_bytes(self):
self.flags(quota_injected_file_content_bytes=12345)
limit = quota.allowed_injected_file_content_bytes(self.context, 23456)
self.assertEqual(limit, 12345)
def test_overridden_allowed_injected_file_content_bytes(self):
self.flags(quota_injected_file_content_bytes=12345)
db.quota_create(self.context, self.project_id,
'injected_file_content_bytes', 5678)
limit = quota.allowed_injected_file_content_bytes(self.context, 23456)
self.assertEqual(limit, 5678)
def test_unlimited_default_allowed_injected_file_content_bytes(self):
self.flags(quota_injected_file_content_bytes=-1)
limit = quota.allowed_injected_file_content_bytes(self.context, 23456)
self.assertEqual(limit, 23456)
def test_unlimited_db_allowed_injected_file_content_bytes(self):
self.flags(quota_injected_file_content_bytes=12345)
db.quota_create(self.context, self.project_id,
'injected_file_content_bytes', -1)
limit = quota.allowed_injected_file_content_bytes(self.context, 23456)
self.assertEqual(limit, 23456)
def _create_with_injected_files(self, files):
self.flags(image_service='nova.image.fake.FakeImageService')
api = compute.API(image_service=self.StubImageService())
inst_type = instance_types.get_instance_type_by_name('m1.small')
image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175'
api.create(self.context, min_count=1, max_count=1,
instance_type=inst_type, image_href=image_uuid,
injected_files=files)
def test_no_injected_files(self):
self.flags(image_service='nova.image.fake.FakeImageService')
api = compute.API(image_service=self.StubImageService())
inst_type = instance_types.get_instance_type_by_name('m1.small')
image_uuid = 'cedef40a-ed67-4d10-800e-17455edce175'
api.create(self.context,
instance_type=inst_type,
image_href=image_uuid)
def test_max_injected_files(self):
files = []
for i in xrange(FLAGS.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 xrange(FLAGS.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 = FLAGS.quota_injected_file_content_bytes
content = ''.join(['a' for i in xrange(max)])
files = [('/test/path', content)]
self._create_with_injected_files(files) # no QuotaError
def test_too_many_injected_file_content_bytes(self):
max = FLAGS.quota_injected_file_content_bytes
content = ''.join(['a' for i in xrange(max + 1)])
files = [('/test/path', content)]
self.assertRaises(exception.QuotaError,
self._create_with_injected_files, files)
def test_allowed_injected_file_path_bytes(self):
self.assertEqual(
quota.allowed_injected_file_path_bytes(self.context),
FLAGS.quota_injected_file_path_bytes)
def test_max_injected_file_path_bytes(self):
max = FLAGS.quota_injected_file_path_bytes
path = ''.join(['a' for i in xrange(max)])
files = [(path, 'config = quotatest')]
self._create_with_injected_files(files) # no QuotaError
def test_too_many_injected_file_path_bytes(self):
max = FLAGS.quota_injected_file_path_bytes
path = ''.join(['a' for i in xrange(max + 1)])
files = [(path, 'config = quotatest')]
self.assertRaises(exception.QuotaError,
self._create_with_injected_files, files)
def test_quota_class_unlimited(self):
self.flags(quota_floating_ips=10)
items = quota.allowed_floating_ips(self.context, 10)
self.assertEqual(items, 10)
self.context.quota_class = 'foo'
db.quota_class_create(self.context, 'foo', 'floating_ips', -1)
items = quota.allowed_floating_ips(self.context, 100)
self.assertEqual(items, 100)