openstack/designate-tempest-plugin
Code Issues Proposed changes

Merge "Test cases for: "FloatingIPs PTR" resords."

Browse Source
This commit is contained in:
Zuul 2021-05-21 06:33:53 +00:00 committed by Gerrit Code Review
commit 8b08b0c0bc
4 changed files with 229 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
designate_tempest_plugin/clients.py
View File

@ -51,6 +51,7 @@ from designate_tempest_plugin.services.dns.v2.json.service_client \
import SevriceClient
from designate_tempest_plugin.services.dns.v2.json.designate_limit_client \
import DesignateLimitClient
from designate_tempest_plugin.services.dns.v2.json.ptr_client import PtrClient
CONF = config.CONF
@ -99,6 +100,7 @@ class ManagerV2(clients.Manager):
self.tsigkey_client = TsigkeyClient(**params)
self.service_client = SevriceClient(**params)
self.designate_limit_client = DesignateLimitClient(**params)
self.ptr_client = PtrClient(**params)
self.query_client = QueryClient(
nameservers=CONF.dns.nameservers,

31
designate_tempest_plugin/services/dns/json/base.py
View File

@ -87,17 +87,25 @@ class DnsClientBase(rest_client.RestClient):
expected_code=expected_code, read_code=int(read_code),
)
def get_uri(self, resource_name, uuid=None, params=None):
def get_uri(self, resource_name, uuid=None, params=None,
uuid_prefix_char=None):
"""Get URI for a specific resource or object.
:param resource_name: The name of the REST resource, e.g., 'zones'.
:param uuid: The unique identifier of an object in UUID format.
:param params: A Python dict that represents the query paramaters to
include in the request URI.
:param uuid_prefix_char: applies to override hardcoded ('/')
prefix UUID character. This parameter enables to set required
by API character, for example ":" instead of "/".
:returns: Relative URI for the resource or object.
"""
uri_pattern = '{pref}/{res}{uuid}{params}'
uuid = '/%s' % uuid if uuid else ''
if uuid_prefix_char:
uuid = uuid_prefix_char + '%s' % uuid if uuid else ''
else:
uuid = '/%s' % uuid if uuid else ''
params = '?%s' % urllib.urlencode(params) if params else ''
return uri_pattern.format(pref=self.uri_prefix,
@ -141,7 +149,7 @@ class DnsClientBase(rest_client.RestClient):
return resp, self.deserialize(resp, body)
def _show_request(self, resource, uuid, headers=None, params=None,
extra_headers=False):
extra_headers=False, uuid_prefix_char=None):
"""Gets a specific object of the specified type.
:param resource: The name of the REST resource, e.g., 'zones'.
:param uuid: Unique identifier of the object in UUID format.
@ -152,9 +160,13 @@ class DnsClientBase(rest_client.RestClient):
method are to be used but additional
headers are needed in the request
pass them in as a dict.
:param uuid_prefix_char: applies to override hardcoded ('/')
prefix UUID character. This parameter enables to set required
by API character, for example ":" instead of "/".
:returns: Serialized object as a dictionary.
"""
uri = self.get_uri(resource, uuid=uuid, params=params)
uri = self.get_uri(resource, uuid=uuid, params=params,
uuid_prefix_char=uuid_prefix_char)
resp, body = self.get(
uri, headers=headers, extra_headers=extra_headers)
@ -199,7 +211,7 @@ class DnsClientBase(rest_client.RestClient):
return resp, self.deserialize(resp, body)
def _update_request(self, resource, uuid, data, params=None, headers=None,
extra_headers=False):
extra_headers=False, uuid_prefix_char=None):
"""Updates the specified object using PATCH request.
:param resource: The name of the REST resource, e.g., 'zones'
:param uuid: Unique identifier of the object in UUID format.
@ -214,13 +226,18 @@ class DnsClientBase(rest_client.RestClient):
method are to be used but additional
headers are needed in the request
pass them in as a dict.
:param uuid_prefix_char: applies to override hardcoded ('/')
prefix UUID character. This parameter enables to set required
by API character, for example ":" instead of "/".
:returns: Serialized object as a dictionary.
"""
body = self.serialize(data)
uri = self.get_uri(resource, uuid=uuid, params=params)
uri = self.get_uri(
resource, uuid=uuid, params=params,
uuid_prefix_char=uuid_prefix_char)
resp, body = self.patch(uri, body=body,
headers=headers, extra_headers=True)
headers=headers, extra_headers=extra_headers)
self.expected_success(self.UPDATE_STATUS_CODES, resp.status)

85
designate_tempest_plugin/services/dns/v2/json/ptr_client.py Normal file
View File

@ -0,0 +1,85 @@
# Copyright 2021 Red Hat.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from tempest.lib.common.utils import data_utils
from designate_tempest_plugin import data_utils as dns_data_utils
from designate_tempest_plugin.services.dns.v2.json import base
from tempest import config
CONF = config.CONF
class PtrClient(base.DnsClientV2Base):
@base.handle_errors
def set_ptr_record(self, floatingip_id, ptr_name=None,
ttl=None, description=None, headers=None):
"""Set a PTR record for the given FloatingIP
:param floatingip_id: valid UUID of floating IP to be used.
:param ptr_name PTR record name or random if not provided.
:param ttl TTL or random valid value if not provided.
:param description Description or random if not provided.
:param headers (dict): The headers to use for the request.
:return: created PTR dictionary.
"""
ptr = {
'ptrdname': ptr_name or dns_data_utils.rand_zone_name(),
'ttl': ttl or dns_data_utils.rand_ttl(),
'description': description or data_utils.rand_name(
'test-ptr')}
return self._update_request(
resource='reverse/floatingips/{}'.format(CONF.identity.region),
uuid=floatingip_id, data=ptr, headers=headers,
uuid_prefix_char=':')[1]
@base.handle_errors
def show_ptr_record(self, floatingip_id, headers=None):
"""Show PTR record for the given FloatingIP
:param floatingip_id: valid UUID of floating IP to show.
:param headers (dict): The headers to use for the request.
:return: Shown PTR dictionary.
"""
return self._show_request(
resource='reverse/floatingips/{}'.format(CONF.identity.region),
uuid=floatingip_id, headers=headers, uuid_prefix_char=':')[1]
@base.handle_errors
def list_ptr_records(self, headers=None):
"""List PTR records for the given FloatingIP
:param headers (dict): The headers to use for the request.
:return: List of PTR records.
"""
return self._list_request(
'reverse/floatingips', headers=headers)[1]['floatingips']
@base.handle_errors
def unset_ptr_record(self, floatingip_id, headers=None):
"""Unset the PTR record for a given FloatingIP
:param floatingip_id: valid UUID of floating IP to unset.
:param headers (dict): The headers to use for the request.
:return: Tuple (Response, Body)
"""
data = {"ptrdname": None}
resp, body = self._update_request(
resource='reverse/floatingips/{}'.format(CONF.identity.region),
uuid=floatingip_id, data=data, headers=headers,
uuid_prefix_char=':')
# Unset PTR should Return a HTTP 202
self.expected_success(202, resp.status)
return resp, body

118
designate_tempest_plugin/tests/api/v2/test_ptrs.py Normal file
View File

@ -0,0 +1,118 @@
# Copyright 2021 Red Hat.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
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
import tempest.test
LOG = logging.getLogger(__name__)
CONF = config.CONF
class BasePtrTest(base.BaseDnsV2Test):
excluded_keys = ['created_at', 'updated_at', 'version', 'links',
'status', 'action']
class DesignatePtrRecord(BasePtrTest, tempest.test.BaseTestCase):
credentials = ['primary']
@classmethod
def setup_credentials(cls):
# Do not create network resources for these test.
cls.set_network_resources()
super(DesignatePtrRecord, cls).setup_credentials()
@classmethod
def setup_clients(cls):
super(DesignatePtrRecord, cls).setup_clients()
cls.primary_ptr_client = cls.os_primary.ptr_client
cls.primary_floating_ip_client = cls.os_primary.floating_ips_client
def _set_ptr(self):
fip_id = self.primary_floating_ip_client.create_floatingip(
floating_network_id=CONF.network.public_network_id)[
'floatingip']['id']
ptr = self.primary_ptr_client.set_ptr_record(fip_id)
self.assertEqual('CREATE', ptr['action'])
self.assertEqual('PENDING', ptr['status'])
return fip_id, ptr
@decorators.idempotent_id('2fb9d6ea-871d-11eb-9f9a-74e5f9e2a801')
def test_set_floatingip_ptr(self):
self._set_ptr()
@decorators.idempotent_id('9179325a-87d0-11eb-9f9a-74e5f9e2a801')
def test_show_floatingip_ptr(self):
fip_id, ptr = self._set_ptr()
show_ptr = self.primary_ptr_client.show_ptr_record(
floatingip_id=fip_id)
self.assertExpected(ptr, show_ptr, self.excluded_keys)
@decorators.idempotent_id('9187a9c6-87d4-11eb-9f9a-74e5f9e2a801')
def test_list_floatingip_ptr_records(self):
number_of_ptr_records = 3
created_ptr_ids = []
for _ in range(number_of_ptr_records):
fip_id, ptr = self._set_ptr()
created_ptr_ids.append(ptr['id'])
received_ptr_ids = sorted(
[item['id'] for item in
self.primary_ptr_client.list_ptr_records()])
self.assertEqual(
sorted(created_ptr_ids), received_ptr_ids,
'Failed - received PTR IDs: {} are not as'
' expected: {}'.format(created_ptr_ids, received_ptr_ids))
@decorators.idempotent_id('499b5a7e-87e1-11eb-b412-74e5f9e2a801')
def test_unset_floatingip_ptr(self):
fip_id, ptr = self._set_ptr()
self.primary_ptr_client.unset_ptr_record(fip_id)
class DesignatePtrRecordNegative(BasePtrTest, tempest.test.BaseTestCase):
credentials = ['primary']
@classmethod
def setup_credentials(cls):
# Do not create network resources for these test.
cls.set_network_resources()
super(DesignatePtrRecordNegative, cls).setup_credentials()
@classmethod
def setup_clients(cls):
super(DesignatePtrRecordNegative, cls).setup_clients()
cls.primary_ptr_client = cls.os_primary.ptr_client
cls.primary_floating_ip_client = cls.os_primary.floating_ips_client
def _set_ptr(self, ptr_name=None, ttl=None, description=None,
headers=None):
fip_id = self.primary_floating_ip_client.create_floatingip(
floating_network_id=CONF.network.public_network_id)[
'floatingip']['id']
ptr = self.primary_ptr_client.set_ptr_record(
fip_id, ptr_name=ptr_name, ttl=ttl, description=description,
headers=headers)
self.assertEqual('CREATE', ptr['action'])
self.assertEqual('PENDING', ptr['status'])
return fip_id, ptr
def test_set_floatingip_ptr_invalid_ttl(self):
LOG.info('Try to set PTR record using invalid TTL value')
with self.assertRaisesDns(lib_exc.BadRequest, 'invalid_object', 400):
self._set_ptr(ttl=-10)