x/poppy
Code Issues Proposed changes

Add API tests for custom & SAN cert services

Browse Source 
This patch also replaces some of the UUID names to be a little more
friendly to the eyes & DRYs up some code to be friendly to the brain.

Change-Id: I07a94b0e946465054e3c4400f72d84d80369949f
This commit is contained in:
Malini Kamalambal
2015-04-16 14:06:46 -04:00
parent 7b68ef7498
commit b84d263122
7 changed files with 392 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
tests/api/base.py
View File

@@ -79,7 +79,7 @@ class TestBase(fixtures.BaseTestFixture):
def generate_random_string(self, prefix='API-Tests', length=12):
"""Generates a random string of given prefix & length"""
random_string = ''.join(random.choice(
string.ascii_uppercase + string.ascii_uppercase + string.digits)
string.ascii_lowercase + string.digits)
for _ in range(length))
random_string = prefix + random_string
return random_string
@@ -108,6 +108,48 @@ class TestBase(fixtures.BaseTestFixture):
return flavor_id
def setup_service(self, service_name, domain_list, origin_list,
caching_list=[], restrictions_list=[], flavor_id=None,
log_delivery=False):
resp = self.client.create_service(
service_name=service_name,
domain_list=domain_list,
origin_list=origin_list,
caching_list=caching_list,
restrictions_list=restrictions_list,
flavor_id=flavor_id,
log_delivery=log_delivery)
self.assertEqual(resp.status_code, 202)
self.service_location = resp.headers['location']
self.client.wait_for_service_status(
location=self.service_location,
status='DEPLOYED',
abort_on_status='FAILED',
retry_interval=self.test_config.status_check_retry_interval,
retry_timeout=self.test_config.status_check_retry_timeout)
return resp
def assert_patch_service_details(self, actual_response, expected_response):
self.assertEqual(actual_response['name'],
expected_response['name'])
self.assertEqual(sorted(actual_response['origins']),
sorted(expected_response['origins']))
self.assertEqual(sorted(actual_response['caching']),
sorted(expected_response['caching']))
self.assertEqual(sorted(actual_response['restrictions']),
sorted(expected_response['restrictions']))
self.assertEqual(actual_response['flavor_id'],
expected_response['flavor_id'])
for item in actual_response['domains']:
if (('certificate' in item) and
(item['certificate'] == 'shared')):
item['domain'] = item['domain'].split('.')[0]
self.assertEqual(sorted(actual_response['domains']),
sorted(expected_response['domains']))
@classmethod
def tearDownClass(cls):
"""Deletes the added resources."""

58
tests/api/services/data_create_service_ssl_domain.json
View File

@@ -21,5 +21,63 @@
"referrer": "www.mywebsite.com",
"request_url": "/click_me"}]}
]
},
"domain_san_ssl": {
"name": "domain_san_ssl",
"domain_list": [{"domain": "sansslwebsite.com",
"protocol": "https",
"certificate": "san"}],
"origin_list": [{"origin": "mywebsite1.com",
"port": 443,
"ssl": true}],
"caching_list": [{"name": "default", "ttl": 3600},
{"name": "home",
"ttl": 1200,
"rules": [{"name" : "index",
"request_url" : "/index.htm"}]}],
"restrictions_list": [
{"name": "test",
"rules": [{"name": "only me",
"referrer": "www.mywebsite.com"}]}
]
},
"domain_custom_ssl": {
"name": "domain_san_ssl",
"domain_list": [{"domain": "customsslwebsite.com",
"protocol": "https",
"certificate": "custom"}],
"origin_list": [{"origin": "mywebsite1.com",
"port": 443,
"ssl": true}],
"caching_list": [{"name": "default", "ttl": 3600},
{"name": "home",
"ttl": 1200,
"rules": [{"name" : "index",
"request_url" : "/index.htm"}]}],
"restrictions_list": [
{"name": "test",
"rules": [{"name": "only me",
"referrer": "www.mywebsite.com"}]}
]
},
"domain_shared_ssl": {
"name": "domain_shared_ssl",
"domain_list": [{"domain": "sharedsslwebsite",
"protocol": "https",
"certificate": "shared"}
],
"origin_list": [{"origin": "mywebsite1.com",
"port": 443,
"ssl": true}],
"caching_list": [{"name": "default", "ttl": 3600},
{"name": "home",
"ttl": 1200,
"rules": [{"name" : "index",
"request_url" : "/index.htm"}]}],
"restrictions_list": [
{"name": "test",
"rules": [{"name": "only me",
"referrer": "www.mywebsite.com"}]}
]
}
}

32
tests/api/services/data_patch_service_ssl_domain.json Normal file
View File

@@ -0,0 +1,32 @@
{
"add_shared_ssl_domain": [
{"op": "add",
"path": "/domains/-",
"value": {"domain": "newshared", "protocol": "https", "certificate": "shared"}}
],
"add_custom_ssl_domain": [
{"op": "add",
"path": "/domains/-",
"value": {"domain": "newcustom.com", "protocol": "https", "certificate": "custom"}}
],
"add_san_ssl_domain": [
{"op": "add",
"path": "/domains/-",
"value": {"domain": "newsan.com", "protocol": "https", "certificate": "custom"}}
],
"replace_with_custom_ssl_domain": [
{"op": "replace",
"path": "/domains/0",
"value": {"domain": "replacewithcustom.com", "protocol": "https", "certificate": "custom"}}
],
"replace_with_san_ssl_domain": [
{"op": "replace",
"path": "/domains/0",
"value": {"domain": "replacewithsan.com", "protocol": "https", "certificate": "san"}}
],
"replace_with_shared_ssl_domain": [
{"op": "replace",
"path": "/domains/0",
"value": {"domain": "replacewithshared", "protocol": "https", "certificate": "shared"}}
]
}

99
tests/api/services/test_services.py
View File

@@ -37,7 +37,8 @@ class TestCreateService(providers.TestProviderBase):
def setUp(self):
super(TestCreateService, self).setUp()
self.service_url = ''
self.service_name = str(uuid.uuid1())
self.service_name = self.generate_random_string(
prefix='api-test-service')
self.flavor_id = self.test_flavor
@attrib.attr('smoke')
@@ -46,7 +47,8 @@ class TestCreateService(providers.TestProviderBase):
domain_list = test_data['domain_list']
for item in domain_list:
item['domain'] = str(uuid.uuid1()) + '.com'
item['domain'] = self.generate_random_string(
prefix='api-test-domain') + '.com'
origin_list = test_data['origin_list']
caching_list = test_data['caching_list']
log_delivery = test_data.get('log_delivery')
@@ -114,46 +116,6 @@ class TestCreateService(providers.TestProviderBase):
msg='Caching List Not Correct for {0} service name {1}'.
format(provider, self.service_name))
@attrib.attr('smoke')
@ddt.file_data('data_create_service_ssl_domain.json')
def test_create_service_ssl_domain_positive(self, test_data):
if self.test_config.run_ssl_tests is False:
self.skipTest(
'SSL tests are currently disabled in configuration')
domain_list = test_data['domain_list']
for item in domain_list:
item['domain'] = str(uuid.uuid1())
origin_list = test_data['origin_list']
caching_list = test_data['caching_list']
flavor_id = self.flavor_id
resp = self.client.create_service(service_name=self.service_name,
domain_list=domain_list,
origin_list=origin_list,
caching_list=caching_list,
flavor_id=flavor_id)
self.assertEqual(resp.status_code, 202)
self.assertEqual(resp.text, '')
self.service_url = resp.headers['location']
resp = self.client.get_service(location=self.service_url)
self.assertEqual(resp.status_code, 200)
self.client.wait_for_service_status(
location=self.service_url,
status='deployed',
abort_on_status='failed',
retry_interval=self.test_config.status_check_retry_interval,
retry_timeout=self.test_config.status_check_retry_timeout)
resp = self.client.get_service(location=self.service_url)
self.assertEqual(resp.status_code, 200)
body = resp.json()
status = body['status']
self.assertEqual(status, 'deployed')
@attrib.attr('smoke')
@ddt.file_data('data_create_service_negative.json')
def test_create_service_negative(self, test_data):
@@ -189,7 +151,8 @@ class TestCreateService(providers.TestProviderBase):
service_name = test_data['name']
domain_list = test_data['domain_list']
for item in domain_list:
item['domain'] = str(uuid.uuid1()) + '.com'
item['domain'] = self.generate_random_string(
prefix='api-test-xss') + '.com'
origin_list = test_data['origin_list']
caching_list = test_data['caching_list']
@@ -274,10 +237,11 @@ class TestListServices(base.TestBase):
def _create_test_service(self):
service_name = str(uuid.uuid1())
self.domain_list = [{"domain": str(uuid.uuid1()) + '.com'}]
self.domain_list = [{"domain": self.generate_random_string(
prefix='api-test-domain') + '.com'}]
self.origin_list = [{"origin": str(uuid.uuid1()) + '.com',
"port": 80, "ssl": False}]
self.origin_list = [{"origin": self.generate_random_string(
prefix='api-test-origin') + '.com', "port": 80, "ssl": False}]
self.caching_list = [{"name": "default", "ttl": 3600},
{"name": "home", "ttl": 1200,
@@ -393,15 +357,17 @@ class TestServiceActions(base.TestBase):
def setUp(self):
super(TestServiceActions, self).setUp()
self.service_name = str(uuid.uuid1())
self.service_name = self.generate_random_string(prefix='API-Test-')
self.flavor_id = self.test_flavor
domain = str(uuid.uuid1()) + u'.com'
domain = self.generate_random_string(
prefix='api-test-domain') + u'.com'
self.domain_list = [
{"domain": domain, "protocol": "http"}
]
origin = str(uuid.uuid1()) + u'.com'
origin = self.generate_random_string(
prefix='api-test-origin') + u'.com'
self.origin_list = [
{
u"origin": origin,
@@ -544,11 +510,11 @@ class TestServicePatch(base.TestBase):
def setUp(self):
super(TestServicePatch, self).setUp()
self.service_name = str(uuid.uuid1())
self.service_name = self.generate_random_string(prefix='api-test')
self.flavor_id = self.test_flavor
self.log_delivery = {"enabled": False}
domain = str(uuid.uuid1()) + '.com'
domain = self.generate_random_string(prefix='api-test-domain') + '.com'
self.domain_list = [
{
"domain": domain,
@@ -556,7 +522,7 @@ class TestServicePatch(base.TestBase):
}
]
origin = str(uuid.uuid1()) + '.com'
origin = self.generate_random_string(prefix='api-test-origin') + '.com'
self.origin_list = [
{
"origin": origin,
@@ -633,32 +599,13 @@ class TestServicePatch(base.TestBase):
self.assertEqual(resp.status_code, 200)
self.assertEqual(body['status'], 'deployed')
def _assert_service_details(self, actual_response, expected_response):
self.assertEqual(actual_response['name'],
expected_response['name'])
self.assertEqual(sorted(actual_response['origins']),
sorted(expected_response['origins']))
self.assertEqual(sorted(actual_response['caching']),
sorted(expected_response['caching']))
self.assertEqual(sorted(actual_response['restrictions']),
sorted(expected_response['restrictions']))
self.assertEqual(actual_response['flavor_id'],
expected_response['flavor_id'])
for item in actual_response['domains']:
if (('certificate' in item) and
(item['certificate'] == 'shared')):
item['domain'] = item['domain'].split('.')[0]
self.assertEqual(sorted(actual_response['domains']),
sorted(expected_response['domains']))
def _replace_domain(self, domain):
if ('protocol' in domain):
if domain['protocol'] == 'https':
if (domain['certificate'] == u'shared'):
return str(uuid.uuid1())
return self.generate_random_string(prefix='api-test-ssl')
return str(uuid.uuid1()) + '.com'
return self.generate_random_string(prefix='api-test-ssl') + '.com'
@ddt.file_data('data_patch_service.json')
def test_patch_service(self, test_data):
@@ -693,7 +640,7 @@ class TestServicePatch(base.TestBase):
body = resp.json()
self.assertEqual(body['status'], 'deployed')
self._assert_service_details(body, expected_service_details)
self.assert_patch_service_details(body, expected_service_details)
@ddt.file_data('data_patch_service_negative.json')
def test_patch_service_HTTP_400(self, test_data):
@@ -709,7 +656,7 @@ class TestServicePatch(base.TestBase):
body = resp.json()
self.assertEqual(body['status'], 'deployed')
self._assert_service_details(body, self.original_service_details)
self.assert_patch_service_details(body, self.original_service_details)
def test_patch_service_claim_relinquish_domain(self):
newdomain = str(uuid.uuid4()) + ".com"
@@ -841,7 +788,7 @@ class TestDefaultServiceFields(providers.TestProviderBase):
def setUp(self):
super(TestDefaultServiceFields, self).setUp()
self.service_url = ''
self.service_name = str(uuid.uuid1())
self.service_name = self.generate_random_string(prefix='api-test')
self.flavor_id = self.test_flavor
@ddt.file_data('data_default_service_values.json')

231
tests/api/services/test_ssl_services.py Normal file
View File

@@ -0,0 +1,231 @@
# coding= utf-8
# Copyright (c) 2015 Rackspace, Inc.
#
# 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 ddt
import jsonpatch
from nose.plugins import attrib
from tests.api import base
@ddt.ddt
class TestCreateSSLService(base.TestBase):
"""Tests for Create Service."""
def setUp(self):
super(TestCreateSSLService, self).setUp()
if self.test_config.run_ssl_tests is False:
self.skipTest(
'SSL tests are currently disabled in configuration')
self.service_url = ''
self.service_name = self.generate_random_string(
prefix='api-test-service')
self.flavor_id = self.test_flavor
@attrib.attr('smoke')
@ddt.file_data('data_create_service_ssl_domain.json')
def test_create_service_ssl_domain_positive(self, test_data):
domain_list = test_data['domain_list']
for item in domain_list:
if item['certificate'] == 'shared':
item['domain'] = self.generate_random_string(
prefix='shared-ssl')
else:
item['domain'] = self.generate_random_string(
prefix='ssl-domain') + '.com'
origin_list = test_data['origin_list']
caching_list = test_data['caching_list']
flavor_id = self.flavor_id
resp = self.client.create_service(service_name=self.service_name,
domain_list=domain_list,
origin_list=origin_list,
caching_list=caching_list,
flavor_id=flavor_id)
self.assertEqual(resp.status_code, 202)
self.assertEqual(resp.text, '')
self.service_url = resp.headers['location']
resp = self.client.get_service(location=self.service_url)
self.assertEqual(resp.status_code, 200)
self.client.wait_for_service_status(
location=self.service_url,
status='deployed',
abort_on_status='failed',
retry_interval=self.test_config.status_check_retry_interval,
retry_timeout=self.test_config.status_check_retry_timeout)
resp = self.client.get_service(location=self.service_url)
self.assertEqual(resp.status_code, 200)
body = resp.json()
status = body['status']
self.assertEqual(status, 'deployed')
def tearDown(self):
if self.service_url != '':
self.client.delete_service(location=self.service_url)
if self.test_config.generate_flavors:
self.client.delete_flavor(flavor_id=self.flavor_id)
super(TestCreateSSLService, self).tearDown()
@ddt.ddt
class TestPatchSSLService(base.TestBase):
"""Tests for Patch SSL Service."""
def setUp(self):
super(TestPatchSSLService, self).setUp()
if self.test_config.run_ssl_tests is False:
self.skipTest(
'SSL tests are currently disabled in configuration')
self.service_name = self.generate_random_string(prefix='api-test')
self.flavor_id = self.test_flavor
self.log_delivery = {"enabled": False}
domain = self.generate_random_string(prefix='api-test-domain')
self.domain_list = [
{
"domain": domain,
"protocol": "https",
"certificate": "shared"
}
]
origin = self.generate_random_string(prefix='api-test-origin') + '.com'
self.origin_list = [
{
"origin": origin,
"port": 443,
"ssl": True,
"rules": [
{
"name": "default",
"request_url": "/*"
}
]
}
]
self.caching_list = [
{
"name": "default",
"ttl": 3600,
"rules": [
{
"name": "default",
"request_url": "/*"
}
]
},
{
"name": "home",
"ttl": 1200,
"rules": [
{
"name": "index",
"request_url": "/index.htm"
}
]
}
]
self.restrictions_list = [
{"name": "website only",
"rules": [{"name": "mywebsite.com",
"referrer": "www.mywebsite.com",
"request_url": "/*"
}]}]
resp = self.setup_service(
service_name=self.service_name,
domain_list=self.domain_list,
origin_list=self.origin_list,
caching_list=self.caching_list,
restrictions_list=self.restrictions_list,
flavor_id=self.flavor_id,
log_delivery=self.log_delivery)
self.service_url = resp.headers["location"]
self.original_service_details = {
"name": self.service_name,
"domains": self.domain_list,
"origins": self.origin_list,
"caching": self.caching_list,
"restrictions": self.restrictions_list,
"flavor_id": self.flavor_id,
"log_delivery": self.log_delivery}
def _replace_domain(self, domain):
if ('protocol' in domain):
if domain['protocol'] == 'https':
if (domain['certificate'] == u'shared'):
return self.generate_random_string(prefix='api-test-ssl')
return self.generate_random_string(prefix='api-test-ssl') + '.com'
@ddt.file_data('data_patch_service_ssl_domain.json')
@ddt.file_data('failed.json')
def test_patch_ssl_domain(self, test_data):
for item in test_data:
if ('domain' in item['path']) and ('value' in item):
if isinstance(item['value'], (list)):
item['value'][0]['domain'] = self._replace_domain(
domain=item['value'][0])
else:
item['value']['domain'] = self._replace_domain(
domain=item['value'])
patch = jsonpatch.JsonPatch(test_data)
expected_service_details = patch.apply(self.original_service_details)
resp = self.client.patch_service(location=self.service_url,
request_body=test_data)
self.assertEqual(resp.status_code, 202)
self.client.wait_for_service_status(
location=self.service_url,
status='deployed',
abort_on_status='failed',
retry_interval=self.test_config.status_check_retry_interval,
retry_timeout=self.test_config.status_check_retry_timeout)
resp = self.client.get_service(location=self.service_url)
body = resp.json()
self.assertEqual(body['status'], 'deployed')
self.assert_patch_service_details(body, expected_service_details)
def tearDown(self):
if self.service_url != '':
self.client.delete_service(location=self.service_url)
if self.test_config.generate_flavors:
self.client.delete_flavor(flavor_id=self.flavor_id)
super(TestPatchSSLService, self).tearDown()

2
tests/api/utils/client.py
View File

@@ -239,7 +239,7 @@ class PoppyClient(client.AutoMarshallingHTTPClient):
service = self.get_service(location=location)
body = service.json()
current_status = body['status']
if (current_status == status):
if (current_status.lower() == status.lower()):
return
if abort_on_status is not None:

6
tests/endtoend/base.py
View File

@@ -111,14 +111,16 @@ class TestBase(fixtures.BaseTestFixture):
return cname_rec
def setup_service(self, service_name, domain_list, origin_list,
caching_list=[], restrictions_list=[], flavor_id=None):
caching_list=[], restrictions_list=[], flavor_id=None,
log_delivery=False):
resp = self.poppy_client.create_service(
service_name=service_name,
domain_list=domain_list,
origin_list=origin_list,
caching_list=caching_list,
restrictions_list=restrictions_list,
flavor_id=flavor_id)
flavor_id=flavor_id,
log_delivery=log_delivery)
self.assertEqual(resp.status_code, 202)
self.service_location = resp.headers['location']