Zone Imports - refactoring and new test cases

1) Refactoring:
   - Replace Python underscore variable with indexing.
   - Use constants.
   - Move negative test cases to dedicated "Negative" class.
   - Client's "create_zone_import" had to be improved.
   - Using provided "wait_until" in the existing tests.
2) "test_create_zone_import_invalid_file_data"
   Try to create a zone import using invalid zone file data.
   Expected: zone gets into the ERROR status.
3) "test_zone_cannot_be_update_by_import"
   Try to update an existing zone using Zone import.
   Expected: zone import gets into the ERROR status.
4) "test_create_zone_import_invalid_content_type" to test
   "415 unsupported_content_type" error response.

Change-Id: Ifce7e6ed910c43aacc566cdf692eec45779811e4
This commit is contained in:
Arkady Shtempler
2022-03-07 12:31:49 +02:00
parent 1852668517
commit 3b5d149cf3
3 changed files with 96 additions and 17 deletions

View File

@@ -21,15 +21,16 @@ class ZoneImportsClient(base.DnsClientV2Base):
@base.handle_errors
def create_zone_import(self, zonefile_data=None,
params=None, wait_until=None):
wait_until=None, headers=None):
"""Create a zone import.
:param zonefile_data: A tuple that represents zone data.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param wait_until: If not None, a waiter for appropriate status
will be activated.
:param headers (dict): The headers to use for the request.
:return: Serialized imported zone as a dictionary.
"""
headers = {'Content-Type': 'text/dns'}
if not headers:
headers = {'Content-Type': 'text/dns'}
zone_data = zonefile_data or dns_data_utils.rand_zonefile_data()
resp, body = self._create_request(
'zones/tasks/imports', zone_data, headers=headers)

View File

@@ -373,3 +373,83 @@ class ZonesImportTest(BaseZonesImportTest):
self.check_list_IDs_RBAC_enforcement(
'ZoneImportsClient', 'list_zone_imports', expected_allowed,
[zone_import['id']], headers=self.all_projects_header)
class ZonesImportTestNegative(BaseZonesImportTest):
credentials = ["primary", "admin", "system_admin"]
@classmethod
def setup_credentials(cls):
# Do not create network resources for these test.
cls.set_network_resources()
super(ZonesImportTestNegative, cls).setup_credentials()
@classmethod
def setup_clients(cls):
super(ZonesImportTestNegative, cls).setup_clients()
cls.zone_client = cls.os_primary.dns_v2.ZonesClient()
cls.client = cls.os_primary.dns_v2.ZoneImportsClient()
def _clean_up_resources(self, zone_import_id):
zone_import = self.client.show_zone_import(zone_import_id)[1]
if zone_import['zone_id']: # A zone was actually created.
waiters.wait_for_zone_import_status(
self.client, zone_import_id, const.COMPLETE)
self.client.delete_zone_import(zone_import['id'])
self.wait_zone_delete(self.zone_client, zone_import['zone_id'])
else: # Import has failed and zone wasn't created.
self.client.delete_zone_import(zone_import['id'])
@decorators.idempotent_id('31eaf25a-9532-11eb-a55d-74e5f9e2a801')
def test_create_zone_import_invalid_ttl(self):
LOG.info('Try to create a zone import using invalid TTL value')
zone_name = dns_data_utils.rand_zone_name(
name="create_zone_import_invalid_ttl", suffix=self.tld_name)
zone_data = dns_data_utils.rand_zonefile_data(name=zone_name,
ttl='zahlabut')
zone_import = self.client.create_zone_import(
zonefile_data=zone_data, wait_until=const.ERROR)[1]
self.addCleanup(self._clean_up_resources, zone_import['id'])
@decorators.idempotent_id('31eaf25a-9532-11eb-a55d-74e5f9e2a801')
def test_create_zone_import_invalid_name(self):
LOG.info('Try to create a zone import using invalid name')
zone_import = self.client.create_zone_import(
zonefile_data=dns_data_utils.rand_zonefile_data(
name='@@@'), wait_until=const.ERROR)[1]
self.addCleanup(self._clean_up_resources, zone_import['id'])
@decorators.idempotent_id('8fd744d2-9dff-11ec-9fb6-201e8823901f')
def test_create_zone_import_invalid_file_data(self):
LOG.info('Try to create a zone import using random generated'
' import file data')
zone_file_data = dns_data_utils.rand_string(size=100)
zone_import = self.client.create_zone_import(zone_file_data)[1]
self.addCleanup(self.client.delete_zone_import, zone_import['id'])
waiters.wait_for_zone_import_status(
self.client, zone_import['id'], const.ERROR)
@decorators.idempotent_id('4fb9494e-9e23-11ec-8378-201e8823901f')
def test_zone_cannot_be_update_by_import(self):
LOG.info('Create a Zone named: "...zone_to_update..."')
zone_name = dns_data_utils.rand_zone_name(
name='zone_to_update', suffix=self.tld_name)
zone = self.zone_client.create_zone(
name=zone_name, wait_until=const.ACTIVE)[1]
self.addCleanup(self.wait_zone_delete, self.zone_client, zone['id'])
LOG.info('Use zone import to update an existing zone, expected: zone'
' import gets into the ERROR status ')
zone_import_data = dns_data_utils.rand_zonefile_data(name=zone_name)
zone_import = self.client.create_zone_import(zone_import_data)[1]
waiters.wait_for_zone_import_status(
self.client, zone_import['id'], const.ERROR)
self.addCleanup(self._clean_up_resources, zone_import['id'])
@decorators.idempotent_id('5fa8016e-6ed1-11ec-9bd7-201e8823901f')
def test_create_zone_import_invalid_content_type(self):
LOG.info('Try to create a zone import using: "Content-Type:Zahlabut"'
' HTTP header in POST request')
with self.assertRaisesDns(
lib_exc.InvalidContentType, 'unsupported_content_type', 415):
self.client.create_zone_import(
headers={'Content-Type': 'Zahlabut'})

View File

@@ -14,6 +14,7 @@
from oslo_log import log as logging
from tempest.lib import decorators
from designate_tempest_plugin.common import constants as const
from designate_tempest_plugin.common import waiters
from designate_tempest_plugin import data_utils as dns_data_utils
from designate_tempest_plugin.tests.api.v2.test_zones_imports import \
@@ -41,29 +42,26 @@ class ZonesImportTest(BaseZonesImportTest):
zonefile = dns_data_utils.rand_zonefile_data(name=zone_name)
LOG.info('Import zone %r', zone_name)
_, zone_import = self.client.create_zone_import(zonefile)
zone_import = self.client.create_zone_import(
zonefile, wait_until=const.COMPLETE)[1]
self.addCleanup(self.client.delete_zone_import, zone_import['id'])
LOG.info('Wait for the zone import to COMPLETE')
waiters.wait_for_zone_import_status(self.client, zone_import['id'],
"COMPLETE")
LOG.info('Check the zone import looks good')
_, zone_import = self.client.show_zone_import(zone_import['id'])
zone_import = self.client.show_zone_import(zone_import['id'])[1]
self.addCleanup(self.wait_zone_delete,
self.zones_client,
zone_import['zone_id'])
self.assertEqual('COMPLETE', zone_import['status'])
self.assertEqual(const.COMPLETE, zone_import['status'])
self.assertIsNotNone(zone_import['zone_id'])
self.assertIsNotNone(zone_import['links'].get('zone'))
LOG.info('Wait for the imported zone to go to ACTIVE')
waiters.wait_for_zone_status(self.zones_client, zone_import['zone_id'],
"ACTIVE")
waiters.wait_for_zone_status(
self.zones_client, zone_import['zone_id'], const.ACTIVE)
LOG.info('Check the imported zone looks good')
_, zone = self.zones_client.show_zone(zone_import['zone_id'])
self.assertEqual('NONE', zone['action'])
self.assertEqual('ACTIVE', zone['status'])
zone = self.zones_client.show_zone(zone_import['zone_id'])[1]
self.assertEqual(const.NONE, zone['action'])
self.assertEqual(const.ACTIVE, zone['status'])
self.assertEqual(zone_name, zone['name'])