openstack
/
designate-tempest-plugin
Code Issues Proposed changes

Merge "Make the API tests compatible with scoped tokens"

This commit is contained in:
Zuul 2021-10-20 06:14:04 +00:00 committed by Gerrit Code Review
parent a9804e184a a3a236331c
commit e713aa22b8
23 changed files with 381 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
designate_tempest_plugin/common/waiters.py
View File

@ -51,16 +51,16 @@ def wait_for_zone_404(client, zone_id):
raise lib_exc.TimeoutException(message)
def wait_for_zone_status(client, zone_id, status):
def wait_for_zone_status(client, zone_id, status, headers=None):
"""Waits for a zone to reach given status."""
LOG.info('Waiting for zone %s to reach %s', zone_id, status)
_, zone = client.show_zone(zone_id)
_, zone = client.show_zone(zone_id, headers=headers)
start = int(time.time())
while zone['status'] != status:
time.sleep(client.build_interval)
_, zone = client.show_zone(zone_id)
_, zone = client.show_zone(zone_id, headers=headers)
status_curr = zone['status']
if status_curr == status:
LOG.info('Zone %s reached %s', zone_id, status)

20
designate_tempest_plugin/config.py
View File

@ -79,4 +79,24 @@ DnsFeatureGroup = [
default=True,
help="Is https://bugs.launchpad.net/designate/+bug/1573141 "
"fixed"),
# Note: Also see the enforce_scope section (from tempest) for Designate API
# scope checking setting.
cfg.BoolOpt('enforce_new_defaults',
default=False,
help='Does the dns service API policies enforce '
'the new keystone default roles? This configuration '
'value should be same as designate.conf: '
'[oslo_policy].enforce_new_defaults option.'),
]
# Extending this enforce_scope group defined in tempest
enforce_scope_group = cfg.OptGroup(name="enforce_scope",
title="OpenStack Services with "
"enforce scope")
EnforceScopeGroup = [
cfg.BoolOpt('designate',
default=False,
help='Does the dns service API policies enforce '
'scope? This configuration value should be same as '
'designate.conf: [oslo_policy].enforce_scope option.'),
]

2
designate_tempest_plugin/plugin.py
View File

@ -58,6 +58,8 @@ class DesignateTempestPlugin(plugins.TempestPlugin):
project_config.DnsGroup)
config.register_opt_group(conf, project_config.dns_feature_group,
project_config.DnsFeatureGroup)
config.register_opt_group(conf, project_config.enforce_scope_group,
project_config.EnforceScopeGroup)
def get_opt_lists(self):
"""

31
designate_tempest_plugin/services/dns/v2/json/pool_client.py
View File

@ -22,7 +22,8 @@ class PoolClient(base.DnsClientV2Base):
"""API V2 Tempest REST client for Pool API"""
@base.handle_errors
def create_pool(self, pool_name=None, ns_records=None, params=None):
def create_pool(self, pool_name=None, ns_records=None, params=None,
project_id=None):
"""Create a pool with the specified parameters.
:param pool_name: name of the pool.
Default: Random Value.
@ -30,12 +31,15 @@ class PoolClient(base.DnsClientV2Base):
with priority.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param project_id: The project ID to associate the pool with.
:return: A tuple with the server response and the created pool.
"""
pool = {
"name": pool_name or data_utils.rand_name(name="Demo pool"),
"ns_records": ns_records or dns_data_utils.rand_ns_records()
}
if project_id:
pool["project_id"] = project_id
resp, body = self._create_request('pools', data=pool, params=params)
@ -45,33 +49,38 @@ class PoolClient(base.DnsClientV2Base):
return resp, body
@base.handle_errors
def show_pool(self, uuid, params=None):
def show_pool(self, uuid, params=None, headers=None):
"""Gets a specific pool.
:param uuid: Unique identifier of the pool in UUID format.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param headers: The optional headers to include in the show request.
:return: Serialized pool as a dictionary.
"""
return self._show_request('pools', uuid, params=params)
return self._show_request('pools', uuid, params=params,
headers=headers)
@base.handle_errors
def list_pools(self, params=None):
def list_pools(self, params=None, headers=None):
"""Gets a list of pools.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param headers: The optional headers to include in the list request.
:return: Serialized pools as a list.
"""
return self._list_request('pools', params=params)
return self._list_request('pools', params=params, headers=headers)
@base.handle_errors
def delete_pool(self, uuid, params=None):
def delete_pool(self, uuid, params=None, headers=None):
"""Deletes a pool having the specified UUID.
:param uuid: The unique identifier of the pool.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param headers: The optional headers to include in the delete request.
:return: A tuple with the server response and the response body.
"""
resp, body = self._delete_request('pools', uuid, params=params)
resp, body = self._delete_request('pools', uuid, params=params,
headers=headers)
# Delete Pool should Return a HTTP 204
self.expected_success(204, resp.status)
@ -80,7 +89,7 @@ class PoolClient(base.DnsClientV2Base):
@base.handle_errors
def update_pool(self, uuid, pool_name=None, ns_records=None,
params=None):
params=None, headers=None, extra_headers=False):
"""Update a pool with the specified parameters.
:param uuid: The unique identifier of the pool.
:param pool_name: name of the pool.
@ -90,6 +99,9 @@ class PoolClient(base.DnsClientV2Base):
Default: Random Value.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param headers: The optional headers to include in the update request.
:param extra_headers: A boolean indicating that the headers replace
default headers or are in addition.
:return: A tuple with the server response and the updated pool.
"""
@ -99,7 +111,8 @@ class PoolClient(base.DnsClientV2Base):
}
resp, body = self._update_request('pools', uuid, pool,
params=params)
params=params, headers=headers,
extra_headers=extra_headers)
# Update Pool should Return a HTTP 202
self.expected_success(202, resp.status)

18
designate_tempest_plugin/services/dns/v2/json/quotas_client.py
View File

@ -19,12 +19,13 @@ from designate_tempest_plugin.services.dns.v2.json import base
class QuotasClient(base.DnsClientV2Base):
@base.handle_errors
def update_quotas(self, zones=None, zone_records=None,
def update_quotas(self, project_id, zones=None, zone_records=None,
zone_recordsets=None, recordset_records=None,
api_export_size=None, project_id=None, params=None,
api_export_size=None, params=None,
headers=None):
"""Update the quotas for the project id
:param project_id: Apply the quotas to this project id
:param zones: The limit on zones per tenant
Default: Random Value
:param zone_records: The limit on records per zone
@ -35,15 +36,11 @@ class QuotasClient(base.DnsClientV2Base):
Default: Random Value
:param api_export_size: The limit on size of on exported zone
Default: Random Value
:param project_id: Apply the quotas to this project id
Default: The project id of this client
:param params: A Python dict that represents the query paramaters 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 created quota.
"""
project_id = project_id or self.tenant_id
quotas = dns_data_utils.rand_quotas(
zones=zones,
zone_records=zone_records,
@ -61,33 +58,28 @@ class QuotasClient(base.DnsClientV2Base):
return resp, body
@base.handle_errors
def show_quotas(self, project_id=None, params=None, headers=None):
def show_quotas(self, project_id, params=None, headers=None):
"""Gets a specific quota.
:param project_id: Show the quotas of this project id
Default: The project id of this client
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param headers (dict): The headers to use for the request.
:return: Serialized quota as a dictionary.
"""
project_id = project_id or self.tenant_id
return self._show_request('quotas', project_id, params=params,
headers=headers, extra_headers=True)
@base.handle_errors
def delete_quotas(self, project_id=None, params=None, headers=None):
def delete_quotas(self, project_id, params=None, headers=None):
"""Resets the quotas for the specified project id
:param project_id: Reset the quotas of this project id
Default: The project id of this client
:param params: A Python dict that represents the query paramaters 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 response body.
"""
project_id = project_id or self.tenant_id
resp, body = self._delete_request(
'quotas', project_id,
params=params, headers=headers,

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

@ -27,7 +27,7 @@ class ZonesClient(base.DnsClientV2Base):
def create_zone(self, name=None, email=None, ttl=None, description=None,
attributes=None, wait_until=False,
zone_type=const.PRIMARY_ZONE_TYPE,
primaries=None, params=None):
primaries=None, params=None, project_id=None):
"""Create a zone with the specified parameters.
@ -50,6 +50,8 @@ class ZonesClient(base.DnsClientV2Base):
Default: None
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param project_id: When specified, overrides the project ID the zone
will be associated with.
:return: A tuple with the server response and the created zone.
"""
@ -76,13 +78,23 @@ class ZonesClient(base.DnsClientV2Base):
' for a SECONDARY zone type')
zone['masters'] = primaries
resp, body = self._create_request('zones', zone, params=params)
headers = None
extra_headers = False
if project_id:
headers = {'x-auth-sudo-project-id': project_id}
extra_headers = True
resp, body = self._create_request('zones', zone, params=params,
headers=headers,
extra_headers=extra_headers)
# Create Zone should Return a HTTP 202
self.expected_success(202, resp.status)
if wait_until:
waiters.wait_for_zone_status(self, body['id'], wait_until)
waiters.wait_for_zone_status(self, body['id'], wait_until,
headers=headers)
return resp, body

34
designate_tempest_plugin/tests/api/admin/test_quotas.py
View File

@ -18,8 +18,8 @@ from tempest.lib import decorators
from designate_tempest_plugin.tests import base
from designate_tempest_plugin import data_utils as dns_data_utils
LOG = logging.getLogger(__name__)
CONF = config.CONF
LOG = logging.getLogger(__name__)
class BaseQuotasTest(base.BaseDnsAdminTest):
@ -34,12 +34,14 @@ class BaseQuotasTest(base.BaseDnsAdminTest):
class QuotasAdminTest(BaseQuotasTest):
credentials = ["admin"]
credentials = ["admin", "primary", "system_admin"]
def setUp(self):
super(QuotasAdminTest, self).setUp()
_, original_quotas = self.admin_client.show_quotas()
_, original_quotas = self.admin_client.show_quotas(
project_id=self.quotas_client.project_id)
self.addCleanup(self.admin_client.update_quotas,
project_id=self.quotas_client.project_id,
**original_quotas['quota'])
@classmethod
@ -51,18 +53,24 @@ class QuotasAdminTest(BaseQuotasTest):
@classmethod
def setup_clients(cls):
super(QuotasAdminTest, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_admin.QuotasClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_admin.QuotasClient()
else:
cls.admin_client = cls.os_admin.dns_admin.QuotasClient()
cls.quotas_client = cls.os_primary.dns_v2.QuotasClient()
@decorators.idempotent_id('ed42f367-e5ba-40d7-a08d-366ad787d21c')
def test_show_quotas(self):
LOG.info("Updating quotas")
quotas = dns_data_utils.rand_quotas()
_, body = self.admin_client.update_quotas(**quotas)
self.addCleanup(self.admin_client.delete_quotas)
_, body = self.admin_client.update_quotas(
project_id=self.quotas_client.project_id, **quotas)
self.addCleanup(self.admin_client.delete_quotas,
project_id=self.quotas_client.project_id)
LOG.info("Fetching quotas")
_, body = self.admin_client.show_quotas()
_, body = self.admin_client.show_quotas(
project_id=self.quotas_client.project_id)
LOG.info("Ensuring the response has all quota types")
self.assertExpected(quotas, body['quota'], self.excluded_keys)
@ -70,7 +78,8 @@ class QuotasAdminTest(BaseQuotasTest):
@decorators.idempotent_id('33e0affb-5d66-4216-881c-f101a779851a')
def test_delete_quotas(self):
LOG.info("Deleting quotas")
_, body = self.admin_client.delete_quotas()
_, body = self.admin_client.delete_quotas(
project_id=self.quotas_client.project_id)
LOG.info("Ensuring an empty response body")
self.assertEqual(body.strip(), b"")
@ -79,8 +88,11 @@ class QuotasAdminTest(BaseQuotasTest):
def test_update_quotas(self):
LOG.info("Updating quotas")
quotas = dns_data_utils.rand_quotas()
_, body = self.admin_client.update_quotas(**quotas)
self.addCleanup(self.admin_client.delete_quotas)
_, body = self.admin_client.update_quotas(
project_id=self.quotas_client.project_id,
**quotas)
self.addCleanup(self.admin_client.delete_quotas,
project_id=self.quotas_client.project_id)
LOG.info("Ensuring the response has all quota types")
self.assertExpected(quotas, body['quota'], self.excluded_keys)

23
designate_tempest_plugin/tests/api/v2/test_blacklists.py
View File

@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from tempest.lib.common.utils import data_utils
@ -19,6 +20,7 @@ from tempest.lib.common.utils import data_utils
from designate_tempest_plugin import data_utils as dns_data_utils
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -28,7 +30,7 @@ class BaseBlacklistsTest(base.BaseDnsV2Test):
class BlacklistsAdminTest(BaseBlacklistsTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -39,7 +41,10 @@ class BlacklistsAdminTest(BaseBlacklistsTest):
@classmethod
def setup_clients(cls):
super(BlacklistsAdminTest, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.BlacklistsClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.BlacklistsClient()
else:
cls.admin_client = cls.os_admin.dns_v2.BlacklistsClient()
@decorators.idempotent_id('3a7f7564-6bdd-446e-addc-a3475b4c3f71')
def test_create_blacklist(self):
@ -112,7 +117,7 @@ class BlacklistsAdminTest(BaseBlacklistsTest):
class TestBlacklistNotFoundAdmin(BaseBlacklistsTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -123,7 +128,10 @@ class TestBlacklistNotFoundAdmin(BaseBlacklistsTest):
@classmethod
def setup_clients(cls):
super(TestBlacklistNotFoundAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.BlacklistsClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.BlacklistsClient()
else:
cls.admin_client = cls.os_admin.dns_v2.BlacklistsClient()
@decorators.idempotent_id('9d65b638-fe98-47a8-853f-fa9244d144cc')
def test_show_blacklist_404(self):
@ -155,7 +163,7 @@ class TestBlacklistNotFoundAdmin(BaseBlacklistsTest):
class TestBlacklistInvalidIdAdmin(BaseBlacklistsTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -166,7 +174,10 @@ class TestBlacklistInvalidIdAdmin(BaseBlacklistsTest):
@classmethod
def setup_clients(cls):
super(TestBlacklistInvalidIdAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.BlacklistsClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.BlacklistsClient()
else:
cls.admin_client = cls.os_admin.dns_v2.BlacklistsClient()
@decorators.idempotent_id('c7bae53f-2edc-45d8-b254-8a81482728c1')
def test_show_blacklist_invalid_uuid(self):

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

@ -12,15 +12,17 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
class DesignateLimit(base.BaseDnsV2Test):
credentials = ['admin']
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -31,7 +33,11 @@ class DesignateLimit(base.BaseDnsV2Test):
@classmethod
def setup_clients(cls):
super(DesignateLimit, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.DesignateLimitClient()
if CONF.enforce_scope.designate:
cls.admin_client = (cls.os_system_admin.dns_v2.
DesignateLimitClient())
else:
cls.admin_client = cls.os_admin.dns_v2.DesignateLimitClient()
@decorators.idempotent_id('828572be-8662-11eb-8ff2-74e5f9e2a801')
def test_list_designate_limits(self):

96
designate_tempest_plugin/tests/api/v2/test_pool.py
View File

@ -15,12 +15,14 @@
from operator import itemgetter
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from tempest.lib.common.utils import data_utils
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -34,7 +36,7 @@ class BasePoolTest(base.BaseDnsV2Test):
class PoolAdminTest(BasePoolTest):
credentials = ['admin']
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -45,12 +47,16 @@ class PoolAdminTest(BasePoolTest):
@classmethod
def setup_clients(cls):
super(PoolAdminTest, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.PoolClient()
else:
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
@decorators.idempotent_id('69257f7c-b3d5-4e1b-998e-0677ad12f125')
def test_create_pool(self):
pool_data = {
"name": "Example Pool",
"project_id": "1",
"ns_records": [{
"hostname": "ns1.example.org.",
"priority": 1}
@ -58,8 +64,10 @@ class PoolAdminTest(BasePoolTest):
}
LOG.info('Create a pool')
_, pool = self.admin_client.create_pool(pool_name=pool_data["name"],
ns_records=pool_data["ns_records"])
self.addCleanup(self.admin_client.delete_pool, pool['id'])
ns_records=pool_data["ns_records"],
project_id=pool_data["project_id"])
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
self.assertEqual(pool_data["name"], pool['name'])
self.assertExpected(pool_data, pool, self.excluded_keys)
@ -67,11 +75,13 @@ class PoolAdminTest(BasePoolTest):
@decorators.idempotent_id('e80eb70a-8ee5-40eb-b06e-599597a8ab7e')
def test_show_pool(self):
LOG.info('Create a pool')
_, pool = self.admin_client.create_pool()
self.addCleanup(self.admin_client.delete_pool, pool['id'])
_, pool = self.admin_client.create_pool(project_id="1")
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
LOG.info('Fetch the pool')
_, body = self.admin_client.show_pool(pool['id'])
_, body = self.admin_client.show_pool(
pool['id'], headers=self.all_projects_header)
LOG.info('Ensure the fetched response matches the created pool')
self.assertExpected(pool, body, self.excluded_keys)
@ -81,36 +91,43 @@ class PoolAdminTest(BasePoolTest):
@decorators.idempotent_id('d8c4c377-5d88-452d-a4d2-c004d72e1abe')
def test_delete_pool(self):
LOG.info('Create a pool')
_, pool = self.admin_client.create_pool()
_, pool = self.admin_client.create_pool(project_id="1")
self.addCleanup(self.admin_client.delete_pool, pool['id'],
ignore_errors=lib_exc.NotFound)
ignore_errors=lib_exc.NotFound,
headers=self.all_projects_header)
LOG.info('Delete the pool')
_, body = self.admin_client.delete_pool(pool['id'])
_, body = self.admin_client.delete_pool(
pool['id'], headers=self.all_projects_header)
self.assertRaises(lib_exc.NotFound,
lambda: self.admin_client.show_pool(pool['id']))
lambda: self.admin_client.show_pool(
pool['id'], headers=self.all_projects_header))
@decorators.idempotent_id('77c85b40-83b2-4c17-9fbf-e6d516cfce90')
def test_list_pools(self):
LOG.info('Create a pool')
_, pool = self.admin_client.create_pool()
self.addCleanup(self.admin_client.delete_pool, pool['id'])
_, pool = self.admin_client.create_pool(project_id="1")
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
LOG.info('List pools')
_, body = self.admin_client.list_pools()
_, body = self.admin_client.list_pools(
headers=self.all_projects_header)
self.assertGreater(len(body['pools']), 0)
@decorators.idempotent_id('fdcc84ce-af65-4af6-a5fc-6c50acbea0f0')
def test_update_pool(self):
LOG.info('Create a pool')
_, pool = self.admin_client.create_pool()
self.addCleanup(self.admin_client.delete_pool, pool['id'])
_, pool = self.admin_client.create_pool(project_id="1")
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
LOG.info('Update the pool')
_, patch_pool = self.admin_client.update_pool(
pool['id'], pool_name="foo")
pool['id'], pool_name="foo", headers=self.all_projects_header,
extra_headers=True)
self.assertEqual("foo", patch_pool["name"])
@ -124,7 +141,7 @@ class PoolAdminTest(BasePoolTest):
class TestPoolNotFoundAdmin(BasePoolTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -135,7 +152,10 @@ class TestPoolNotFoundAdmin(BasePoolTest):
@classmethod
def setup_clients(cls):
super(TestPoolNotFoundAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.PoolClient()
else:
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
@decorators.idempotent_id('56281b2f-dd5a-4376-8c32-aba771062fa5')
def test_show_pool_404(self):
@ -167,7 +187,7 @@ class TestPoolNotFoundAdmin(BasePoolTest):
class TestPoolInvalidIdAdmin(BasePoolTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -178,7 +198,10 @@ class TestPoolInvalidIdAdmin(BasePoolTest):
@classmethod
def setup_clients(cls):
super(TestPoolInvalidIdAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.PoolClient()
else:
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
@decorators.idempotent_id('081d0188-42a7-4953-af0e-b022960715e2')
def test_show_pool_invalid_uuid(self):
@ -211,7 +234,7 @@ class TestPoolInvalidIdAdmin(BasePoolTest):
class TestPoolAdminNegative(BasePoolTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -222,7 +245,10 @@ class TestPoolAdminNegative(BasePoolTest):
@classmethod
def setup_clients(cls):
super(TestPoolAdminNegative, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.PoolClient()
else:
cls.admin_client = cls.os_admin.dns_v2.PoolClient()
@decorators.idempotent_id('0a8cdc1e-ac02-11eb-ae06-74e5f9e2a801')
def test_create_pool_invalid_name(self):
@ -249,37 +275,43 @@ class TestPoolAdminNegative(BasePoolTest):
# Note: Update pool API is deprecated for removal.
def test_update_pool_with_invalid_name(self):
LOG.info('Create a pool')
pool = self.admin_client.create_pool()[1]
self.addCleanup(self.admin_client.delete_pool, pool['id'])
pool = self.admin_client.create_pool(project_id="1")[1]
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
LOG.info('Update the pool using a name that is too long')
with self.assertRaisesDns(lib_exc.BadRequest, 'invalid_object', 400):
self.admin_client.update_pool(
pool['id'],
pool_name=data_utils.rand_name(name="Huge_size_name") * 10000)
pool_name=data_utils.rand_name(name="Huge_size_name") * 10000,
headers=self.all_projects_header, extra_headers=True)
@decorators.idempotent_id('2e496596-ac07-11eb-ae06-74e5f9e2a801')
def test_update_pool_with_invalid_hostname_in_ns_records(self):
# Note: Update pool API is deprecated for removal.
LOG.info('Create a pool')
pool = self.admin_client.create_pool()[1]
self.addCleanup(self.admin_client.delete_pool, pool['id'])
pool = self.admin_client.create_pool(project_id="1")[1]
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
LOG.info('Update the pool using invalid hostname in ns_records')
with self.assertRaisesDns(lib_exc.BadRequest, 'invalid_object', 400):
self.admin_client.update_pool(
pool['id'],
ns_records=[{"hostname": "ns1_example_org_", "priority": 1}])
ns_records=[{"hostname": "ns1_example_org_", "priority": 1}],
headers=self.all_projects_header, extra_headers=True)
@decorators.idempotent_id('3e934624-ac07-11eb-ae06-74e5f9e2a801')
def test_update_pool_with_invalid_priority_in_ns_records(self):
# Note: Update pool API is deprecated for removal.
LOG.info('Create a pool')
pool = self.admin_client.create_pool()[1]
self.addCleanup(self.admin_client.delete_pool, pool['id'])
pool = self.admin_client.create_pool(project_id="1")[1]
self.addCleanup(self.admin_client.delete_pool, pool['id'],
headers=self.all_projects_header)
LOG.info('Update the pool using invalid priority in ns_records')
with self.assertRaisesDns(lib_exc.BadRequest, 'invalid_object', 400):
self.admin_client.update_pool(
pool['id'],
ns_records=[{"hostname": "ns1.example.org.", "priority": -1}])
ns_records=[{"hostname": "ns1.example.org.", "priority": -1}],
headers=self.all_projects_header, extra_headers=True)

7
designate_tempest_plugin/tests/api/v2/test_ptrs.py
View File

@ -19,9 +19,8 @@ from tempest.lib import exceptions as lib_exc
from designate_tempest_plugin.tests import base
import tempest.test
LOG = logging.getLogger(__name__)
CONF = config.CONF
LOG = logging.getLogger(__name__)
class BasePtrTest(base.BaseDnsV2Test):
@ -30,7 +29,7 @@ class BasePtrTest(base.BaseDnsV2Test):
class DesignatePtrRecord(BasePtrTest, tempest.test.BaseTestCase):
credentials = ['primary']
credentials = ["primary"]
@classmethod
def setup_credentials(cls):
@ -89,7 +88,7 @@ class DesignatePtrRecord(BasePtrTest, tempest.test.BaseTestCase):
class DesignatePtrRecordNegative(BasePtrTest, tempest.test.BaseTestCase):
credentials = ['primary']
credentials = ["primary"]
@classmethod
def setup_credentials(cls):

89
designate_tempest_plugin/tests/api/v2/test_quotas.py
View File

@ -19,15 +19,13 @@ from tempest.lib import exceptions as lib_exc
from designate_tempest_plugin.tests import base
from designate_tempest_plugin import data_utils as dns_data_utils
LOG = logging.getLogger(__name__)
CONF = config.CONF
LOG = logging.getLogger(__name__)
class QuotasV2Test(base.BaseDnsV2Test):
credentials = ['primary', 'admin', 'alt']
credentials = ["primary", "admin", "system_admin", "alt"]
@classmethod
def setup_credentials(cls):
@ -48,16 +46,17 @@ class QuotasV2Test(base.BaseDnsV2Test):
def setup_clients(cls):
super(QuotasV2Test, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.QuotasClient()
else:
cls.admin_client = cls.os_admin.dns_v2.QuotasClient()
cls.quotas_client = cls.os_primary.dns_v2.QuotasClient()
cls.admin_client = cls.os_admin.dns_v2.QuotasClient()
cls.alt_client = cls.os_alt.dns_v2.QuotasClient()
def _store_quotas(self, project_id=None, cleanup=True):
def _store_quotas(self, project_id, cleanup=True):
"""Remember current quotas and reset them after the test"""
params = {}
if project_id:
params['project_id'] = project_id
params['headers'] = {'X-Auth-All-Projects': True}
params = {'project_id': project_id,
'headers': self.all_projects_header}
_r, original_quotas = self.admin_client.show_quotas(**params)
params.update(original_quotas)
@ -67,33 +66,45 @@ class QuotasV2Test(base.BaseDnsV2Test):
@decorators.idempotent_id('1dac991a-9e2e-452c-a47a-26ac37381ec5')
def test_show_quotas(self):
self._store_quotas()
self._store_quotas(project_id=self.quotas_client.project_id)
LOG.info("Updating quotas")
quotas = dns_data_utils.rand_quotas()
_, body = self.admin_client.update_quotas(**quotas)
self.addCleanup(self.admin_client.delete_quotas)
_, body = self.admin_client.update_quotas(
project_id=self.quotas_client.project_id,
headers=self.all_projects_header,
**quotas)
LOG.info("Fetching quotas")
_, body = self.admin_client.show_quotas()
_, body = self.admin_client.show_quotas(
project_id=self.quotas_client.project_id,
headers=self.all_projects_header)
LOG.info("Ensuring the response has all quota types")
self.assertExpected(quotas, body, [])
@decorators.idempotent_id('0448b089-5803-4ce3-8a6c-5c15ff75a2cc')
def test_delete_quotas(self):
self._store_quotas()
self._store_quotas(project_id=self.quotas_client.project_id)
LOG.info("Deleting quotas")
_, body = self.admin_client.delete_quotas()
_, body = self.admin_client.delete_quotas(
project_id=self.quotas_client.project_id,
headers=self.all_projects_header)
LOG.info("Ensuring an empty response body")
self.assertEqual(body.strip(), b"")
@decorators.idempotent_id('76d24c87-1b39-4e19-947c-c08e1380dc61')
def test_update_quotas(self):
self._store_quotas()
if CONF.enforce_scope.designate:
raise self.skipException(
"System scoped tokens do not have a project_id.")
self._store_quotas(project_id=self.admin_client.project_id)
LOG.info("Updating quotas")
quotas = dns_data_utils.rand_quotas()
_, body = self.admin_client.update_quotas(**quotas)
_, body = self.admin_client.update_quotas(
project_id=self.admin_client.project_id,
**quotas)
LOG.info("Ensuring the response has all quota types")
self.assertExpected(quotas, body, [])
@ -101,21 +112,22 @@ class QuotasV2Test(base.BaseDnsV2Test):
@decorators.idempotent_id('9b09b3e2-7e88-4569-bce3-9be2f7ac70c3')
def test_update_quotas_other_project(self):
project_id = self.quotas_client.tenant_id
project_id = self.quotas_client.project_id
self._store_quotas(project_id=project_id)
LOG.info("Updating quotas for %s ", project_id)
quotas = dns_data_utils.rand_quotas()
request = quotas.copy()
request['project_id'] = project_id
request['headers'] = {'X-Auth-All-Projects': True}
_, body = self.admin_client.update_quotas(**request)
_, body = self.admin_client.update_quotas(
project_id=project_id,
headers=self.all_projects_header,
**request)
LOG.info("Ensuring the response has all quota types")
self.assertExpected(quotas, body, [])
_, client_body = self.quotas_client.show_quotas()
_, client_body = self.quotas_client.show_quotas(project_id=project_id)
self.assertExpected(quotas, client_body, [])
@ -129,24 +141,28 @@ class QuotasV2Test(base.BaseDnsV2Test):
LOG.info("Resetting quotas to default for %s ", project_id)
self.admin_client.delete_quotas(
project_id=project_id,
headers={'X-Auth-All-Projects': True})
headers=self.all_projects_header)
_, default_quotas = self.admin_client.show_quotas(
project_id=project_id, headers={'X-Auth-All-Projects': True})
project_id=project_id,
headers=self.all_projects_header)
LOG.info("Updating quotas for %s ", project_id)
quotas = dns_data_utils.rand_quotas()
request = quotas.copy()
request['project_id'] = project_id
request['headers'] = {'X-Auth-All-Projects': True}
_, body = self.admin_client.update_quotas(**request)
_, body = self.admin_client.update_quotas(
project_id=project_id,
headers=self.all_projects_header,
**request)
self.admin_client.delete_quotas(
project_id=project_id,
headers={'X-Auth-All-Projects': True})
headers=self.all_projects_header)
_, final_quotas = self.admin_client.show_quotas(
project_id=project_id, headers={'X-Auth-All-Projects': True})
project_id=project_id,
headers=self.all_projects_header)
self.assertExpected(default_quotas, final_quotas, [])
@ -158,18 +174,13 @@ class QuotasV2Test(base.BaseDnsV2Test):
"is not being verified.")
project_id = 'project-that-does-not-exist'
original_quotas = self._store_quotas(project_id=project_id,
cleanup=False)
LOG.info("Updating quotas for non-existing %s ", project_id)
quotas = dns_data_utils.rand_quotas()
request = quotas.copy()
request['project_id'] = project_id
request['headers'] = {'X-Auth-All-Projects': True}
with self.assertRaisesDns(lib_exc.BadRequest, 'invalid_project', 400):
self.admin_client.update_quotas(**request)
_, client_body = self.quotas_client.show_quotas()
self.assertExpected(original_quotas, client_body, [])
self.admin_client.update_quotas(
project_id=project_id,
headers=self.all_projects_header,
**request)

28
designate_tempest_plugin/tests/api/v2/test_recordset.py
View File

@ -22,9 +22,8 @@ from designate_tempest_plugin.tests import base
from designate_tempest_plugin.common import waiters
from designate_tempest_plugin import data_utils
LOG = logging.getLogger(__name__)
CONF = config.CONF
LOG = logging.getLogger(__name__)
class BaseRecordsetsTest(base.BaseDnsV2Test):
@ -50,7 +49,7 @@ class BaseRecordsetsTest(base.BaseDnsV2Test):
@ddt.ddt
class RecordsetsTest(BaseRecordsetsTest):
credentials = ["admin", 'primary', 'alt']
credentials = ["admin", "system_admin", "primary", "alt"]
@classmethod
def setup_credentials(cls):
@ -61,12 +60,16 @@ class RecordsetsTest(BaseRecordsetsTest):
@classmethod
def setup_clients(cls):
super(RecordsetsTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.RecordsetClient()
cls.admin_zone_client = cls.os_system_admin.dns_v2.ZonesClient()
else:
cls.admin_client = cls.os_admin.dns_v2.RecordsetClient()
cls.admin_zone_client = cls.os_admin.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.RecordsetClient()
cls.alt_client = cls.os_alt.dns_v2.RecordsetClient()
cls.admin_client = cls.os_admin.dns_v2.RecordsetClient()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.alt_zone_client = cls.os_alt.dns_v2.ZonesClient()
cls.admin_zone_client = cls.os_admin.dns_v2.ZonesClient()
@decorators.attr(type='smoke')
@decorators.idempotent_id('631d74fd-6909-4684-a61b-5c4d2f92c3e7')
@ -276,7 +279,7 @@ class RecordsetsTest(BaseRecordsetsTest):
self.assertRaises(lib_exc.Forbidden,
lambda: self.alt_client.list_recordset(
self.zone['id'],
headers={'x-auth-all-projects': True}))
headers=self.all_projects_header))
LOG.info('Re-Fetch Recordsets as Admin tenant for a Primary project '
'using "x-auth-all-projects" HTTP header.')
@ -286,7 +289,7 @@ class RecordsetsTest(BaseRecordsetsTest):
primary_recordsets_ids = [
item['id'] for item in self.admin_client.list_recordset(
self.zone['id'],
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'limit': 1000})[1]['recordsets']]
for recordset_id in [body_pr_1['id'], body_pr_2['id']]:
@ -299,7 +302,7 @@ class RecordsetsTest(BaseRecordsetsTest):
@ddt.ddt
class RecordsetsNegativeTest(BaseRecordsetsTest):
credentials = ['primary', 'alt']
credentials = ["primary", "alt"]
@classmethod
def setup_credentials(cls):
@ -540,7 +543,7 @@ class RootRecordsetsTests(BaseRecordsetsTest):
class RecordsetOwnershipTest(BaseRecordsetsTest):
credentials = ['primary', 'alt', 'admin']
credentials = ["primary", "alt", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -551,9 +554,12 @@ class RecordsetOwnershipTest(BaseRecordsetsTest):
@classmethod
def setup_clients(cls):
super(RecordsetOwnershipTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.RecordsetClient()
else:
cls.admin_client = cls.os_admin.dns_v2.RecordsetClient()
cls.client = cls.os_primary.dns_v2.RecordsetClient()
cls.alt_client = cls.os_alt.dns_v2.RecordsetClient()
cls.admin_client = cls.os_admin.dns_v2.RecordsetClient()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.alt_zone_client = cls.os_alt.dns_v2.ZonesClient()
@ -706,7 +712,7 @@ class RecordsetOwnershipTest(BaseRecordsetsTest):
# in parallel will impact the list result set. Since the default
# pagination limit is only 20, we set a param limit of 1000 here.
recordsets = self.admin_client.list_owned_recordsets(
headers={'x-auth-all-projects': True}, params={'limit': 1000})
headers=self.all_projects_header, params={'limit': 1000})
LOG.info('Received by API recordsets are {} '.format(recordsets))
project_ids_api = set([item['project_id'] for item in recordsets])
for prj_id in project_ids_used:

12
designate_tempest_plugin/tests/api/v2/test_service_statuses.py
View File

@ -18,15 +18,13 @@ from tempest.lib import decorators
from designate_tempest_plugin.tests import base
LOG = logging.getLogger(__name__)
CONF = config.CONF
LOG = logging.getLogger(__name__)
class ServiceStatus(base.BaseDnsV2Test):
credentials = ['primary', 'admin']
credentials = ["primary", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -37,9 +35,11 @@ class ServiceStatus(base.BaseDnsV2Test):
@classmethod
def setup_clients(cls):
super(ServiceStatus, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.ServiceClient()
else:
cls.admin_client = cls.os_admin.dns_v2.ServiceClient()
cls.client = cls.os_primary.dns_v2.ServiceClient()
cls.admin_client = cls.os_admin.dns_v2.ServiceClient()
@decorators.idempotent_id('bf277a76-8583-11eb-a557-74e5f9e2a801')
def test_list_service_statuses(self):

23
designate_tempest_plugin/tests/api/v2/test_tld.py
View File

@ -13,12 +13,14 @@
# under the License.
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from tempest.lib.common.utils import data_utils
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -27,7 +29,7 @@ class BaseTldTest(base.BaseDnsV2Test):
class TldAdminTest(BaseTldTest):
credentials = ['admin']
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -38,7 +40,10 @@ class TldAdminTest(BaseTldTest):
@classmethod
def setup_clients(cls):
super(TldAdminTest, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.TldClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.TldClient()
else:
cls.admin_client = cls.os_admin.dns_v2.TldClient()
@classmethod
def resource_setup(cls):
@ -129,7 +134,7 @@ class TldAdminTest(BaseTldTest):
class TestTldNotFoundAdmin(BaseTldTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -140,7 +145,10 @@ class TestTldNotFoundAdmin(BaseTldTest):
@classmethod
def setup_clients(cls):
super(TestTldNotFoundAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.TldClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.TldClient()
else:
cls.admin_client = cls.os_admin.dns_v2.TldClient()
@decorators.idempotent_id('b237d5ee-0d76-4294-a3b6-c2f8bf4b0e30')
def test_show_tld_404(self):
@ -172,7 +180,7 @@ class TestTldNotFoundAdmin(BaseTldTest):
class TestTldInvalidIdAdmin(BaseTldTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -183,7 +191,10 @@ class TestTldInvalidIdAdmin(BaseTldTest):
@classmethod
def setup_clients(cls):
super(TestTldInvalidIdAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.TldClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.TldClient()
else:
cls.admin_client = cls.os_admin.dns_v2.TldClient()
@decorators.idempotent_id('f9ec0730-57ff-4720-8d06-e11d377c7cfc')
def test_show_tld_invalid_uuid(self):

37
designate_tempest_plugin/tests/api/v2/test_transfer_accepts.py
View File

@ -12,12 +12,14 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as logging
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
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -27,7 +29,7 @@ class BaseTransferAcceptTest(base.BaseDnsV2Test):
class TransferAcceptTest(BaseTransferAcceptTest):
credentials = ['primary', 'alt', 'admin']
credentials = ["primary", "alt", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -50,9 +52,18 @@ class TransferAcceptTest(BaseTransferAcceptTest):
cls.alt_accept_client = cls.os_alt.dns_v2.TransferAcceptClient()
# Admin clients
cls.admin_zone_client = cls.os_admin.dns_v2.ZonesClient()
cls.admin_request_client = cls.os_admin.dns_v2.TransferRequestClient()
cls.admin_accept_client = cls.os_admin.dns_v2.TransferAcceptClient()
if CONF.enforce_scope.designate:
cls.admin_zone_client = cls.os_system_admin.dns_v2.ZonesClient()
cls.admin_request_client = (cls.os_system_admin.dns_v2.
TransferRequestClient())
cls.admin_accept_client = (cls.os_system_admin.dns_v2.
TransferAcceptClient())
else:
cls.admin_zone_client = cls.os_admin.dns_v2.ZonesClient()
cls.admin_request_client = (cls.os_admin.dns_v2.
TransferRequestClient())
cls.admin_accept_client = (cls.os_admin.dns_v2.
TransferAcceptClient())
@decorators.idempotent_id('1c6baf97-a83e-4d2e-a5d8-9d37fb7808f3')
def test_create_transfer_accept(self):
@ -60,7 +71,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
_, zone = self.prm_zone_client.create_zone(wait_until='ACTIVE')
self.addCleanup(
self.wait_zone_delete, self.admin_zone_client, zone['id'],
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Create a zone transfer_request')
@ -89,7 +100,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
_, zone = self.prm_zone_client.create_zone(wait_until='ACTIVE')
self.addCleanup(
self.wait_zone_delete, self.admin_zone_client, zone['id'],
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Create a zone transfer_request')
@ -125,7 +136,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
zone = self.prm_zone_client.create_zone(wait_until='ACTIVE')[1]
self.addCleanup(
self.wait_zone_delete, self.admin_zone_client, zone['id'],
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Create a Primary zone transfer_request')
@ -170,7 +181,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
zone = self.prm_zone_client.create_zone(wait_until='ACTIVE')[1]
self.addCleanup(
self.wait_zone_delete, self.admin_zone_client, zone['id'],
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Create a Primary zone transfer_request')
@ -203,7 +214,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
admin_client_accept_ids = [
item['id'] for item in
self.admin_accept_client.list_transfer_accept(
headers={'x-auth-all-projects': True}, params={'limit': 1000})]
headers=self.all_projects_header, params={'limit': 1000})]
for tr_id in transfer_request_ids:
self.assertIn(
tr_id, admin_client_accept_ids,
@ -217,7 +228,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
admin_client_accept_ids = [
item['id'] for item in
self.admin_accept_client.list_transfer_accept(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'status': 'COMPLETE'})]
for tr_id in transfer_request_ids:
self.assertIn(
@ -234,7 +245,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
admin_client_accept_ids = [
item['id'] for item in
self.admin_accept_client.list_transfer_accept(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'status': not_existing_status})]
self.assertEmpty(
admin_client_accept_ids,
@ -255,7 +266,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
# for a zone.
self.addCleanup(
self.wait_zone_delete, self.admin_zone_client, zone['id'],
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
ignore_errors=lib_exc.NotFound)
LOG.info('Create a zone transfer_request as primary tenant')
@ -294,7 +305,7 @@ class TransferAcceptTest(BaseTransferAcceptTest):
class TransferAcceptTestNegative(BaseTransferAcceptTest):
credentials = ['primary', 'alt', 'admin']
credentials = ["primary", "alt", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):

12
designate_tempest_plugin/tests/api/v2/test_transfer_request.py
View File

@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as logging
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
@ -19,6 +20,7 @@ from tempest.lib import exceptions as lib_exc
from designate_tempest_plugin.tests import base
from designate_tempest_plugin import data_utils as dns_data_utils
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -27,7 +29,7 @@ class BaseTransferRequestTest(base.BaseDnsV2Test):
class TransferRequestTest(BaseTransferRequestTest):
credentials = ['primary', 'alt', 'admin']
credentials = ["primary", "alt", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -39,11 +41,15 @@ class TransferRequestTest(BaseTransferRequestTest):
def setup_clients(cls):
super(TransferRequestTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = (cls.os_system_admin.dns_v2.
TransferRequestClient())
else:
cls.admin_client = cls.os_admin.dns_v2.TransferRequestClient()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.alt_zone_client = cls.os_alt.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.TransferRequestClient()
cls.alt_client = cls.os_alt.dns_v2.TransferRequestClient()
cls.admin_client = cls.os_admin.dns_v2.TransferRequestClient()
@decorators.idempotent_id('2381d489-ad84-403d-b0a2-8b77e4e966bf')
def test_create_transfer_request(self):
@ -233,7 +239,7 @@ class TransferRequestTest(BaseTransferRequestTest):
# pagination limit is only 20, we set a param limit of 1000 here.
request_ids = [
item['id'] for item in self.admin_client.list_transfer_requests(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'limit': 1000})[1]['transfer_requests']]
for request_id in [primary_transfer_request['id'],

23
designate_tempest_plugin/tests/api/v2/test_tsigkey.py
View File

@ -13,12 +13,14 @@
# under the License.
from oslo_log import log as logging
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
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -27,7 +29,7 @@ class BaseTsigkeyTest(base.BaseDnsV2Test):
class TsigkeyAdminTest(BaseTsigkeyTest):
credentials = ['primary', 'admin']
credentials = ["primary", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -38,8 +40,11 @@ class TsigkeyAdminTest(BaseTsigkeyTest):
@classmethod
def setup_clients(cls):
super(TsigkeyAdminTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.TsigkeyClient()
else:
cls.admin_client = cls.os_admin.dns_v2.TsigkeyClient()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.admin_client = cls.os_admin.dns_v2.TsigkeyClient()
@decorators.idempotent_id('e7b484e3-7ed5-4840-89d7-1e696986f8e4')
def test_create_tsigkey(self):
@ -140,7 +145,7 @@ class TsigkeyAdminTest(BaseTsigkeyTest):
class TestTsigkeyNotFoundAdmin(BaseTsigkeyTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -151,7 +156,10 @@ class TestTsigkeyNotFoundAdmin(BaseTsigkeyTest):
@classmethod
def setup_clients(cls):
super(TestTsigkeyNotFoundAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.TsigkeyClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.TsigkeyClient()
else:
cls.admin_client = cls.os_admin.dns_v2.TsigkeyClient()
@decorators.idempotent_id('824c9b49-edc5-4282-929e-467a158d23e4')
def test_show_tsigkey_404(self):
@ -183,7 +191,7 @@ class TestTsigkeyNotFoundAdmin(BaseTsigkeyTest):
class TestTsigkeyInvalidIdAdmin(BaseTsigkeyTest):
credentials = ["admin"]
credentials = ["admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -194,7 +202,10 @@ class TestTsigkeyInvalidIdAdmin(BaseTsigkeyTest):
@classmethod
def setup_clients(cls):
super(TestTsigkeyInvalidIdAdmin, cls).setup_clients()
cls.admin_client = cls.os_admin.dns_v2.TsigkeyClient()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.TsigkeyClient()
else:
cls.admin_client = cls.os_admin.dns_v2.TsigkeyClient()
@decorators.idempotent_id('2a8dfc75-9884-4b1c-8f1f-ed835d96f2fe')
def test_show_tsigkey_invalid_uuid(self):

16
designate_tempest_plugin/tests/api/v2/test_zone_tasks.py
View File

@ -16,6 +16,7 @@ from socket import gaierror
from unittest import expectedFailure
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
@ -26,6 +27,7 @@ from designate_tempest_plugin.tests import base
from designate_tempest_plugin.services.dns.query.query_client \
import SingleQueryClient
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -35,7 +37,7 @@ class BaseZonesTest(base.BaseDnsV2Test):
class ZoneTasks(BaseZonesTest):
credentials = ['primary', 'alt', 'admin']
credentials = ["primary", "alt", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -46,8 +48,11 @@ class ZoneTasks(BaseZonesTest):
@classmethod
def setup_clients(cls):
super(ZoneTasks, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.ZonesClient()
else:
cls.admin_client = cls.os_admin.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.ZonesClient()
cls.admin_client = cls.os_admin.dns_v2.ZonesClient()
cls.alt_client = cls.os_alt.dns_v2.ZonesClient()
@decorators.idempotent_id('287e2cd0-a0e7-11eb-b962-74e5f9e2a801')
@ -105,7 +110,7 @@ class ZoneTasks(BaseZonesTest):
class ZoneTasksNegative(BaseZonesTest):
credentials = ['primary', 'alt', 'admin']
credentials = ["primary", "alt", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
@ -116,8 +121,11 @@ class ZoneTasksNegative(BaseZonesTest):
@classmethod
def setup_clients(cls):
super(ZoneTasksNegative, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.ZonesClient()
else:
cls.admin_client = cls.os_admin.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.ZonesClient()
cls.admin_client = cls.os_admin.dns_v2.ZonesClient()
cls.alt_client = cls.os_alt.dns_v2.ZonesClient()
def _query_nameserver(self, nameserver, query_timeout,

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

@ -13,6 +13,7 @@
# under the License.
import uuid
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib.common.utils import data_utils
from tempest.lib import exceptions as lib_exc
@ -24,6 +25,8 @@ from designate_tempest_plugin import data_utils as dns_data_utils
from designate_tempest_plugin.tests import base
from designate_tempest_plugin.common import waiters
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -33,7 +36,7 @@ class BaseZonesTest(base.BaseDnsV2Test):
class ZonesTest(BaseZonesTest):
credentials = ['admin', 'primary']
credentials = ["admin", "system_admin", "primary"]
@classmethod
def setup_credentials(cls):
@ -44,8 +47,11 @@ class ZonesTest(BaseZonesTest):
@classmethod
def setup_clients(cls):
super(ZonesTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.pool_client = cls.os_system_admin.dns_v2.PoolClient()
else:
cls.pool_client = cls.os_admin.dns_v2.PoolClient()
cls.client = cls.os_primary.dns_v2.ZonesClient()
cls.pool_client = cls.os_admin.dns_v2.PoolClient()
@decorators.idempotent_id('9d2e20fc-e56f-4a62-9c61-9752a9ec615c')
def test_create_zones(self):
@ -192,7 +198,7 @@ class ZonesTest(BaseZonesTest):
class ZonesAdminTest(BaseZonesTest):
credentials = ['primary', 'admin', 'alt']
credentials = ["primary", "admin", "system_admin", "alt"]
@classmethod
def setup_credentials(cls):
@ -203,8 +209,11 @@ class ZonesAdminTest(BaseZonesTest):
@classmethod
def setup_clients(cls):
super(ZonesAdminTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.ZonesClient()
else:
cls.admin_client = cls.os_admin.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.ZonesClient()
cls.admin_client = cls.os_admin.dns_v2.ZonesClient()
cls.alt_client = cls.os_alt.dns_v2.ZonesClient()
@decorators.idempotent_id('f6fe8cce-8b04-11eb-a861-74e5f9e2a801')
@ -261,19 +270,22 @@ class ZonesAdminTest(BaseZonesTest):
self.alt_client, alt_zone['id'], 'ACTIVE')
LOG.info('Create zone "C" using Admin client')
admin_zone = self.admin_client.create_zone()[1]
admin_zone = self.admin_client.create_zone(
project_id="FakeProjectID")[1]
self.addCleanup(
self.wait_zone_delete, self.admin_client, admin_zone['id'])
self.wait_zone_delete, self.admin_client, admin_zone['id'],
headers=self.all_projects_header)
LOG.info('Wait till the zone is ACTIVE')
waiters.wait_for_zone_status(
self.admin_client, admin_zone['id'], 'ACTIVE')
self.admin_client, admin_zone['id'], 'ACTIVE',
headers=self.all_projects_header)
LOG.info('As admin user list all projects zones')
# Note: This is an all-projects list call, so other tests running
# in parallel will impact the list result set. Since the default
# pagination limit is only 20, we set a param limit of 1000 here.
body = self.admin_client.list_zones(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'limit': 1000})[1]['zones']
listed_zone_ids = [item['id'] for item in body]
@ -288,7 +300,7 @@ class ZonesAdminTest(BaseZonesTest):
class ZoneOwnershipTest(BaseZonesTest):
credentials = ['primary', 'alt']
credentials = ["primary", "alt"]
@classmethod
def setup_credentials(cls):

21
designate_tempest_plugin/tests/api/v2/test_zones_exports.py
View File

@ -13,11 +13,13 @@
# under the License.
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from designate_tempest_plugin.tests import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -27,7 +29,7 @@ class BaseZoneExportsTest(base.BaseDnsV2Test):
class ZonesExportTest(BaseZoneExportsTest):
credentials = ['primary', 'admin', 'alt']
credentials = ["primary", "admin", "system_admin", "alt"]
@classmethod
def setup_credentials(cls):
@ -38,11 +40,14 @@ class ZonesExportTest(BaseZoneExportsTest):
@classmethod
def setup_clients(cls):
super(ZonesExportTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.ZoneExportsClient()
else:
cls.admin_client = cls.os_admin.dns_v2.ZoneExportsClient()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.alt_zone_client = cls.os_alt.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.ZoneExportsClient()
cls.alt_client = cls.os_alt.dns_v2.ZoneExportsClient()
cls.admin_client = cls.os_admin.dns_v2.ZoneExportsClient()
@decorators.idempotent_id('2dd8a9a0-98a2-4bf6-bb51-286583b30f40')
def test_create_zone_export(self):
@ -150,7 +155,7 @@ class ZonesExportTest(BaseZoneExportsTest):
# pagination limit is only 20, we set a param limit of 1000 here.
listed_exports_ids = [
item['id'] for item in self.admin_client.list_zone_exports(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'limit': 1000})[1]['exports']]
LOG.info('Make sure that all previously created zone '
@ -185,7 +190,7 @@ class ZonesExportTest(BaseZoneExportsTest):
' expected: empty list')
self.assertEqual(
[], self.admin_client.list_zone_exports(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'status': 'ZAHLABUT'})[1]['exports'],
'Failed, filtered result is expected to be empty.')
@ -193,7 +198,7 @@ class ZonesExportTest(BaseZoneExportsTest):
' expected: empty list')
self.assertEqual(
[], self.admin_client.list_zone_exports(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'message': 'ZABABUN'})[1]['exports'],
'Failed, filtered result is expected to be empty.')
@ -201,7 +206,7 @@ class ZonesExportTest(BaseZoneExportsTest):
'a primary zone. Expected: single zone export is listed')
self.assertEqual(
1, len(self.admin_client.list_zone_exports(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'zone_id': primary_zone['id']})[1]['exports']),
'Failed, filtered result should contain a single zone '
'(primary zone export)')
@ -210,13 +215,13 @@ class ZonesExportTest(BaseZoneExportsTest):
'an alt zone expected: empty list (it was deleted)')
self.assertEqual(
[], self.admin_client.list_zone_exports(
headers={'x-auth-all-projects': True},
headers=self.all_projects_header,
params={'zone_id': alt_zone['id']})[1]['exports'],
'Failed, filtered result should be empty.')
class ZonesExportTestNegative(BaseZoneExportsTest):
credentials = ['primary', 'alt']
credentials = ["primary", "alt"]
@classmethod
def setup_credentials(cls):

17
designate_tempest_plugin/tests/api/v2/test_zones_imports.py
View File

@ -13,6 +13,7 @@
# under the License.
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
@ -21,6 +22,7 @@ from designate_tempest_plugin.common import waiters
from designate_tempest_plugin.tests import base
from designate_tempest_plugin import data_utils as dns_data_utils
CONF = config.CONF
LOG = logging.getLogger(__name__)
@ -30,7 +32,7 @@ class BaseZonesImportTest(base.BaseDnsV2Test):
class ZonesImportTest(BaseZonesImportTest):
credentials = ['primary', 'admin', 'alt']
credentials = ["primary", "admin", "system_admin", "alt"]
@classmethod
def setup_credentials(cls):
@ -41,10 +43,13 @@ class ZonesImportTest(BaseZonesImportTest):
@classmethod
def setup_clients(cls):
super(ZonesImportTest, cls).setup_clients()
if CONF.enforce_scope.designate:
cls.admin_client = cls.os_system_admin.dns_v2.ZoneImportsClient()
else:
cls.admin_client = cls.os_admin.dns_v2.ZoneImportsClient()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.ZoneImportsClient()
cls.alt_client = cls.os_alt.dns_v2.ZoneImportsClient()
cls.admin_client = cls.os_admin.dns_v2.ZoneImportsClient()
def clean_up_resources(self, zone_import_id):
zone_import = self.client.show_zone_import(zone_import_id)[1]
@ -195,15 +200,15 @@ class ZonesImportTest(BaseZonesImportTest):
'"x-auth-all-projects" HTTP header, Expected: 403 Forbidden')
self.assertRaises(
lib_exc.Forbidden, lambda: self.alt_client.list_zone_imports(
headers={'x-auth-all-projects': True}))
headers=self.all_projects_header))
LOG.info('As Admin tenant list import zones for all projects')
# Note: This is an all-projects list call, so other tests running
# in parallel will impact the list result set. Since the default
# pagination limit is only 20, we set a param limit of 1000 here.
body = self.admin_client.list_zone_imports(headers={
'x-auth-all-projects': True},
params={'limit': 1000})[1]['imports']
body = self.admin_client.list_zone_imports(
headers=self.all_projects_header,
params={'limit': 1000})[1]['imports']
LOG.info('Ensure the fetched response includes previously '
'created import ID')

2
designate_tempest_plugin/tests/base.py
View File

@ -136,6 +136,8 @@ class BaseDnsTest(test.BaseTestCase):
class BaseDnsV2Test(BaseDnsTest):
"""Base class for DNS V2 API tests."""
all_projects_header = {'X-Auth-All-Projects': True}
@classmethod
def skip_checks(cls):
super(BaseDnsV2Test, cls).skip_checks()