openstack/deb-designate
Code Issues Proposed changes

First Tempest tests for recordsets

Browse Source 
* Added decorators for parameterized tests
* Added models/clients for recordsets
* Move v2 specific clients into DesignateV2Test base class
* Added data generation for A, AAAA, CNAME, and MX recordsets
* Added test cases that do CRUD on A, AAAA, CNAME, and MX recordsets

Change-Id: I7755167b707ca66f8f2a3362579e3e976b4aa315
This commit is contained in:
Paul Glass 2015-03-24 18:24:37 +00:00
parent 4620275cdb
commit c10c853d2b
8 changed files with 405 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
functionaltests/api/v2/base.py Normal file
View File

@ -0,0 +1,65 @@
"""
Copyright 2015 Rackspace
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.exceptions import NotFound
from functionaltests.api.v2.clients.recordset_client import RecordsetClient
from functionaltests.api.v2.clients.zone_client import ZoneClient
from functionaltests.api.v2.clients.quotas_client import QuotasClient
from functionaltests.api.v2.models.quotas_model import QuotasModel
from functionaltests.common.base import BaseDesignateTest
class DesignateV2Test(BaseDesignateTest):
def __init__(self, *args, **kwargs):
super(DesignateV2Test, self).__init__(*args, **kwargs)
self.zone_client = ZoneClient(self.base_client)
self.quotas_client = QuotasClient(self.base_client)
self.recordset_client = RecordsetClient(self.base_client)
def wait_for_zone(self, zone_id):
self.wait_for_condition(lambda: self.is_zone_active(zone_id))
def wait_for_zone_404(self, zone_id):
self.wait_for_condition(lambda: self.is_zone_404(zone_id))
def is_zone_active(self, zone_id):
resp, model = self.zone_client.get_zone(zone_id)
self.assertEqual(resp.status, 200)
if model.status == 'ACTIVE':
return True
elif model.status == 'ERROR':
raise Exception("Saw ERROR status")
return False
def is_zone_404(self, zone_id):
try:
# tempest_lib rest client raises exceptions on bad status codes
resp, model = self.zone_client.get_zone(zone_id)
except NotFound:
return True
return False
def increase_quotas(self):
self.quotas_client.patch_quotas(
self.quotas_client.client.tenant_id,
QuotasModel.from_dict({
'quota': {
'zones': 9999999,
'recordset_records': 9999999,
'zone_records': 9999999,
'zone_recordsets': 9999999}}))

60
functionaltests/api/v2/clients/recordset_client.py Normal file
View File

@ -0,0 +1,60 @@
"""
Copyright 2015 Rackspace
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 functionaltests.api.v2.models.recordset_model import RecordsetModel
from functionaltests.api.v2.models.recordset_model import RecordsetListModel
class RecordsetClient(object):
def __init__(self, client):
self.client = client
@classmethod
def recordsets_uri(cls, zone_id):
return "/v2/zones/{0}/recordsets".format(zone_id)
@classmethod
def recordset_uri(cls, zone_id, recordset_id):
return "{0}/{1}".format(cls.recordsets_uri(zone_id), recordset_id)
@classmethod
def deserialize(cls, resp, body, model_type):
return resp, model_type.from_json(body)
def list_recordsets(self, zone_id, **kwargs):
resp, body = self.client.get(self.recordsets_uri(zone_id), **kwargs)
return self.deserialize(resp, body, RecordsetListModel)
def get_recordset(self, zone_id, recordset_id, **kwargs):
resp, body = self.client.get(self.recordset_uri(zone_id, recordset_id),
**kwargs)
return self.deserialize(resp, body, RecordsetModel)
def post_recordset(self, zone_id, recordset_model, **kwargs):
resp, body = self.client.post(self.recordsets_uri(zone_id),
body=recordset_model.to_json(), **kwargs)
return self.deserialize(resp, body, RecordsetModel)
def put_recordset(self, zone_id, recordset_id, recordset_model, **kwargs):
resp, body = self.client.put(self.recordset_uri(zone_id, recordset_id),
body=recordset_model.to_json(), **kwargs)
return self.deserialize(resp, body, RecordsetModel)
def delete_recordset(self, zone_id, recordset_id, **kwargs):
resp, body = self.client.delete(
self.recordset_uri(zone_id, recordset_id), **kwargs)
return self.deserialize(resp, body, RecordsetModel)

33
functionaltests/api/v2/models/recordset_model.py Normal file
View File

@ -0,0 +1,33 @@
"""
Copyright 2015 Rackspace
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 functionaltests.common.models import BaseModel
from functionaltests.common.models import CollectionModel
from functionaltests.common.models import EntityModel
class RecordsetData(BaseModel):
pass
class RecordsetModel(EntityModel):
ENTITY_NAME = 'recordset'
MODEL_TYPE = RecordsetData
class RecordsetListModel(CollectionModel):
COLLECTION_NAME = 'recordsets'
MODEL_TYPE = RecordsetData

101
functionaltests/api/v2/test_recordset.py Normal file
View File

@ -0,0 +1,101 @@
"""
Copyright 2015 Rackspace
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.exceptions import NotFound
from functionaltests.common import datagen
from functionaltests.common import utils
from functionaltests.api.v2.base import DesignateV2Test
@utils.parameterized_class
class RecordsetTest(DesignateV2Test):
def setUp(self):
super(RecordsetTest, self).setUp()
self.increase_quotas()
resp, self.zone = self.zone_client.post_zone(
datagen.random_zone_data())
self.wait_for_zone(self.zone.id)
def wait_for_recordset(self, zone_id, recordset_id):
self.wait_for_condition(
lambda: self.is_recordset_active(zone_id, recordset_id))
def wait_for_404(self, zone_id, recordset_id):
self.wait_for_condition(
lambda: self.is_recordset_404(zone_id, recordset_id))
def is_recordset_active(self, zone_id, recordset_id):
resp, model = self.recordset_client.get_recordset(
zone_id, recordset_id)
self.assertEqual(resp.status, 200)
if model.status == 'ACTIVE':
return True
elif model.status == 'ERROR':
raise Exception("Saw ERROR status")
return False
def is_recordset_404(self, zone_id, recordset_id):
try:
self.recordset_client.get_recordset(zone_id, recordset_id)
except NotFound:
return True
return False
def test_list_recordsets(self):
resp, model = self.recordset_client.list_recordsets(self.zone.id)
self.assertEqual(resp.status, 200)
@utils.parameterized({
'A': dict(
make_recordset=lambda z: datagen.random_a_recordset(z.name)),
'AAAA': dict(
make_recordset=lambda z: datagen.random_aaaa_recordset(z.name)),
'CNAME': dict(
make_recordset=lambda z: datagen.random_cname_recordset(z.name)),
'MX': dict(
make_recordset=lambda z: datagen.random_mx_recordset(z.name)),
})
def test_crud_recordset(self, make_recordset):
post_model = make_recordset(self.zone)
resp, post_resp_model = self.recordset_client.post_recordset(
self.zone.id, post_model)
self.assertEqual(resp.status, 202, "on post response")
self.assertEqual(post_resp_model.status, "PENDING")
self.assertEqual(post_resp_model.name, post_model.name)
self.assertEqual(post_resp_model.records, post_model.records)
self.assertEqual(post_resp_model.ttl, post_model.ttl)
recordset_id = post_resp_model.id
self.wait_for_recordset(self.zone.id, recordset_id)
put_model = make_recordset(self.zone)
del put_model.name # don't try to update the name
resp, put_resp_model = self.recordset_client.put_recordset(
self.zone.id, recordset_id, put_model)
self.assertEqual(resp.status, 202, "on put response")
self.assertEqual(put_resp_model.status, "PENDING")
self.assertEqual(put_resp_model.name, post_model.name)
self.assertEqual(put_resp_model.records, put_model.records)
self.assertEqual(put_resp_model.ttl, put_model.ttl)
self.wait_for_recordset(self.zone.id, recordset_id)
resp, delete_resp_model = self.recordset_client.delete_recordset(
self.zone.id, recordset_id)
self.assertEqual(resp.status, 202, "on delete response")
self.wait_for_404(self.zone.id, recordset_id)

39
functionaltests/api/v2/test_zone.py
View File

@ -14,46 +14,25 @@ See the License for the specific language governing permissions and
limitations under the License.
"""
from functionaltests.api.v2.clients.zone_client import ZoneClient
from functionaltests.api.v2.clients.quotas_client import QuotasClient
from functionaltests.api.v2.models.quotas_model import QuotasModel
from functionaltests.common import datagen
from functionaltests.common.base import BaseDesignateTest
from functionaltests.api.v2.base import DesignateV2Test
class ZoneTest(BaseDesignateTest):
def __init__(self, *args, **kwargs):
super(ZoneTest, self).__init__(*args, **kwargs)
self.client = ZoneClient(self.base_client)
self.quotas_client = QuotasClient(self.base_client)
class ZoneTest(DesignateV2Test):
def setUp(self):
super(ZoneTest, self).setUp()
self.quotas_client.patch_quotas(
self.quotas_client.client.tenant_id,
QuotasModel.from_dict({
'quota': {
'zones': 9999999,
'recordset_records': 9999999,
'zone_records': 9999999,
'zone_recordsets': 9999999}}))
def wait_for_zone(self, zone_id):
self.wait_for_condition(lambda: self.is_zone_active(zone_id))
def wait_for_404(self, zone_id):
self.wait_for_condition(lambda: self.is_zone_404(zone_id))
self.increase_quotas()
def _create_zone(self, zone_model):
resp, model = self.client.post_zone(zone_model)
resp, model = self.zone_client.post_zone(zone_model)
self.assertEqual(resp.status, 202)
self.wait_for_zone(model.id)
return resp, model
def test_list_zones(self):
self._create_zone(datagen.random_zone_data())
resp, model = self.client.list_zones()
resp, model = self.zone_client.list_zones()
self.assertEqual(resp.status, 200)
self.assertGreater(len(model.zones), 0)
@ -66,12 +45,12 @@ class ZoneTest(BaseDesignateTest):
patch_model = datagen.random_zone_data()
del patch_model.name # don't try to override the zone name
resp, new_model = self.client.patch_zone(old_model.id,
resp, new_model = self.zone_client.patch_zone(old_model.id,
patch_model)
self.assertEqual(resp.status, 202)
self.wait_for_zone(new_model.id)
resp, model = self.client.get_zone(new_model.id)
resp, model = self.zone_client.get_zone(new_model.id)
self.assertEqual(resp.status, 200)
self.assertEqual(new_model.id, old_model.id)
self.assertEqual(new_model.name, old_model.name)
@ -80,6 +59,6 @@ class ZoneTest(BaseDesignateTest):
def test_delete_zone(self):
resp, model = self._create_zone(datagen.random_zone_data())
resp, model = self.client.delete_zone(model.id)
resp, model = self.zone_client.delete_zone(model.id)
self.assertEqual(resp.status, 202)
self.wait_for_404(model.id)
self.wait_for_zone_404(model.id)

18
functionaltests/common/base.py
View File

@ -17,7 +17,6 @@ limitations under the License.
import time
import tempest_lib.base
from tempest_lib.exceptions import NotFound
from functionaltests.common.client import DesignateClient
@ -35,20 +34,3 @@ class BaseDesignateTest(tempest_lib.base.BaseTestCase):
return
time.sleep(interval)
raise Exception("Timed out after {0} seconds".format(timeout))
def is_zone_active(self, zone_id):
resp, model = self.client.get_zone(zone_id)
self.assertEqual(resp.status, 200)
if model.status == 'ACTIVE':
return True
elif model.status == 'ERROR':
raise Exception("Saw ERROR status")
return False
def is_zone_404(self, zone_id):
try:
# tempest_lib rest client raises exceptions on bad status codes
resp, model = self.client.get_zone(zone_id)
except NotFound:
return True
return False

54
functionaltests/common/datagen.py
View File

@ -17,12 +17,20 @@ limitations under the License.
import random
from functionaltests.api.v2.models.zone_model import ZoneModel
from functionaltests.api.v2.models.recordset_model import RecordsetModel
def random_ip():
return ".".join(str(random.randrange(0, 256)) for _ in range(4))
def random_ipv6():
def hexes(n):
return "".join(random.choice("1234567890abcdef") for _ in range(n))
result = ":".join(hexes(4) for _ in range(8))
return result.replace("0000", "0")
def random_string(prefix='rand', n=8, suffix=''):
"""Return a string containing random digits
@ -52,3 +60,49 @@ def random_zone_data(name=None, email=None, ttl=None, description=None):
'email': email,
'ttl': random.randint(1200, 8400),
'description': description})
def random_recordset_data(record_type, zone_name, name=None, records=None,
ttl=None):
"""Generate random recordset data, with optional overrides
:return: A RecordsetModel
"""
if name is None:
name = random_string(prefix=record_type, suffix='.' + zone_name)
if records is None:
records = [random_ip()]
if ttl is None:
ttl = random.randint(1200, 8400)
return RecordsetModel.from_dict({
'type': record_type,
'name': name,
'records': records,
'ttl': ttl})
def random_a_recordset(zone_name, ip=None, **kwargs):
if ip is None:
ip = random_ip()
return random_recordset_data('A', zone_name, records=[ip], **kwargs)
def random_aaaa_recordset(zone_name, ip=None, **kwargs):
if ip is None:
ip = random_ipv6()
return random_recordset_data('AAAA', zone_name, records=[ip], **kwargs)
def random_cname_recordset(zone_name, cname=None, **kwargs):
if cname is None:
cname = zone_name
return random_recordset_data('CNAME', zone_name, records=[cname], **kwargs)
def random_mx_recordset(zone_name, pref=None, host=None, **kwargs):
if pref is None:
pref = str(random.randint(0, 65535))
if host is None:
host = random_string(prefix='mail', suffix='.' + zone_name)
data = "{0} {1}".format(pref, host)
return random_recordset_data('MX', zone_name, records=[data], **kwargs)

83
functionaltests/common/utils.py Normal file
View File

@ -0,0 +1,83 @@
"""
Copyright 2015 Rackspace
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import functools
import types
def def_method(f, *args, **kwargs):
@functools.wraps(f)
def new_method(self):
return f(self, *args, **kwargs)
return new_method
def parameterized_class(cls):
"""A class decorator for running parameterized test cases.
Mark your class with @parameterized_class.
Mark your test cases with @parameterized.
"""
test_functions = {
k: v for k, v in vars(cls).iteritems() if k.startswith('test')
}
for name, f in test_functions.iteritems():
if not hasattr(f, '_test_data'):
continue
# remove the original test function from the class
delattr(cls, name)
# add a new test function to the class for each entry in f._test_data
for tag, args in f._test_data.iteritems():
new_name = "{0}_{1}".format(f.__name__, tag)
if hasattr(cls, new_name):
raise Exception(
"Parameterized test case '{0}.{1}' created from '{0}.{2}' "
"already exists".format(cls.__name__, new_name, name))
# Using `def new_method(self): f(self, **args)` is not sufficient
# (all new_methods use the same args value due to late binding).
# Instead, use this factory function.
new_method = def_method(f, **args)
# To add a method to a class, available for all instances:
# MyClass.method = types.MethodType(f, None, MyClass)
setattr(cls, new_name, types.MethodType(new_method, None, cls))
return cls
def parameterized(data):
"""A function decorator for parameterized test cases.
Example:
@parameterized({
'zero': dict(val=0),
'one': dict(val=1),
})
def test_val(self, val):
self.assertEqual(self.get_val(), val)
The above will generate two test cases:
`test_val_zero` which runs with val=0
`test_val_one` which runs with val=1
:param data: A dictionary that looks like {tag: {arg1: val1, ...}}
"""
def wrapped(f):
f._test_data = data
return f
return wrapped