Merge "Add quota tests in unit tests."

This commit is contained in:
Jenkins 2016-04-15 23:41:18 +00:00 committed by Gerrit Code Review
commit 7a35d352da
3 changed files with 114 additions and 0 deletions

View File

@ -1461,6 +1461,28 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
self.assertEqual(data['NeutronError']['message'], msg)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_ports_native_quotas(self):
quota = 1
cfg.CONF.set_override('quota_port', quota, group='QUOTAS')
with self.network() as network:
res = self._create_port(self.fmt, network['network']['id'])
self.assertEqual(webob.exc.HTTPCreated.code, res.status_int)
res = self._create_port(self.fmt, network['network']['id'])
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_ports_bulk_native_quotas(self):
if self._skip_native_bulk:
self.skipTest("Plugin does not support native bulk port create")
quota = 4
cfg.CONF.set_override('quota_port', quota, group='QUOTAS')
with self.network() as network:
res = self._create_port_bulk(self.fmt, quota + 1,
network['network']['id'],
'test', True)
self._validate_behavior_on_bulk_failure(
res, 'ports',
errcode=webob.exc.HTTPConflict.code)
def test_update_port_update_ip(self):
"""Test update of port IP.
@ -2503,6 +2525,16 @@ class TestNetworksV2(NeutronDbPluginV2TestCase):
res = self._create_network_bulk(self.fmt, 2, 'test', True)
self._validate_behavior_on_bulk_success(res, 'networks')
def test_create_networks_native_quotas(self):
quota = 1
cfg.CONF.set_override('quota_network', quota, group='QUOTAS')
res = self._create_network(fmt=self.fmt, name='net',
admin_state_up=True)
self.assertEqual(webob.exc.HTTPCreated.code, res.status_int)
res = self._create_network(fmt=self.fmt, name='net',
admin_state_up=True)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_networks_bulk_native_quotas(self):
if self._skip_native_bulk:
self.skipTest("Plugin does not support native bulk network create")
@ -4476,6 +4508,32 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_subnets_native_quotas(self):
quota = 1
cfg.CONF.set_override('quota_subnet', quota, group='QUOTAS')
with self.network() as network:
res = self._create_subnet(
self.fmt, network['network']['id'], '10.0.0.0/24',
tenant_id=network['network']['tenant_id'])
self.assertEqual(webob.exc.HTTPCreated.code, res.status_int)
res = self._create_subnet(
self.fmt, network['network']['id'], '10.1.0.0/24',
tenant_id=network['network']['tenant_id'])
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_subnets_bulk_native_quotas(self):
if self._skip_native_bulk:
self.skipTest("Plugin does not support native bulk subnet create")
quota = 4
cfg.CONF.set_override('quota_subnet', quota, group='QUOTAS')
with self.network() as network:
res = self._create_subnet_bulk(self.fmt, quota + 1,
network['network']['id'],
'test')
self._validate_behavior_on_bulk_failure(
res, 'subnets',
errcode=webob.exc.HTTPConflict.code)
def test_show_subnet(self):
with self.network() as network:
with self.subnet(network=network) as subnet:

View File

@ -678,6 +678,15 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
)
self.assertEqual(exc.HTTPForbidden.code, res.status_int)
def test_create_routers_native_quotas(self):
tenant_id = _uuid()
quota = 1
cfg.CONF.set_override('quota_router', quota, group='QUOTAS')
res = self._create_router(self.fmt, tenant_id)
self.assertEqual(exc.HTTPCreated.code, res.status_int)
res = self._create_router(self.fmt, tenant_id)
self.assertEqual(exc.HTTPConflict.code, res.status_int)
def test_router_list(self):
with self.router() as v1, self.router() as v2, self.router() as v3:
routers = (v1, v2, v3)
@ -2623,6 +2632,22 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
floating_ip='10.0.0.10',
http_status=exc.HTTPConflict.code)
def test_create_floatingips_native_quotas(self):
quota = 1
cfg.CONF.set_override('quota_floatingip', quota, group='QUOTAS')
with self.subnet() as public_sub:
self._set_net_external(public_sub['subnet']['network_id'])
res = self._create_floatingip(
self.fmt,
public_sub['subnet']['network_id'],
subnet_id=public_sub['subnet']['id'])
self.assertEqual(exc.HTTPCreated.code, res.status_int)
res = self._create_floatingip(
self.fmt,
public_sub['subnet']['network_id'],
subnet_id=public_sub['subnet']['id'])
self.assertEqual(exc.HTTPConflict.code, res.status_int)
def test_router_specify_id_backend(self):
plugin = manager.NeutronManager.get_service_plugins()[
service_constants.L3_ROUTER_NAT]

View File

@ -16,6 +16,7 @@
import contextlib
import mock
from oslo_config import cfg
import oslo_db.exception as exc
import six
import testtools
@ -1505,6 +1506,36 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.deserialize(self.fmt, res)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_security_groups_native_quotas(self):
quota = 1
cfg.CONF.set_override('quota_security_group', quota, group='QUOTAS')
name = 'quota_test'
description = 'quota_test'
res = self._create_security_group(self.fmt, name, description)
self.assertEqual(webob.exc.HTTPCreated.code, res.status_int)
res = self._create_security_group(self.fmt, name, description)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_security_group_rules_native_quotas(self):
# avoid the number of default security group rules
quota = 7
cfg.CONF.set_override(
'quota_security_group_rule', quota, group='QUOTAS')
name = 'quota_test'
description = 'quota_test'
with self.security_group(name, description) as sg:
security_group_id = sg['security_group']['id']
rule = self._build_security_group_rule(
security_group_id, 'ingress',
const.PROTO_NAME_TCP, '22', '22')
res = self._create_security_group_rule(self.fmt, rule)
self.assertEqual(webob.exc.HTTPCreated.code, res.status_int)
rule = self._build_security_group_rule(
security_group_id, 'egress',
const.PROTO_NAME_TCP, '22', '22')
res = self._create_security_group_rule(self.fmt, rule)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
def test_create_security_group_rule_different_security_group_ids(self):
if self._skip_native_bulk:
self.skipTest("Plugin does not support native bulk "