4f3dbe4fea
This new test is checking that, by default, the quota engine will check the resource usage before setting the quota. If the new quota is below the resource usage, an exception is returned. This patch is also adding the flag "force=True" in other tests setting quotas for resources; these tests do not need the quota engine to verify the current resource usage. Related-Bug: #1953170 Change-Id: Ia193cd5a2aacc4243b5807eb7757b32e66f12365
768 lines
32 KiB
Python
768 lines
32 KiB
Python
# Copyright 2013 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 random
|
|
|
|
from neutron_lib import constants
|
|
from tempest.lib.common.utils import data_utils
|
|
from tempest.lib.common.utils import test_utils
|
|
from tempest.lib import decorators
|
|
from tempest.lib import exceptions
|
|
import testtools
|
|
|
|
from neutron_tempest_plugin.api import base
|
|
from neutron_tempest_plugin.api import base_security_groups
|
|
from oslo_log import log
|
|
|
|
LOG = log.getLogger(__name__)
|
|
|
|
|
|
class BaseSecGroupTest(base.BaseAdminNetworkTest):
|
|
|
|
required_extensions = ['security-group']
|
|
|
|
def _test_create_list_update_show_delete_security_group(self):
|
|
sg_kwargs = {}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
security_group = self.create_security_group(**sg_kwargs)
|
|
|
|
# List security groups and verify if created group is there in response
|
|
security_groups = self.client.list_security_groups()['security_groups']
|
|
self.assertIn(security_group['id'],
|
|
{sg['id'] for sg in security_groups})
|
|
|
|
# Update the security group
|
|
new_name = data_utils.rand_name('security')
|
|
new_description = data_utils.rand_name('security-description')
|
|
updated_security_group = self.client.update_security_group(
|
|
security_group['id'], name=new_name,
|
|
description=new_description)['security_group']
|
|
|
|
# Verify if security group is updated
|
|
self.assertEqual(updated_security_group['name'], new_name)
|
|
self.assertEqual(updated_security_group['description'],
|
|
new_description)
|
|
|
|
# Show details of the updated security group
|
|
observed_security_group = self.client.show_security_group(
|
|
security_group['id'])['security_group']
|
|
self.assertEqual(observed_security_group['name'], new_name)
|
|
self.assertEqual(observed_security_group['description'],
|
|
new_description)
|
|
|
|
def _test_show_security_group_contains_all_rules(self):
|
|
sg_kwargs = {}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
security_group = self.create_security_group(**sg_kwargs)
|
|
protocol = random.choice(list(base_security_groups.V4_PROTOCOL_NAMES))
|
|
security_group_rule = self.create_security_group_rule(
|
|
security_group=security_group,
|
|
project={'id': self.admin_client.project_id},
|
|
client=self.admin_client,
|
|
protocol=protocol,
|
|
direction=constants.INGRESS_DIRECTION)
|
|
|
|
observed_security_group = self.client.show_security_group(
|
|
security_group['id'])['security_group']
|
|
observerd_security_group_rules_ids = [
|
|
sgr['id'] for sgr in
|
|
observed_security_group['security_group_rules']]
|
|
self.assertIn(
|
|
security_group_rule['id'], observerd_security_group_rules_ids)
|
|
|
|
def _test_list_security_group_rules_contains_all_rules(self):
|
|
"""Test list security group rules.
|
|
|
|
This test checks if all SG rules which belongs to the tenant OR
|
|
which belongs to the tenant's security group are listed.
|
|
"""
|
|
sg_kwargs = {}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
security_group = self.create_security_group(**sg_kwargs)
|
|
protocol = random.choice(list(base_security_groups.V4_PROTOCOL_NAMES))
|
|
security_group_rule = self.create_security_group_rule(
|
|
security_group=security_group,
|
|
project={'id': self.admin_client.project_id},
|
|
client=self.admin_client,
|
|
protocol=protocol,
|
|
direction=constants.INGRESS_DIRECTION)
|
|
|
|
# Create also other SG with some custom rule to check that regular user
|
|
# can't see this rule
|
|
sg_kwargs = {
|
|
'project': {'id': self.admin_client.project_id},
|
|
'client': self.admin_client
|
|
}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
admin_security_group = self.create_security_group(**sg_kwargs)
|
|
admin_security_group_rule = self.create_security_group_rule(
|
|
security_group=admin_security_group,
|
|
project={'id': self.admin_client.project_id},
|
|
client=self.admin_client,
|
|
protocol=protocol,
|
|
direction=constants.INGRESS_DIRECTION)
|
|
|
|
rules = self.client.list_security_group_rules()['security_group_rules']
|
|
rules_ids = [rule['id'] for rule in rules]
|
|
self.assertIn(security_group_rule['id'], rules_ids)
|
|
self.assertNotIn(admin_security_group_rule['id'], rules_ids)
|
|
|
|
def _test_create_bulk_sec_groups(self):
|
|
# Creates 2 sec-groups in one request
|
|
sec_nm = [data_utils.rand_name('secgroup'),
|
|
data_utils.rand_name('secgroup')]
|
|
body = self.client.create_bulk_security_groups(
|
|
sec_nm, stateless=self.stateless_sg)
|
|
created_sec_grps = body['security_groups']
|
|
self.assertEqual(2, len(created_sec_grps))
|
|
for secgrp in created_sec_grps:
|
|
self.addCleanup(self.client.delete_security_group,
|
|
secgrp['id'])
|
|
self.assertIn(secgrp['name'], sec_nm)
|
|
self.assertIsNotNone(secgrp['id'])
|
|
|
|
def _test_create_sec_groups_with_the_same_name(self):
|
|
same_name_sg_number = 5
|
|
sg_name = 'sg_zahlabut'
|
|
sg_names = [sg_name] * same_name_sg_number
|
|
sg_kwargs = {}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
for name in sg_names:
|
|
sg_kwargs['name'] = name
|
|
self.create_security_group(**sg_kwargs)
|
|
sec_groups = [item['id'] for item in
|
|
self.client.list_security_groups(
|
|
name=sg_name)['security_groups']]
|
|
self.assertEqual(
|
|
same_name_sg_number, len(set(sec_groups)),
|
|
'Failed - expected number of groups with the same name'
|
|
' is: {}'.format(same_name_sg_number))
|
|
|
|
|
|
class StatefulSecGroupTest(BaseSecGroupTest):
|
|
|
|
stateless_sg = False
|
|
|
|
@decorators.idempotent_id('bfd128e5-3c92-44b6-9d66-7fe29d22c802')
|
|
def test_create_list_update_show_delete_security_group(self):
|
|
self._test_create_list_update_show_delete_security_group()
|
|
|
|
@decorators.idempotent_id('1fff0d57-bb6c-4528-9c1d-2326dce1c087')
|
|
def test_show_security_group_contains_all_rules(self):
|
|
self._test_show_security_group_contains_all_rules()
|
|
|
|
@decorators.idempotent_id('b5923b1a-4d33-44e1-af25-088dcb55b02b')
|
|
def test_list_security_group_rules_contains_all_rules(self):
|
|
self._test_list_security_group_rules_contains_all_rules()
|
|
|
|
@decorators.idempotent_id('7c0ecb10-b2db-11e6-9b14-000c29248b0d')
|
|
def test_create_bulk_sec_groups(self):
|
|
self._test_create_bulk_sec_groups()
|
|
|
|
@decorators.idempotent_id('e93f33d8-57ea-11eb-b69b-74e5f9e2a801')
|
|
def test_create_sec_groups_with_the_same_name(self):
|
|
self._test_create_sec_groups_with_the_same_name()
|
|
|
|
|
|
class StatelessSecGroupTest(BaseSecGroupTest):
|
|
|
|
required_extensions = ['security-group', 'stateful-security-group']
|
|
stateless_sg = True
|
|
|
|
@decorators.idempotent_id('0214d58a-2177-47e1-af83-dcd45c024829')
|
|
def test_create_list_update_show_delete_security_group(self):
|
|
self._test_create_list_update_show_delete_security_group()
|
|
|
|
@decorators.idempotent_id('ddbc0e4c-840f-44ab-8718-0b95b7c7b575')
|
|
def test_show_security_group_contains_all_rules(self):
|
|
self._test_show_security_group_contains_all_rules()
|
|
|
|
@decorators.idempotent_id('cdf3a63a-08fe-4091-bab4-62180847990f')
|
|
def test_list_security_group_rules_contains_all_rules(self):
|
|
self._test_list_security_group_rules_contains_all_rules()
|
|
|
|
@decorators.idempotent_id('b33e612e-65f0-467b-9bf2-b5b2ce67f72f')
|
|
def test_create_bulk_sec_groups(self):
|
|
self._test_create_bulk_sec_groups()
|
|
|
|
@decorators.idempotent_id('a6896935-db18-413d-95f5-4f465e0e2209')
|
|
def test_create_sec_groups_with_the_same_name(self):
|
|
self._test_create_sec_groups_with_the_same_name()
|
|
|
|
@decorators.idempotent_id('0a6c1476-3d1a-11ec-b0ec-0800277ac3d9')
|
|
def test_stateless_security_group_update(self):
|
|
security_group = self.create_security_group(stateful=True)
|
|
|
|
# List security groups and verify if created group is there in response
|
|
security_groups = self.client.list_security_groups()['security_groups']
|
|
found = False
|
|
for sg in security_groups:
|
|
if sg['id'] == security_group['id']:
|
|
found = True
|
|
break
|
|
self.assertTrue(found)
|
|
self.assertTrue(sg['stateful'])
|
|
|
|
# Switch to stateless
|
|
updated_security_group = self.client.update_security_group(
|
|
security_group['id'], stateful=False)['security_group']
|
|
|
|
# Verify if security group is updated
|
|
self.assertFalse(updated_security_group['stateful'])
|
|
|
|
observed_security_group = self.client.show_security_group(
|
|
security_group['id'])['security_group']
|
|
self.assertFalse(observed_security_group['stateful'])
|
|
|
|
# Switch back to stateful
|
|
updated_security_group = self.client.update_security_group(
|
|
security_group['id'], stateful=True)['security_group']
|
|
|
|
# Verify if security group is stateful again
|
|
self.assertTrue(updated_security_group['stateful'])
|
|
|
|
observed_security_group = self.client.show_security_group(
|
|
security_group['id'])['security_group']
|
|
self.assertTrue(observed_security_group['stateful'])
|
|
|
|
|
|
class BaseSecGroupQuota(base.BaseAdminNetworkTest):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
# NOTE(slaweq): we don't know exactly how many rule templates may be
|
|
# created in the neutron db and used for every SG so, as in this test
|
|
# class we are checking quotas of SG, not SG rules, lets set quota for
|
|
# SG rules to -1
|
|
self._set_sg_rules_quota(-1)
|
|
|
|
def _create_max_allowed_sg_amount(self):
|
|
sg_amount = self._get_sg_amount()
|
|
sg_quota = self._get_sg_quota()
|
|
sg_to_create = sg_quota - sg_amount
|
|
self._create_security_groups(sg_to_create)
|
|
|
|
def _create_security_groups(self, amount):
|
|
for _ in range(amount):
|
|
sg = self.create_security_group()
|
|
self.addCleanup(self.delete_security_group, sg)
|
|
|
|
def _increase_sg_quota(self):
|
|
sg_quota = self._get_sg_quota()
|
|
new_sg_quota = 2 * sg_quota
|
|
self._set_sg_quota(new_sg_quota)
|
|
self.assertEqual(self._get_sg_quota(), new_sg_quota,
|
|
"Security group quota wasn't changed correctly")
|
|
|
|
def _decrease_sg_quota(self):
|
|
sg_quota = self._get_sg_quota()
|
|
new_sg_quota = sg_quota // 2
|
|
self._set_sg_quota(new_sg_quota)
|
|
self.assertEqual(self._get_sg_quota(), new_sg_quota,
|
|
"Security group quota wasn't changed correctly")
|
|
|
|
def _set_quota(self, val, resource):
|
|
res_quota = self._get_quota(resource)
|
|
project_id = self.client.project_id
|
|
self.admin_client.update_quotas(project_id, **{resource: val,
|
|
'force': True})
|
|
self.addCleanup(self.admin_client.update_quotas,
|
|
project_id, **{resource: res_quota, 'force': True})
|
|
|
|
def _get_quota(self, resource):
|
|
project_id = self.client.project_id
|
|
quotas = self.admin_client.show_quotas(project_id)
|
|
return quotas['quota'][resource]
|
|
|
|
def _set_sg_quota(self, val):
|
|
return self._set_quota(val, 'security_group')
|
|
|
|
def _get_sg_quota(self):
|
|
return self._get_quota('security_group')
|
|
|
|
def _get_sg_amount(self):
|
|
project_id = self.client.project_id
|
|
filter_query = {'project_id': project_id}
|
|
security_groups = self.client.list_security_groups(**filter_query)
|
|
return len(security_groups['security_groups'])
|
|
|
|
def _set_sg_rules_quota(self, val):
|
|
return self._set_quota(val, 'security_group_rule')
|
|
|
|
|
|
class SecGroupQuotaTest(BaseSecGroupQuota):
|
|
|
|
credentials = ['primary', 'admin']
|
|
required_extensions = ['security-group', 'quotas']
|
|
|
|
@decorators.idempotent_id('1826aa02-090d-4717-b43a-50ee449b02e7')
|
|
def test_sg_quota_values(self):
|
|
values = [-1, 0, 10, 2147483647]
|
|
for value in values:
|
|
self._set_sg_quota(value)
|
|
self.assertEqual(value, self._get_sg_quota())
|
|
|
|
@decorators.idempotent_id('df7981fb-b83a-4779-b13e-65494ef44a72')
|
|
def test_max_allowed_sg_amount(self):
|
|
self._create_max_allowed_sg_amount()
|
|
self.assertEqual(self._get_sg_quota(), self._get_sg_amount())
|
|
|
|
@decorators.idempotent_id('623d909c-6ef8-43d6-93ee-97086e2651e8')
|
|
def test_sg_quota_increased(self):
|
|
self._create_max_allowed_sg_amount()
|
|
self._increase_sg_quota()
|
|
self._create_max_allowed_sg_amount()
|
|
self.assertEqual(self._get_sg_quota(), self._get_sg_amount(),
|
|
"Amount of security groups doesn't match quota")
|
|
|
|
@decorators.idempotent_id('ba95676c-8d9a-4482-b4ec-74d51a4602a6')
|
|
def test_sg_quota_decrease_less_than_created(self):
|
|
self._create_max_allowed_sg_amount()
|
|
self._decrease_sg_quota()
|
|
|
|
@decorators.idempotent_id('d43cf1a7-aa7e-4c41-9340-627a1a6ab961')
|
|
def test_create_sg_when_quota_disabled(self):
|
|
sg_amount = self._get_sg_amount()
|
|
self._set_sg_quota(-1)
|
|
self._create_security_groups(10)
|
|
new_sg_amount = self._get_sg_amount()
|
|
self.assertGreater(new_sg_amount, sg_amount)
|
|
|
|
|
|
class BaseSecGroupRulesQuota(base.BaseAdminNetworkTest):
|
|
|
|
def _create_max_allowed_sg_rules_amount(self, port_index=1):
|
|
sg_rules_amount = self._get_sg_rules_amount()
|
|
sg_rules_quota = self._get_sg_rules_quota()
|
|
sg_rules_to_create = sg_rules_quota - sg_rules_amount
|
|
port_index += sg_rules_to_create
|
|
self._create_security_group_rules(sg_rules_to_create,
|
|
port_index=port_index)
|
|
|
|
def _create_security_group_rules(self, amount, port_index=1):
|
|
for i in range(amount):
|
|
ingress_rule = self.create_security_group_rule(**{
|
|
'project_id': self.client.project_id,
|
|
'direction': 'ingress',
|
|
'port_range_max': port_index + i,
|
|
'port_range_min': port_index + i,
|
|
'protocol': 'tcp'})
|
|
self.addCleanup(
|
|
self.client.delete_security_group_rule, ingress_rule['id'])
|
|
|
|
def _increase_sg_rules_quota(self):
|
|
sg_rules_quota = self._get_sg_rules_quota()
|
|
new_sg_rules_quota = 2 * sg_rules_quota
|
|
self._set_sg_rules_quota(new_sg_rules_quota)
|
|
self.assertGreater(self._get_sg_rules_quota(), sg_rules_quota,
|
|
"Security group rule quota wasnt changed correctly")
|
|
return new_sg_rules_quota
|
|
|
|
def _decrease_sg_rules_quota(self):
|
|
sg_rules_quota = self._get_sg_rules_quota()
|
|
new_sg_rules_quota = sg_rules_quota // 2
|
|
self._set_sg_rules_quota(new_sg_rules_quota)
|
|
return new_sg_rules_quota
|
|
|
|
def _set_sg_rules_quota(self, val):
|
|
project_id = self.client.project_id
|
|
self.admin_client.update_quotas(project_id,
|
|
**{'security_group_rule': val,
|
|
'force': True})
|
|
LOG.info('Trying to update security group rule quota {} '.format(val))
|
|
|
|
def _get_sg_rules_quota(self):
|
|
project_id = self.client.project_id
|
|
quotas = self.admin_client.show_quotas(project_id)
|
|
return quotas['quota']['security_group_rule']
|
|
|
|
def _get_sg_rules_amount(self):
|
|
project_id = self.client.project_id
|
|
filter_query = {'project_id': project_id}
|
|
security_group_rules = self.client.list_security_group_rules(
|
|
**filter_query)
|
|
return len(security_group_rules['security_group_rules'])
|
|
|
|
|
|
class SecGroupRulesQuotaTest(BaseSecGroupRulesQuota):
|
|
|
|
credentials = ['primary', 'admin']
|
|
required_extensions = ['security-group', 'quotas']
|
|
|
|
def setUp(self):
|
|
super(SecGroupRulesQuotaTest, self).setUp()
|
|
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
|
|
self.admin_client.reset_quotas, self.client.project_id)
|
|
self._set_sg_rules_quota(10)
|
|
|
|
@decorators.idempotent_id('77ec038c-5638-11ea-8e2d-0242ac130003')
|
|
def test_sg_rules_quota_increased(self):
|
|
"""Test security group rules quota increased.
|
|
|
|
This test checks if it is possible to increase the SG rules Quota
|
|
value and creates security group rules according to new quota value.
|
|
"""
|
|
self._create_max_allowed_sg_rules_amount()
|
|
new_quota = self._increase_sg_rules_quota()
|
|
port_index = new_quota
|
|
self._create_max_allowed_sg_rules_amount(port_index)
|
|
quota_set = self._get_sg_rules_quota()
|
|
self.assertEqual(quota_set, self._get_sg_rules_amount(),
|
|
"Amount of security groups rules doesn't match quota")
|
|
|
|
@decorators.idempotent_id('37508c8d-270b-4b93-8007-72876a1fec38')
|
|
def test_sg_rules_quota_values(self):
|
|
"""Test security group rules quota values.
|
|
|
|
This test checks if it is possible to change the SG rules Quota
|
|
values, different values.
|
|
"""
|
|
sg_rules_quota = self._get_sg_rules_quota()
|
|
project_id = self.client.project_id
|
|
self.addCleanup(self.admin_client.update_quotas,
|
|
project_id, **{'security_group_rule': sg_rules_quota,
|
|
'force': True})
|
|
values = [-1, 0, 10, 2147483647]
|
|
for value in values:
|
|
self._set_sg_rules_quota(value)
|
|
self.assertEqual(value, self._get_sg_rules_quota())
|
|
|
|
@decorators.idempotent_id('4459e066-d9c8-4a13-9e98-018f95ce2dbf')
|
|
def test_create_sg_rules_when_quota_disabled(self):
|
|
sg_rules_amount = self._get_sg_rules_amount()
|
|
self._set_sg_rules_quota(-1)
|
|
self._create_security_group_rules(10, port_index=100)
|
|
new_sg_rules_amount = self._get_sg_rules_amount()
|
|
self.assertGreater(new_sg_rules_amount, sg_rules_amount)
|
|
|
|
@decorators.idempotent_id('3fc39bd6-3132-4e6f-a09c-456fb18d600c')
|
|
def test_sg_rules_quota_decrease_less_than_created(self):
|
|
self._create_max_allowed_sg_rules_amount()
|
|
new_quota = self._decrease_sg_rules_quota()
|
|
self.assertEqual(self._get_sg_rules_quota(), new_quota)
|
|
|
|
|
|
class BaseSecGroupProtocolTest(base.BaseNetworkTest):
|
|
|
|
protocol_names = base_security_groups.V4_PROTOCOL_NAMES
|
|
protocol_ints = base_security_groups.V4_PROTOCOL_INTS
|
|
|
|
def _test_security_group_rule_protocols(self, protocols):
|
|
sg_kwargs = {}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
security_group = self.create_security_group(**sg_kwargs)
|
|
for protocol in protocols:
|
|
self._test_security_group_rule(
|
|
security_group=security_group,
|
|
protocol=str(protocol),
|
|
direction=constants.INGRESS_DIRECTION,
|
|
ethertype=self.ethertype)
|
|
|
|
def _test_security_group_rule(self, security_group, **kwargs):
|
|
security_group_rule = self.create_security_group_rule(
|
|
security_group=security_group, **kwargs)
|
|
observed_security_group_rule = self.client.show_security_group_rule(
|
|
security_group_rule['id'])['security_group_rule']
|
|
for key, value in kwargs.items():
|
|
self.assertEqual(value, security_group_rule[key],
|
|
"{!r} does not match.".format(key))
|
|
self.assertEqual(value, observed_security_group_rule[key],
|
|
"{!r} does not match.".format(key))
|
|
|
|
|
|
class StatefulSecGroupProtocolTest(BaseSecGroupProtocolTest):
|
|
stateless_sg = False
|
|
|
|
@decorators.idempotent_id('282e3681-aa6e-42a7-b05c-c341aa1e3cdf')
|
|
def test_security_group_rule_protocol_names(self):
|
|
self._test_security_group_rule_protocols(protocols=self.protocol_names)
|
|
|
|
@decorators.idempotent_id('66e47f1f-20b6-4417-8839-3cc671c7afa3')
|
|
def test_security_group_rule_protocol_ints(self):
|
|
self._test_security_group_rule_protocols(protocols=self.protocol_ints)
|
|
|
|
|
|
class StatelessSecGroupProtocolTest(BaseSecGroupProtocolTest):
|
|
required_extensions = ['security-group', 'stateful-security-group']
|
|
stateless_sg = True
|
|
|
|
@decorators.idempotent_id('3a065cdd-99bd-409f-a08e-385c6674bec2')
|
|
def test_security_group_rule_protocol_names(self):
|
|
self._test_security_group_rule_protocols(protocols=self.protocol_names)
|
|
|
|
@decorators.idempotent_id('b0332b5d-6fac-49d5-a79d-ae4fe62600f7')
|
|
def test_security_group_rule_protocol_ints(self):
|
|
self._test_security_group_rule_protocols(protocols=self.protocol_ints)
|
|
|
|
|
|
class BaseSecGroupProtocolIPv6Test(BaseSecGroupProtocolTest):
|
|
|
|
_ip_version = constants.IP_VERSION_6
|
|
protocol_names = base_security_groups.V6_PROTOCOL_NAMES
|
|
protocol_ints = base_security_groups.V6_PROTOCOL_INTS
|
|
|
|
def _test_security_group_rule_protocol_legacy_icmpv6(self):
|
|
# These legacy protocols can be used to create security groups,
|
|
# but they could be shown either with their passed protocol name,
|
|
# or a canonical-ized version, depending on the neutron version.
|
|
# So we check against a list of possible values.
|
|
# TODO(haleyb): Remove once these legacy names are deprecated
|
|
protocols = {constants.PROTO_NAME_IPV6_ICMP_LEGACY:
|
|
constants.PROTO_NAME_IPV6_ICMP,
|
|
constants.PROTO_NAME_ICMP:
|
|
constants.PROTO_NAME_IPV6_ICMP}
|
|
for key, value in protocols.items():
|
|
self._test_security_group_rule_legacy(
|
|
protocol_list=[str(key), str(value)],
|
|
protocol=str(key),
|
|
direction=constants.INGRESS_DIRECTION,
|
|
ethertype=self.ethertype)
|
|
|
|
def _test_security_group_rule_legacy(self, protocol_list, **kwargs):
|
|
sg_kwargs = {}
|
|
if self.stateless_sg:
|
|
sg_kwargs['stateful'] = False
|
|
security_group = self.create_security_group(**sg_kwargs)
|
|
security_group_rule = self.create_security_group_rule(
|
|
security_group=security_group, **kwargs)
|
|
observed_security_group_rule = self.client.show_security_group_rule(
|
|
security_group_rule['id'])['security_group_rule']
|
|
for key, value in kwargs.items():
|
|
if key == 'protocol':
|
|
self.assertIn(security_group_rule[key], protocol_list,
|
|
"{!r} does not match.".format(key))
|
|
self.assertIn(observed_security_group_rule[key], protocol_list,
|
|
"{!r} does not match.".format(key))
|
|
else:
|
|
self.assertEqual(value, security_group_rule[key],
|
|
"{!r} does not match.".format(key))
|
|
self.assertEqual(value, observed_security_group_rule[key],
|
|
"{!r} does not match.".format(key))
|
|
|
|
|
|
class StatefulSecGroupProtocolIPv6Test(BaseSecGroupProtocolIPv6Test):
|
|
stateless_sg = False
|
|
|
|
@decorators.idempotent_id('c7d17b41-3b4e-4add-bb3b-6af59baaaffa')
|
|
def test_security_group_rule_protocol_legacy_icmpv6(self):
|
|
self._test_security_group_rule_protocol_legacy_icmpv6()
|
|
|
|
|
|
class StatelessSecGroupProtocolIPv6Test(BaseSecGroupProtocolIPv6Test):
|
|
required_extensions = ['security-group', 'stateful-security-group']
|
|
stateless_sg = True
|
|
|
|
@decorators.idempotent_id('a034814e-0fa5-4437-8e6f-0d2eebd668b3')
|
|
def test_security_group_rule_protocol_legacy_icmpv6(self):
|
|
self._test_security_group_rule_protocol_legacy_icmpv6()
|
|
|
|
|
|
class RbacSharedSecurityGroupTest(base.BaseAdminNetworkTest):
|
|
|
|
force_tenant_isolation = True
|
|
credentials = ['primary', 'alt', 'admin']
|
|
required_extensions = ['security-group', 'rbac-security-groups']
|
|
|
|
@classmethod
|
|
def resource_setup(cls):
|
|
super(RbacSharedSecurityGroupTest, cls).resource_setup()
|
|
cls.client2 = cls.os_alt.network_client
|
|
|
|
def _create_security_group(self):
|
|
return self.create_security_group(
|
|
name=data_utils.rand_name('test-sg'),
|
|
project={'id': self.admin_client.project_id})
|
|
|
|
def _make_admin_sg_shared_to_project_id(self, project_id):
|
|
sg = self._create_security_group()
|
|
rbac_policy = self.admin_client.create_rbac_policy(
|
|
object_type='security_group',
|
|
object_id=sg['id'],
|
|
action='access_as_shared',
|
|
target_tenant=project_id,
|
|
)['rbac_policy']
|
|
return {'security_group': sg, 'rbac_policy': rbac_policy}
|
|
|
|
@decorators.idempotent_id('2a41eb8f-2a35-11e9-bae9-acde48001122')
|
|
def test_policy_target_update(self):
|
|
res = self._make_admin_sg_shared_to_project_id(
|
|
self.client.project_id)
|
|
# change to client2
|
|
update_res = self.admin_client.update_rbac_policy(
|
|
res['rbac_policy']['id'], target_tenant=self.client2.project_id)
|
|
self.assertEqual(self.client2.project_id,
|
|
update_res['rbac_policy']['target_tenant'])
|
|
# make sure everything else stayed the same
|
|
res['rbac_policy'].pop('target_tenant')
|
|
update_res['rbac_policy'].pop('target_tenant')
|
|
self.assertEqual(res['rbac_policy'], update_res['rbac_policy'])
|
|
|
|
@decorators.idempotent_id('2a619a8a-2a35-11e9-90d9-acde48001122')
|
|
def test_port_presence_prevents_policy_rbac_policy_deletion(self):
|
|
res = self._make_admin_sg_shared_to_project_id(
|
|
self.client2.project_id)
|
|
sg_id = res['security_group']['id']
|
|
net = self.create_network(client=self.client2)
|
|
port = self.client2.create_port(
|
|
network_id=net['id'],
|
|
security_groups=[sg_id])['port']
|
|
|
|
# a port with shared sg should prevent the deletion of an
|
|
# rbac-policy required for it to be shared
|
|
with testtools.ExpectedException(exceptions.Conflict):
|
|
self.admin_client.delete_rbac_policy(res['rbac_policy']['id'])
|
|
|
|
# cleanup
|
|
self.client2.delete_port(port['id'])
|
|
self.admin_client.delete_rbac_policy(res['rbac_policy']['id'])
|
|
|
|
@decorators.idempotent_id('2a81795c-2a35-11e9-9d86-acde48001122')
|
|
def test_regular_client_shares_to_another_regular_client(self):
|
|
# owned by self.admin_client
|
|
sg = self._create_security_group()
|
|
with testtools.ExpectedException(exceptions.NotFound):
|
|
self.client.show_security_group(sg['id'])
|
|
rbac_policy = self.admin_client.create_rbac_policy(
|
|
object_type='security_group', object_id=sg['id'],
|
|
action='access_as_shared',
|
|
target_tenant=self.client.project_id)['rbac_policy']
|
|
self.client.show_security_group(sg['id'])
|
|
|
|
self.assertIn(rbac_policy,
|
|
self.admin_client.list_rbac_policies()['rbac_policies'])
|
|
# ensure that 'client2' can't see the rbac-policy sharing the
|
|
# sg to it because the rbac-policy belongs to 'client'
|
|
self.assertNotIn(rbac_policy['id'], [p['id'] for p in
|
|
self.client2.list_rbac_policies()['rbac_policies']])
|
|
|
|
@decorators.idempotent_id('2a9fd480-2a35-11e9-9cb6-acde48001122')
|
|
def test_filter_fields(self):
|
|
sg = self._create_security_group()
|
|
self.admin_client.create_rbac_policy(
|
|
object_type='security_group', object_id=sg['id'],
|
|
action='access_as_shared', target_tenant=self.client2.project_id)
|
|
field_args = (('id',), ('id', 'action'), ('object_type', 'object_id'),
|
|
('project_id', 'target_tenant'))
|
|
for fields in field_args:
|
|
res = self.admin_client.list_rbac_policies(fields=fields)
|
|
self.assertEqual(set(fields), set(res['rbac_policies'][0].keys()))
|
|
|
|
@decorators.idempotent_id('2abf8f9e-2a35-11e9-85f7-acde48001122')
|
|
def test_rbac_policy_show(self):
|
|
res = self._make_admin_sg_shared_to_project_id(
|
|
self.client.project_id)
|
|
p1 = res['rbac_policy']
|
|
p2 = self.admin_client.create_rbac_policy(
|
|
object_type='security_group',
|
|
object_id=res['security_group']['id'],
|
|
action='access_as_shared',
|
|
target_tenant='*')['rbac_policy']
|
|
|
|
self.assertEqual(
|
|
p1, self.admin_client.show_rbac_policy(p1['id'])['rbac_policy'])
|
|
self.assertEqual(
|
|
p2, self.admin_client.show_rbac_policy(p2['id'])['rbac_policy'])
|
|
|
|
@decorators.idempotent_id('2adf6bd7-2a35-11e9-9c62-acde48001122')
|
|
def test_filter_rbac_policies(self):
|
|
sg = self._create_security_group()
|
|
rbac_pol1 = self.admin_client.create_rbac_policy(
|
|
object_type='security_group', object_id=sg['id'],
|
|
action='access_as_shared',
|
|
target_tenant=self.client2.project_id)['rbac_policy']
|
|
rbac_pol2 = self.admin_client.create_rbac_policy(
|
|
object_type='security_group', object_id=sg['id'],
|
|
action='access_as_shared',
|
|
target_tenant=self.admin_client.project_id)['rbac_policy']
|
|
res1 = self.admin_client.list_rbac_policies(id=rbac_pol1['id'])[
|
|
'rbac_policies']
|
|
res2 = self.admin_client.list_rbac_policies(id=rbac_pol2['id'])[
|
|
'rbac_policies']
|
|
self.assertEqual(1, len(res1))
|
|
self.assertEqual(1, len(res2))
|
|
self.assertEqual(rbac_pol1['id'], res1[0]['id'])
|
|
self.assertEqual(rbac_pol2['id'], res2[0]['id'])
|
|
|
|
@decorators.idempotent_id('2aff3900-2a35-11e9-96b3-acde48001122')
|
|
def test_regular_client_blocked_from_sharing_anothers_policy(self):
|
|
sg = self._make_admin_sg_shared_to_project_id(
|
|
self.client.project_id)['security_group']
|
|
with testtools.ExpectedException(exceptions.BadRequest):
|
|
self.client.create_rbac_policy(
|
|
object_type='security_group', object_id=sg['id'],
|
|
action='access_as_shared',
|
|
target_tenant=self.client2.project_id)
|
|
|
|
# make sure the rbac-policy is invisible to the tenant for which it's
|
|
# being shared
|
|
self.assertFalse(self.client.list_rbac_policies()['rbac_policies'])
|
|
|
|
|
|
class SecGroupSearchCriteriaTest(base.BaseSearchCriteriaTest):
|
|
|
|
required_extensions = ['security-group']
|
|
resource = 'security-group'
|
|
|
|
@classmethod
|
|
def resource_setup(cls):
|
|
super(SecGroupSearchCriteriaTest, cls).resource_setup()
|
|
cls.security_group = cls.create_security_group()
|
|
for name in cls.resource_names:
|
|
cls.create_security_group(name=name)
|
|
|
|
@decorators.idempotent_id('0064aa80-8a29-442d-a8de-9101af8210fa')
|
|
def test_list_sorts_by_name_asc(self):
|
|
self._test_list_sorts_asc()
|
|
|
|
@decorators.idempotent_id('35e86832-53cd-4e63-97ec-31a2413da591')
|
|
def test_list_sorts_by_name_desc(self):
|
|
self._test_list_sorts_desc()
|
|
|
|
@decorators.idempotent_id('b9654cdc-80bc-43f8-844e-dfe88fd2f125')
|
|
def test_list_pagination(self):
|
|
self._test_list_pagination()
|
|
|
|
@decorators.idempotent_id('5c78bd57-e6e9-4e71-a05c-9c4082a3f139')
|
|
def test_list_no_pagination_limit_0(self):
|
|
self._test_list_no_pagination_limit_0()
|
|
|
|
|
|
class SecGroupNormalizedCidrTest(base.BaseNetworkTest):
|
|
|
|
required_extensions = ['security-group', 'security-groups-normalized-cidr']
|
|
|
|
@decorators.idempotent_id('f87bb108-205c-4f06-a378-81a26f71b829')
|
|
def test_normalized_cidr_in_rule(self):
|
|
security_group = self.create_security_group()
|
|
rule = self.create_security_group_rule(
|
|
security_group=security_group,
|
|
direction=constants.INGRESS_DIRECTION,
|
|
remote_ip_prefix='10.0.0.34/24')
|
|
self.assertEqual('10.0.0.0/24', rule['normalized_cidr'])
|
|
self.assertEqual('10.0.0.34/24', rule['remote_ip_prefix'])
|
|
|
|
rule = self.create_security_group_rule(
|
|
security_group=security_group,
|
|
remote_group_id=security_group['id'],
|
|
direction=constants.INGRESS_DIRECTION)
|
|
self.assertIsNone(rule['normalized_cidr'])
|
|
self.assertIsNone(rule['remote_ip_prefix'])
|