openstack/designate-tempest-plugin
Code Issues Proposed changes

Extending "Shared Zones" test coverage

Browse Source 
This patch adds:
1) API and Scenario test cases for Admin tenant
2) Quota tests, to ensure: "quotas enforced against owner"
3) Negative scenario tests

Change-Id: I0bbc6504ba2ef75bba8268d5e91b2588a4215555
This commit is contained in:
Arkady Shtempler 2023-03-01 15:37:45 +02:00
parent 759ee725b8
commit fdbe35c32f
6 changed files with 785 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
designate_tempest_plugin/services/dns/v2/json/shared_zones_client.py
View File

@ -18,17 +18,19 @@ from designate_tempest_plugin.services.dns.v2.json import base
class SharedZonesClient(base.DnsClientV2Base):
@base.handle_errors
def create_zone_share(self, zone_id, target_project_id):
def create_zone_share(self, zone_id, target_project_id, headers=None):
"""Create a new zone share for a project ID.
:param zone_id: Zone UUID to share
:param target_project_id: Project ID that will gain access to specified
zone
:param headers: (dict): The headers to use for the request.
:return: Zone share dict
"""
resp, body = self._create_request(
'zones/{}/shares'.format(zone_id),
data={'target_project_id': target_project_id})
data={'target_project_id': target_project_id}, headers=headers,
extra_headers=True)
# Endpoint should Return a HTTP 201
self.expected_success(201, resp.status)
@ -36,15 +38,16 @@ class SharedZonesClient(base.DnsClientV2Base):
return resp, body
@base.handle_errors
def show_zone_share(self, zone_id, zone_share_id):
def show_zone_share(self, zone_id, zone_share_id, headers=None):
"""Get the zone share object
:param zone_id: Zone UUID for the share
:param zone_share_id: The zone share ID
:param headers: (dict): The headers to use for the request.
:return: Zone share dict
"""
return self._show_request('zones/{}/shares'.format(zone_id),
zone_share_id)
zone_share_id, headers=headers)
@base.handle_errors
def list_zone_shares(self, zone_id, params=None, headers=None):
@ -53,21 +56,23 @@ class SharedZonesClient(base.DnsClientV2Base):
:param zone_id: Zone UUID to query for the shares
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param headers: (dict): The headers to use for the request.
:return: Zone shares list.
"""
return self._list_request('zones/{}/shares'.format(zone_id),
params=params, headers=headers)
@base.handle_errors
def delete_zone_share(self, zone_id, zone_share_id):
def delete_zone_share(self, zone_id, zone_share_id, headers=None):
"""Deletes the zone share
:param zone_id: Zone UUID for the share
:param zone_share_id: The zone share ID
:param headers: (dict): The headers to use for the request.
:return: None
"""
resp, body = self._delete_request('zones/{}/shares'.format(zone_id),
zone_share_id)
zone_share_id, headers=headers)
# Endpoint should Return a HTTP 204 - No Content
self.expected_success(204, resp.status)

20
designate_tempest_plugin/services/dns/v2/json/zones_client.py
View File

@ -48,7 +48,7 @@ class ZonesClient(base.DnsClientV2Base):
Default: PRIMARY
:param primaries: List of Primary nameservers. Required for SECONDARY
Default: None
:param params: A Python dict that represents the query paramaters to
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param project_id: When specified, overrides the project ID the zone
will be associated with.
@ -102,7 +102,7 @@ class ZonesClient(base.DnsClientV2Base):
def show_zone(self, uuid, params=None, headers=None):
"""Gets a specific zone.
:param uuid: Unique identifier of the zone in UUID format.
:param params: A Python dict that represents the query paramaters to
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param headers (dict): The headers to use for the request.
:return: Serialized zone as a dictionary.
@ -114,7 +114,7 @@ class ZonesClient(base.DnsClientV2Base):
def show_zone_nameservers(self, zone_uuid, params=None, headers=None):
"""Gets list of Zone Name Servers
:param zone_uuid: Unique identifier of the zone in UUID format.
:param params: A Python dict that represents the query paramaters to
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param headers (dict): The headers to use for the request.
:return: Serialized nameservers as a list.
@ -126,7 +126,7 @@ class ZonesClient(base.DnsClientV2Base):
@base.handle_errors
def list_zones(self, params=None, headers=None):
"""Gets a list of zones.
:param params: A Python dict that represents the query paramaters to
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param headers (dict): The headers to use for the request.
:return: Serialized zones as a list.
@ -134,14 +134,20 @@ class ZonesClient(base.DnsClientV2Base):
return self._list_request('zones', params=params, headers=headers)
@base.handle_errors
def delete_zone(self, uuid, params=None, headers=None):
def delete_zone(self, uuid, params=None, headers=None, delete_shares=None):
"""Deletes a zone having the specified UUID.
:param uuid: The unique identifier of the zone.
:param params: A Python dict that represents the query paramaters to
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param headers (dict): The headers to use for the request.
:param delete_shares: if set, delete-shares modifier will be activated
:return: A tuple with the server response and the response body.
"""
if delete_shares:
if headers:
headers['x-designate-delete-shares'] = True
else:
headers = {'x-designate-delete-shares': True}
resp, body = self._delete_request(
'zones', uuid, params=params, headers=headers)
@ -163,7 +169,7 @@ class ZonesClient(base.DnsClientV2Base):
:param description: A description of the zone.
Default: Random Value
:param wait_until: Block until the zone reaches the desiered status
:param params: A Python dict that represents the query paramaters to
:param params: A Python dict that represents the query parameters to
include in the request URI.
:param headers (dict): The headers to use for the request.
:return: A tuple with the server response and the updated zone.

283
designate_tempest_plugin/tests/api/v2/test_shared_zones.py
View File

@ -16,6 +16,7 @@ from oslo_log import log as logging
from oslo_utils import uuidutils
from oslo_utils import versionutils
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
@ -58,7 +59,7 @@ class BaseSharedZoneTest(base.BaseDnsV2Test):
@classmethod
def resource_cleanup(cls):
cls.zones_client.delete_zone(
cls.zone['id'], ignore_errors=lib_exc.NotFound)
cls.zone['id'], ignore_errors=lib_exc.NotFound, delete_shares=True)
cls.admin_tld_client.delete_tld(cls.class_tld[1]['id'])
super(BaseSharedZoneTest, cls).resource_cleanup()
@ -68,8 +69,10 @@ class BaseSharedZoneTest(base.BaseDnsV2Test):
if CONF.enforce_scope.designate:
cls.admin_tld_client = cls.os_system_admin.dns_v2.TldClient()
cls.adm_shr_client = cls.os_system_admin.dns_v2.SharedZonesClient()
else:
cls.admin_tld_client = cls.os_admin.dns_v2.TldClient()
cls.adm_shr_client = cls.os_admin.dns_v2.SharedZonesClient()
cls.alt_zone_client = cls.os_alt.dns_v2.ZonesClient()
cls.demo_zone_client = cls.os_demo.dns_v2.ZonesClient()
cls.share_zone_client = cls.os_primary.dns_v2.SharedZonesClient()
@ -280,3 +283,281 @@ class NegativeSharedZonesTest(BaseSharedZoneTest):
lib_exc.Forbidden,
self.alt_zone_client.create_zone,
name=sub_zone_name)
@decorators.idempotent_id('957ba3f8-c250-11ed-a8b1-201e8823901f')
def test_share_zone_with_yourself_is_not_allowed(self):
with self.assertRaisesDns(lib_exc.BadRequest, 'bad_request', 400):
self.share_zone_client.create_zone_share(
zone_id=self.zone['id'],
target_project_id=self.share_zone_client.project_id)
class AdminSharedZonesTest(BaseSharedZoneTest):
@classmethod
def setup_credentials(cls):
# Do not create network resources for these test.
cls.set_network_resources()
super(AdminSharedZonesTest, cls).setup_credentials()
@decorators.idempotent_id('2bb7bcb2-b824-11ed-9e56-201e8823901f')
def test_create_zone_share_all_projects_header(self):
LOG.info(
'Admin user creates shared zone for Alt tenant'
' using "x-auth-all-projects" header')
all_projects_header = self.all_projects_header
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=all_projects_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=all_projects_header)
self.assertTrue(uuidutils.is_uuid_like(shared_zone['id']))
self.assertEqual(self.zone['id'], shared_zone['zone_id'])
self.assertEqual(self.adm_shr_client.project_id,
shared_zone['project_id'])
self.assertEqual(self.alt_zone_client.project_id,
shared_zone['target_project_id'])
self.assertIsNotNone(shared_zone['created_at'])
self.assertIsNone(shared_zone['updated_at'])
self.assertIsNotNone(shared_zone['links'])
@decorators.idempotent_id('f26cd3ac-b8fa-11ed-b4ca-201e8823901f')
def test_create_zone_share_sudo_project_header(self):
LOG.info(
'Admin user creates shared zone for Alt tenant '
'using "x-auth-sudo-project-id" header')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=sudo_header)
self.assertTrue(uuidutils.is_uuid_like(shared_zone['id']))
self.assertEqual(self.zone['id'], shared_zone['zone_id'])
self.assertEqual(self.share_zone_client.project_id,
shared_zone['project_id'])
self.assertEqual(self.alt_zone_client.project_id,
shared_zone['target_project_id'])
self.assertIsNotNone(shared_zone['created_at'])
self.assertIsNone(shared_zone['updated_at'])
self.assertIsNotNone(shared_zone['links'])
@decorators.idempotent_id('ce2688e8-b90a-11ed-b4ca-201e8823901f')
def test_show_shared_zone_all_projects_header(self):
LOG.info(
'Admin user creates shared zone for Alt tenant'
' using "x-auth-all-projects" header')
all_projects_header = self.all_projects_header
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=all_projects_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=all_projects_header)
LOG.info('Admin user shows shared zone and validates its content')
body = self.adm_shr_client.show_zone_share(
self.zone['id'], shared_zone['id'], headers=all_projects_header)[1]
self.assertExpected(shared_zone, body, self.excluded_keys)
@decorators.idempotent_id('46f7db22-b90c-11ed-b4ca-201e8823901f')
def test_delete_zone_share_sudo_project_header(self):
LOG.info(
'Admin user creates shared zone for Alt tenant'
' using "x-auth-sudo-project-id" header')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=sudo_header,
ignore_errors=lib_exc.NotFound)
LOG.info('As Admin delete zone share and ensure it was deleted')
self.adm_shr_client.delete_zone_share(
self.zone['id'], shared_zone['id'], headers=sudo_header)
self.assertRaises(lib_exc.NotFound,
self.adm_shr_client.show_zone_share,
self.zone['id'], shared_zone['id'], headers=sudo_header)
@decorators.idempotent_id('2eedfd60-b90f-11ed-b4ca-201e8823901f')
def test_list_zone_shares_all_projects_header(self):
LOG.info(
"Admin user shares Primary's zone with Alt tenant"
" using 'x-auth-all-projects' header")
all_projects_header = self.all_projects_header
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=all_projects_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=all_projects_header)
LOG.info(
"Admin user shares Primary's zone with Demo tenant"
" using 'x-auth-all-projects' header")
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.demo_zone_client.project_id,
headers=all_projects_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=all_projects_header)
LOG.info('Admin user lists zone shares')
body = self.adm_shr_client.list_zone_shares(
self.zone['id'], headers=all_projects_header)[1]
self.assertEqual(2, len(body['shared_zones']))
targets = []
for share in body['shared_zones']:
targets.append(share['target_project_id'])
self.assertIn(self.alt_zone_client.project_id, targets)
self.assertIn(self.demo_zone_client.project_id, targets)
class AdminSharedZonesTestNegative(BaseSharedZoneTest):
@classmethod
def setup_credentials(cls):
# Do not create network resources for these test.
cls.set_network_resources()
super(AdminSharedZonesTestNegative, cls).setup_credentials()
@decorators.idempotent_id('595ae1fc-bce4-11ed-baf2-201e8823901f')
def test_create_zone_share_invalid_project_id(self):
LOG.info(
'Admin user tries to create shared zone, using non existing '
'project ID id in "x-auth-sudo-project-id" header')
sudo_header = {
'x-auth-sudo-project-id': data_utils.rand_uuid()}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.create_zone_share,
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)
@decorators.idempotent_id('aa42d82e-bcf6-11ed-baf2-201e8823901f')
def test_create_zone_share_invalid_zone_id(self):
LOG.info(
'Admin user tries to create shared zone, using non existing '
'zone ID and "x-auth-sudo-project-id" header')
sudo_header = {
'x-auth-sudo-project-id': self.alt_zone_client.project_id}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.create_zone_share,
data_utils.rand_uuid(), self.alt_zone_client.project_id,
headers=sudo_header)
@decorators.idempotent_id('9e7202ba-bd94-11ed-80f5-201e8823901f')
def test_show_shared_zone_invalid_shared_zone_id(self):
LOG.info('Admin tries to show shared zone using not '
'existing shared zone ID')
sudo_header = {
'x-auth-sudo-project-id': self.alt_zone_client.project_id}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.show_zone_share,
self.zone['id'], data_utils.rand_uuid(), headers=sudo_header)
@decorators.idempotent_id('8852329c-bd95-11ed-80f5-201e8823901f')
def test_show_shared_zone_invalid_project_id(self):
LOG.info('Admin tries to show shared zone '
'using not existing project ID')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=sudo_header)
sudo_header_invalid_project_id = {
'x-auth-sudo-project-id': data_utils.rand_uuid()}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.show_zone_share,
self.zone['id'], shared_zone['id'],
headers=sudo_header_invalid_project_id)
@decorators.idempotent_id('871e7e1c-bd9a-11ed-80f5-201e8823901f')
@decorators.skip_because(bug="2009819")
def test_list_zone_shares_invalid_zone_id(self):
LOG.info('Admin user tries to list shared zone '
'using not existing zone ID')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.list_zone_shares,
data_utils.rand_uuid(), headers=sudo_header)
@decorators.idempotent_id('e71068c8-bdb1-11ed-80f5-201e8823901f')
@ decorators.skip_because(bug="2009819")
def test_list_zone_shares_invalid_project_id(self):
LOG.info('Admin user tries to list shared zone using '
'not existing project ID')
sudo_header = {
'x-auth-sudo-project-id': data_utils.rand_uuid()}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.list_zone_shares,
self.zone['id'], headers=sudo_header)
@decorators.idempotent_id('7136b430-bdb2-11ed-80f5-201e8823901f')
def test_delete_zone_share_invalid_project_id(self):
LOG.info('Admin user creates shared zone for Alt user')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=sudo_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Admin user tries to delete the shared zone '
'using non existing project ID')
invalid_sudo_header = {
'x-auth-sudo-project-id': data_utils.rand_uuid()}
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.delete_zone_share,
self.zone['id'], shared_zone['id'],
headers=invalid_sudo_header)
@decorators.idempotent_id('d44c65e2-bdc1-11ed-80f5-201e8823901f')
def test_delete_zone_share_invalid_shared_zone_id(self):
LOG.info('Admin user creates shared zone for Alt user')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=sudo_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Admin user tries to delete the shared zone '
'using non existing shared zone ID')
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.delete_zone_share,
self.zone['id'], data_utils.rand_uuid(),
headers=sudo_header)
@decorators.idempotent_id('06de2342-bdc2-11ed-80f5-201e8823901f')
def test_delete_zone_share_invalid_zone_id(self):
LOG.info('Admin user creates shared zone for Alt user')
sudo_header = {
'x-auth-sudo-project-id': self.share_zone_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
self.zone['id'], self.alt_zone_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, self.zone['id'],
shared_zone['id'], headers=sudo_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Admin user tries to delete the shared zone '
'using non zone ID')
self.assertRaises(
lib_exc.NotFound, self.adm_shr_client.delete_zone_share,
data_utils.rand_uuid(), shared_zone['id'],
headers=sudo_header)

10
designate_tempest_plugin/tests/api/v2/test_zones.py
View File

@ -11,7 +11,6 @@
# 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 uuid
from oslo_log import log as logging
from oslo_utils import versionutils
from tempest import config
@ -254,7 +253,7 @@ class ZonesTest(BaseZonesTest):
LOG.info('Delete the zone using delete-shares')
body = self.zones_client.delete_zone(
zone['id'], headers={'x-designate-delete-shares': True})[1]
zone['id'], delete_shares=True)[1]
LOG.info('Ensure we respond with DELETE+PENDING')
self.assertEqual(const.DELETE, body['action'])
@ -686,7 +685,7 @@ class ZonesNegativeTest(BaseZonesTest):
def test_show_not_existing_zone(self):
LOG.info('Fetch non existing zone')
self.assertRaises(lib_exc.NotFound,
lambda: self.zones_client.show_zone(uuid.uuid1()))
lambda: self.zones_client.show_zone(data_utils.rand_uuid()))
@decorators.idempotent_id('736e3b50-92e0-11eb-9d02-74e5f9e2a801')
def test_use_invalid_id_to_show_zone(self):
@ -698,14 +697,15 @@ class ZonesNegativeTest(BaseZonesTest):
def test_delete_non_existing_zone(self):
LOG.info('Delete non existing zone')
self.assertRaises(lib_exc.NotFound,
lambda: self.zones_client.delete_zone(uuid.uuid1()))
lambda: self.zones_client.delete_zone(data_utils.rand_uuid()))
@decorators.idempotent_id('e391e30a-92e0-11eb-9d02-74e5f9e2a801')
def test_update_non_existing_zone(self):
LOG.info('Update non existing zone')
self.assertRaises(lib_exc.NotFound,
lambda: self.zones_client.update_zone(
uuid.uuid1(), description=data_utils.rand_name()))
data_utils.rand_uuid(),
description=data_utils.rand_name()))
@decorators.idempotent_id('925192f2-0ed8-4591-8fe7-a9fa028f90a0')
def test_list_zones_dot_json_fails(self):

184
designate_tempest_plugin/tests/scenario/v2/test_quotas.py
View File

@ -13,6 +13,7 @@
# under the License.
import random
from oslo_log import log as logging
from oslo_utils import versionutils
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
@ -375,3 +376,186 @@ class QuotasBoundary(base.BaseDnsV2Test, tempest.test.BaseTestCase):
self.admin_zones_client, zone['id'],
headers=sudo_header,
ignore_errors=lib_exc.NotFound)
class SharedZonesQuotaTest(base.BaseDnsV2Test):
credentials = ['primary', 'admin', 'system_admin']
@classmethod
def setup_clients(cls):
super(SharedZonesQuotaTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_tld_client = cls.os_system_admin.dns_v2.TldClient()
cls.adm_project_client = cls.os_system_admin.projects_client
cls.adm_quota_client = cls.os_system_admin.dns_v2.QuotasClient()
cls.adm_zone_client = cls.os_system_admin.dns_v2.ZonesClient()
cls.adm_shr_client = cls.os_system_admin.dns_v2.SharedZonesClient()
else:
cls.admin_tld_client = cls.os_admin.dns_v2.TldClient()
cls.adm_project_client = cls.os_admin.projects_client
cls.adm_quota_client = cls.os_admin.dns_v2.QuotasClient()
cls.adm_zone_client = cls.os_admin.dns_v2.ZonesClient()
cls.adm_shr_client = cls.os_admin.dns_v2.SharedZonesClient()
cls.share_zone_client = cls.os_primary.dns_v2.SharedZonesClient()
cls.rec_client = cls.os_primary.dns_v2.RecordsetClient()
cls.export_zone_client = cls.os_primary.dns_v2.ZoneExportsClient()
@classmethod
def resource_setup(cls):
super(SharedZonesQuotaTest, cls).resource_setup()
if not versionutils.is_compatible('2.1', cls.api_version,
same_major=False):
raise cls.skipException(
'The shared zones scenario tests require Designate API '
'version 2.1 or newer. Skipping Shared Zones scenario tests.')
# Make sure we have an allowed TLD available
tld_name = dns_data_utils.rand_zone_name(name='SharedZonesTest')
cls.tld_name = f'.{tld_name}'
cls.class_tld = cls.admin_tld_client.create_tld(tld_name=tld_name[:-1])
@classmethod
def resource_cleanup(cls):
cls.admin_tld_client.delete_tld(cls.class_tld[1]['id'])
super(SharedZonesQuotaTest, cls).resource_cleanup()
def _create_shared_zone_for_project(
self, zone_name, project_id, sudo_header):
"""Admin creates Zone for project ID and shares it with Primary"""
zone_name = dns_data_utils.rand_zone_name(
name=zone_name,
suffix=self.tld_name)
zone = self.adm_zone_client.create_zone(
name=zone_name, project_id=project_id, wait_until=const.ACTIVE)[1]
self.addCleanup(
self.wait_zone_delete, self.adm_zone_client, zone['id'],
headers=sudo_header, delete_shares=True)
shared_zone = self.adm_shr_client.create_zone_share(
zone['id'], self.rec_client.project_id,
headers=sudo_header)[1]
self.addCleanup(self.adm_shr_client.delete_zone_share,
zone['id'], shared_zone['id'], headers=sudo_header)
return zone, shared_zone
@decorators.attr(type='slow')
@decorators.idempotent_id('31968688-c0d3-11ed-b04f-201e8823901f')
@decorators.skip_because(bug="1992445")
def test_zone_recordsets_enforced_against_owner(self):
# Create a dedicated Project "A" for shared zone test
tenant_id = self.adm_project_client.create_project(
name=data_utils.rand_name(
name='SharedZonesQuotaTest'))['project']['id']
self.addCleanup(self.adm_project_client.delete_project, tenant_id)
# Set Quotas "zone_recordsets:1" for a project "A"
sudo_header = {'x-auth-sudo-project-id': tenant_id}
quotas = {
'zones': 7, 'zone_recordsets': 1, 'zone_records': 7,
'recordset_records': 7, 'api_export_size': 7}
self.adm_quota_client.set_quotas(
project_id=tenant_id, quotas=quotas,
headers=sudo_header)
# Admin creates a zone for project "A" and shares it Primary
zone = self._create_shared_zone_for_project(
zone_name='test_zone_recordsets_enforced_against_owner',
project_id=tenant_id, sudo_header=sudo_header)[0]
# Primary creates a first recodset - should PASS
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'])
recordset = self.rec_client.create_recordset(
zone['id'], recordset_data)[1]
self.addCleanup(self.wait_recordset_delete, self.rec_client,
zone['id'], recordset['id'], ignore_errors=lib_exc.NotFound)
# Primary creates a second recodset - should FAIL
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'])
with self.assertRaisesDns(
lib_exc.OverLimit, 'over_quota', 413):
self.rec_client.create_recordset(zone['id'], recordset_data)
@decorators.attr(type='slow')
@decorators.idempotent_id('cbb8d6b4-c64e-11ed-80d8-201e8823901f')
def test_zone_recocrds_enforced_against_owner(self):
# Create a dedicated Project "A" for shared zone test
tenant_id = self.adm_project_client.create_project(
name=data_utils.rand_name(
name='SharedZonesQuotaTest'))['project']['id']
self.addCleanup(self.adm_project_client.delete_project, tenant_id)
# Set Quotas "zone_records:1" for a project "A"
sudo_header = {'x-auth-sudo-project-id': tenant_id}
quotas = {
'zones': 7, 'zone_recordsets': 7, 'zone_records': 1,
'recordset_records': 7, 'api_export_size': 7}
self.adm_quota_client.set_quotas(
project_id=tenant_id, quotas=quotas,
headers=sudo_header)
# Admin creates a zone for project "A" and share it with Primary
zone = self._create_shared_zone_for_project(
zone_name='test_zone_recocrds_enforced_against_owner',
project_id=tenant_id, sudo_header=sudo_header)[0]
# Primary creates recordset with (single record) --> Should PASS
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'])
body = self.rec_client.create_recordset(
zone['id'], recordset_data)[1]
self.addCleanup(
self.wait_recordset_delete, self.rec_client,
zone['id'], body['id'])
# Primary creates one more recordset (single record) --> Should FAIL
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'])
with self.assertRaisesDns(
lib_exc.OverLimit, 'over_quota', 413):
self.rec_client.create_recordset(
zone['id'], recordset_data)
@decorators.attr(type='slow')
@decorators.idempotent_id('40b436f0-c895-11ed-8900-201e8823901f')
def test_recordset_records_enforced_against_owner(self):
# Create a dedicated Project "A" for shared zone test
tenant_id = self.adm_project_client.create_project(
name=data_utils.rand_name(
name='SharedZonesQuotaTest'))['project']['id']
self.addCleanup(self.adm_project_client.delete_project, tenant_id)
# Set Quotas "zone_records:1" for a project "A"
sudo_header = {'x-auth-sudo-project-id': tenant_id}
quotas = {
'zones': 7, 'zone_recordsets': 7, 'zone_records': 7,
'recordset_records': 1, 'api_export_size': 7}
self.adm_quota_client.set_quotas(
project_id=tenant_id, quotas=quotas,
headers=sudo_header)
# Admin creates a zone for project "A" and share it with Primary
zone = self._create_shared_zone_for_project(
zone_name='test_recordset_records_enforced_against_owner',
project_id=tenant_id, sudo_header=sudo_header)[0]
# Primary creates recordset with (single record) --> Should PASS
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'])
body = self.rec_client.create_recordset(
zone['id'], recordset_data)[1]
self.addCleanup(
self.wait_recordset_delete, self.rec_client,
zone['id'], body['id'])
# Primary creates one more recordset (two records) --> Should FAIL
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'], number_of_records=2)
with self.assertRaisesDns(
lib_exc.OverLimit, 'over_quota', 413):
self.rec_client.create_recordset(
zone['id'], recordset_data)

290
designate_tempest_plugin/tests/scenario/v2/test_shared_zones.py
View File

@ -19,6 +19,8 @@ from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from designate_tempest_plugin import data_utils as dns_data_utils
from designate_tempest_plugin.common import constants as const
from designate_tempest_plugin.common import waiters
from designate_tempest_plugin.tests import base
CONF = config.CONF
@ -34,12 +36,15 @@ class SharedZonesTest(base.BaseDnsV2Test):
super(SharedZonesTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_tld_client = cls.os_system_admin.dns_v2.TldClient()
cls.adm_shr_client = cls.os_system_admin.dns_v2.SharedZonesClient()
else:
cls.admin_tld_client = cls.os_admin.dns_v2.TldClient()
cls.adm_shr_client = cls.os_admin.dns_v2.SharedZonesClient()
cls.share_zone_client = cls.os_primary.dns_v2.SharedZonesClient()
cls.rec_client = cls.os_primary.dns_v2.RecordsetClient()
cls.alt_rec_client = cls.os_alt.dns_v2.RecordsetClient()
cls.demo_rec_client = cls.os_demo.dns_v2.RecordsetClient()
cls.primary_import_client = cls.os_primary.dns_v2.ZoneImportsClient()
@classmethod
def resource_setup(cls):
@ -196,3 +201,288 @@ class SharedZonesTest(base.BaseDnsV2Test):
self.assertRaises(lib_exc.NotFound,
self.alt_rec_client.show_recordset,
zone['id'], recordset['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('fb170be8-c0bc-11ed-99a3-201e8823901f')
def test_admin_zone_share_CRUD_recordset(self):
# Create a zone to share with the alt credential
zone_name = dns_data_utils.rand_zone_name(name='TestZone',
suffix=self.tld_name)
LOG.info('Create a zone: %s', zone_name)
zone = self.zones_client.create_zone(name=zone_name)[1]
self.addCleanup(self.wait_zone_delete, self.zones_client, zone['id'],
ignore_errors=lib_exc.NotFound)
# Generate recordset data to be used latter in the test
recordset_data = dns_data_utils.rand_recordset_data(
record_type='A', zone_name=zone['name'])
# Check that the alt user has no access to the zone before the share
self.assertRaises(lib_exc.NotFound,
self.alt_rec_client.create_recordset,
zone['id'], recordset_data)
# Admin creates shared zone for Alt using "x-auth-sudo-project-id"
sudo_header = {
'x-auth-sudo-project-id': self.zones_client.project_id}
shared_zone = self.adm_shr_client.create_zone_share(
zone['id'], self.alt_rec_client.project_id,
headers=sudo_header)[1]
self.addCleanup(
self.adm_shr_client.delete_zone_share, zone['id'],
shared_zone['id'], headers=sudo_header)
# Check that the alt user can create a recordset on the shared zone
recordset = self.alt_rec_client.create_recordset(zone['id'],
recordset_data)[1]
self.addCleanup(self.wait_recordset_delete, self.alt_rec_client,
zone['id'], recordset['id'], ignore_errors=lib_exc.NotFound)
# Check that the alt user can see the alt recordset
show_recordset = self.alt_rec_client.show_recordset(
zone['id'], recordset['id'])[1]
self.assertEqual(recordset['id'], show_recordset['id'])
# Check that the zone owner can see the alt recordset
show_recordset = self.rec_client.show_recordset(
zone['id'], recordset['id'])[1]
self.assertEqual(recordset['id'], show_recordset['id'])
recordset_data = {
'ttl': dns_data_utils.rand_ttl(start=recordset['ttl'] + 1)
}
# Check that the alt user can update a recordset on the shared zone
update = self.alt_rec_client.update_recordset(zone['id'],
recordset['id'], recordset_data)[1]
self.assertNotEqual(recordset['ttl'], update['ttl'])
recordset_data = {
'ttl': dns_data_utils.rand_ttl(start=update['ttl'] + 1)
}
# Check that the zone owner can update a recordset on the shared zone
primary_update = self.rec_client.update_recordset(zone['id'],
recordset['id'], recordset_data)[1]
self.assertNotEqual(update['ttl'], primary_update['ttl'])
# Check that the alt user can delete it's recordset
self.alt_rec_client.delete_recordset(zone['id'], recordset['id'])
LOG.info('Ensure successful deletion of Recordset')
self.assertRaises(lib_exc.NotFound,
self.alt_rec_client.show_recordset,
zone['id'], recordset['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('b7dd37b8-c3ea-11ed-a102-201e8823901f')
def test_share_imported_zone(self):
# Primary user imports zone from a zone file
zone_name = dns_data_utils.rand_zone_name(
name="test_share_imported_zone", suffix=self.tld_name)
zone_data = dns_data_utils.rand_zonefile_data(name=zone_name)
zone_import = self.primary_import_client.create_zone_import(
zonefile_data=zone_data)[1]
self.addCleanup(
self.primary_import_client.delete_zone_import, zone_import['id'])
waiters.wait_for_zone_import_status(
self.primary_import_client, zone_import['id'], const.COMPLETE)
# Primary shares previously created zone with Alt user
zone_id = self.primary_import_client.show_zone_import(
zone_import['id'])[1]['zone_id']
shared_zone = self.share_zone_client.create_zone_share(
zone_id, self.alt_rec_client.project_id)[1]
self.addCleanup(self.share_zone_client.delete_zone_share,
zone_id, shared_zone['id'])
class SharedZonesTestNegative(base.BaseDnsV2Test):
credentials = ['primary', 'admin', 'system_admin', 'alt',
['demo', 'member']]
@classmethod
def setup_clients(cls):
super(SharedZonesTestNegative, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_tld_client = cls.os_system_admin.dns_v2.TldClient()
cls.adm_shr_client = cls.os_system_admin.dns_v2.SharedZonesClient()
else:
cls.admin_tld_client = cls.os_admin.dns_v2.TldClient()
cls.adm_shr_client = cls.os_admin.dns_v2.SharedZonesClient()
cls.share_zone_client = cls.os_primary.dns_v2.SharedZonesClient()
cls.alt_export_client = cls.os_alt.dns_v2.ZoneExportsClient()
cls.primary_export_client = cls.os_primary.dns_v2.ZoneExportsClient()
cls.alt_zone_client = cls.os_alt.dns_v2.ZonesClient()
cls.primary_import_client = cls.os_primary.dns_v2.ZoneImportsClient()
cls.alt_import_client = cls.os_alt.dns_v2.ZoneImportsClient()
cls.prm_transfer_client = cls.os_primary.dns_v2.TransferRequestClient()
cls.alt_transfer_client = cls.os_alt.dns_v2.TransferRequestClient()
@classmethod
def resource_setup(cls):
super(SharedZonesTestNegative, cls).resource_setup()
if not versionutils.is_compatible('2.1', cls.api_version,
same_major=False):
raise cls.skipException(
'The shared zones scenario tests require Designate API '
'version 2.1 or newer. Skipping Shared Zones scenario tests.')
# Make sure we have an allowed TLD available
tld_name = dns_data_utils.rand_zone_name(name='SharedZonesTest')
cls.tld_name = f'.{tld_name}'
cls.class_tld = cls.admin_tld_client.create_tld(tld_name=tld_name[:-1])
@classmethod
def resource_cleanup(cls):
cls.admin_tld_client.delete_tld(cls.class_tld[1]['id'])
super(SharedZonesTestNegative, cls).resource_cleanup()
def _create_shared_zone(self, zone_name):
# Primary tenant creates zone and shares it with Alt tenant
zone_name = dns_data_utils.rand_zone_name(
name=zone_name, suffix=self.tld_name)
LOG.info('Create a zone: %s', zone_name)
zone = self.zones_client.create_zone(name=zone_name)[1]
self.addCleanup(self.wait_zone_delete, self.zones_client, zone['id'],
ignore_errors=lib_exc.NotFound)
shared_zone = self.share_zone_client.create_zone_share(
zone['id'], self.alt_export_client.project_id)[1]
self.addCleanup(self.share_zone_client.delete_zone_share,
zone['id'], shared_zone['id'])
return zone, shared_zone
@decorators.attr(type='slow')
@decorators.idempotent_id('1d2c91c2-c328-11ed-a033-201e8823901f')
def test_alt_create_export_for_shared_zone(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_create_export_for_shared_zone')[0]
self.assertRaises(
lib_exc.NotFound,
self.alt_export_client.create_zone_export, zone['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('1e74410c-c32c-11ed-a033-201e8823901f')
def test_alt_list_shared_zone_exports(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_list_shared_zone_exports')[0]
# Primary creates zone export
zone_export = self.primary_export_client.create_zone_export(
zone['id'])[1]
self.addCleanup(
self.primary_export_client.delete_zone_export, zone_export['id'])
waiters.wait_for_zone_export_status(
self.primary_export_client, zone_export['id'], const.COMPLETE)
# Primary lists zone exports
prim_zone_exports = self.primary_export_client.list_zone_exports()[1]
self.assertEqual(1, len(prim_zone_exports['exports']),
'Failed, no zone exports listed for a primary tenant')
# Alt tries to list Primary's zone exports
alt_zone_exports = self.alt_export_client.list_zone_exports()[1]
self.assertEqual(
0, len(alt_zone_exports['exports']),
'Failed, Alt tenant is expected to receive an '
'empty list of zone exports')
@decorators.attr(type='slow')
@decorators.idempotent_id('cac8ea8e-c33b-11ed-a033-201e8823901f')
def test_alt_delete_shared_zone_export(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_delete_shared_zone_export')[0]
self.assertRaises(
lib_exc.NotFound,
self.alt_export_client.delete_zone_export, zone['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('962df772-c33d-11ed-a033-201e8823901f')
def test_alt_fails_to_show_exported_zonefile_for_shared_zone(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_show_exported_zonefile_for_shared_zone')[0]
self.assertRaises(
lib_exc.NotFound,
self.alt_export_client.show_exported_zonefile, zone['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('089136f2-c3e4-11ed-a102-201e8823901f')
def test_alt_shows_shared_zones_nameservers(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_shows_shared_zones_nameservers')[0]
self.assertRaises(
lib_exc.NotFound,
self.alt_zone_client.show_zone_nameservers, zone['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('6376d4ca-c3f6-11ed-a102-201e8823901f')
def test_alt_transfers_shared_zone(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_transfers_shared_zone')[0]
# Alt creates a zone transfer_request
self.assertRaises(
lib_exc.NotFound,
self.alt_transfer_client.create_transfer_request, zone['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('80ffbd8a-c3f7-11ed-a102-201e8823901f')
def test_alt_show_delete_transfers_of_shared_zone(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_show_delete_transfers_of_shared_zone')[0]
# Primary user creates a zone transfer_request
transfer_request = self.prm_transfer_client.create_transfer_request(
zone['id'])[1]
self.addCleanup(
self.prm_transfer_client.delete_transfer_request,
transfer_request['id'])
self.assertEqual('ACTIVE', transfer_request['status'])
# Alt shows a zone transfer_request
self.assertRaises(
lib_exc.NotFound,
self.alt_transfer_client.show_transfer_request, zone['id'])
# Alt deletes a zone transfer_request
self.assertRaises(
lib_exc.NotFound,
self.alt_transfer_client.delete_transfer_request, zone['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('1da0ff64-c3f8-11ed-a102-201e8823901f')
def test_alt_lists_transfers_of_shared_zone(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_lists_transfers_of_shared_zone')[0]
# Primary user creates a zone transfer_request
transfer = self.prm_transfer_client.create_transfer_request(
zone['id'])[1]
self.addCleanup(self.prm_transfer_client.delete_transfer_request,
transfer['id'])
self.assertEqual('ACTIVE', transfer['status'])
transfer = self.prm_transfer_client.list_transfer_requests()[1]
self.assertEqual(
1, len(transfer['transfer_requests']),
'Failed, there is no transfer request listed for a primary user')
# Alt user lists shared zone transfer requests
transfer = self.alt_transfer_client.list_transfer_requests()[1]
self.assertEqual(
0, len(transfer['transfer_requests']),
'Failed, transfer request list should be empty for for Alt user')
@decorators.attr(type='slow')
@decorators.idempotent_id('1702c1d6-c643-11ed-8d86-201e8823901f')
def test_alt_abandon_shared_zone(self):
# Primary creates Zone and shares it with Alt
zone = self._create_shared_zone(
'test_alt_lists_transfers_of_shared_zone')[0]
self.assertRaises(
lib_exc.Forbidden, self.alt_zone_client.abandon_zone,
zone['id'])