7c2ad79c30
JSON Schema use validator (draft3 or draft4) to validate the given schema against given instance. By default it use draft4 validator which is current one. If new validator will be released in future and that become default then, there might be some compatibility issues between draft4 and new version. For example from draft3 to draft4, 'ip-version' format has been changed to 'ipv4'. This commits explicitly pass the validator while doing the schema validation. Also adds unit tests to check the current version of validator. If any new validator will be available in future, then we can use that by doing proper modification in schema files if needed. Change-Id: Ie11e3ba60926f247a82670b7da37d7890d16280c
1052 lines
36 KiB
Python
1052 lines
36 KiB
Python
# Copyright 2013 IBM Corp.
|
|
#
|
|
# 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 json
|
|
|
|
import httplib2
|
|
import jsonschema
|
|
from oslotest import mockpatch
|
|
import six
|
|
|
|
from tempest_lib.common import rest_client
|
|
from tempest_lib import exceptions
|
|
from tempest_lib.tests import base
|
|
from tempest_lib.tests import fake_auth_provider
|
|
from tempest_lib.tests import fake_http
|
|
|
|
|
|
class BaseRestClientTestClass(base.TestCase):
|
|
|
|
url = 'fake_endpoint'
|
|
|
|
def setUp(self):
|
|
super(BaseRestClientTestClass, self).setUp()
|
|
self.fake_auth_provider = fake_auth_provider.FakeAuthProvider()
|
|
self.rest_client = rest_client.RestClient(
|
|
self.fake_auth_provider, None, None)
|
|
self.stubs.Set(httplib2.Http, 'request', self.fake_http.request)
|
|
self.useFixture(mockpatch.PatchObject(self.rest_client,
|
|
'_log_request'))
|
|
|
|
|
|
class TestRestClientHTTPMethods(BaseRestClientTestClass):
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2()
|
|
super(TestRestClientHTTPMethods, self).setUp()
|
|
self.useFixture(mockpatch.PatchObject(self.rest_client,
|
|
'_error_checker'))
|
|
|
|
def test_post(self):
|
|
__, return_dict = self.rest_client.post(self.url, {}, {})
|
|
self.assertEqual('POST', return_dict['method'])
|
|
|
|
def test_get(self):
|
|
__, return_dict = self.rest_client.get(self.url)
|
|
self.assertEqual('GET', return_dict['method'])
|
|
|
|
def test_delete(self):
|
|
__, return_dict = self.rest_client.delete(self.url)
|
|
self.assertEqual('DELETE', return_dict['method'])
|
|
|
|
def test_patch(self):
|
|
__, return_dict = self.rest_client.patch(self.url, {}, {})
|
|
self.assertEqual('PATCH', return_dict['method'])
|
|
|
|
def test_put(self):
|
|
__, return_dict = self.rest_client.put(self.url, {}, {})
|
|
self.assertEqual('PUT', return_dict['method'])
|
|
|
|
def test_head(self):
|
|
self.useFixture(mockpatch.PatchObject(self.rest_client,
|
|
'response_checker'))
|
|
__, return_dict = self.rest_client.head(self.url)
|
|
self.assertEqual('HEAD', return_dict['method'])
|
|
|
|
def test_copy(self):
|
|
__, return_dict = self.rest_client.copy(self.url)
|
|
self.assertEqual('COPY', return_dict['method'])
|
|
|
|
|
|
class TestRestClientNotFoundHandling(BaseRestClientTestClass):
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2(404)
|
|
super(TestRestClientNotFoundHandling, self).setUp()
|
|
|
|
def test_post(self):
|
|
self.assertRaises(exceptions.NotFound, self.rest_client.post,
|
|
self.url, {}, {})
|
|
|
|
|
|
class TestRestClientHeadersJSON(TestRestClientHTTPMethods):
|
|
TYPE = "json"
|
|
|
|
def _verify_headers(self, resp):
|
|
self.assertEqual(self.rest_client._get_type(), self.TYPE)
|
|
resp = dict((k.lower(), v) for k, v in six.iteritems(resp))
|
|
self.assertEqual(self.header_value, resp['accept'])
|
|
self.assertEqual(self.header_value, resp['content-type'])
|
|
|
|
def setUp(self):
|
|
super(TestRestClientHeadersJSON, self).setUp()
|
|
self.rest_client.TYPE = self.TYPE
|
|
self.header_value = 'application/%s' % self.rest_client._get_type()
|
|
|
|
def test_post(self):
|
|
resp, __ = self.rest_client.post(self.url, {})
|
|
self._verify_headers(resp)
|
|
|
|
def test_get(self):
|
|
resp, __ = self.rest_client.get(self.url)
|
|
self._verify_headers(resp)
|
|
|
|
def test_delete(self):
|
|
resp, __ = self.rest_client.delete(self.url)
|
|
self._verify_headers(resp)
|
|
|
|
def test_patch(self):
|
|
resp, __ = self.rest_client.patch(self.url, {})
|
|
self._verify_headers(resp)
|
|
|
|
def test_put(self):
|
|
resp, __ = self.rest_client.put(self.url, {})
|
|
self._verify_headers(resp)
|
|
|
|
def test_head(self):
|
|
self.useFixture(mockpatch.PatchObject(self.rest_client,
|
|
'response_checker'))
|
|
resp, __ = self.rest_client.head(self.url)
|
|
self._verify_headers(resp)
|
|
|
|
def test_copy(self):
|
|
resp, __ = self.rest_client.copy(self.url)
|
|
self._verify_headers(resp)
|
|
|
|
|
|
class TestRestClientUpdateHeaders(BaseRestClientTestClass):
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2()
|
|
super(TestRestClientUpdateHeaders, self).setUp()
|
|
self.useFixture(mockpatch.PatchObject(self.rest_client,
|
|
'_error_checker'))
|
|
self.headers = {'X-Configuration-Session': 'session_id'}
|
|
|
|
def test_post_update_headers(self):
|
|
__, return_dict = self.rest_client.post(self.url, {},
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
def test_get_update_headers(self):
|
|
__, return_dict = self.rest_client.get(self.url,
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
def test_delete_update_headers(self):
|
|
__, return_dict = self.rest_client.delete(self.url,
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
def test_patch_update_headers(self):
|
|
__, return_dict = self.rest_client.patch(self.url, {},
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
def test_put_update_headers(self):
|
|
__, return_dict = self.rest_client.put(self.url, {},
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
def test_head_update_headers(self):
|
|
self.useFixture(mockpatch.PatchObject(self.rest_client,
|
|
'response_checker'))
|
|
|
|
__, return_dict = self.rest_client.head(self.url,
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
def test_copy_update_headers(self):
|
|
__, return_dict = self.rest_client.copy(self.url,
|
|
extra_headers=True,
|
|
headers=self.headers)
|
|
|
|
self.assertDictContainsSubset(
|
|
{'X-Configuration-Session': 'session_id',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'},
|
|
return_dict['headers']
|
|
)
|
|
|
|
|
|
class TestRestClientParseRespJSON(BaseRestClientTestClass):
|
|
TYPE = "json"
|
|
|
|
keys = ["fake_key1", "fake_key2"]
|
|
values = ["fake_value1", "fake_value2"]
|
|
item_expected = dict((key, value) for (key, value) in zip(keys, values))
|
|
list_expected = {"body_list": [
|
|
{keys[0]: values[0]},
|
|
{keys[1]: values[1]},
|
|
]}
|
|
dict_expected = {"body_dict": {
|
|
keys[0]: values[0],
|
|
keys[1]: values[1],
|
|
}}
|
|
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2()
|
|
super(TestRestClientParseRespJSON, self).setUp()
|
|
self.rest_client.TYPE = self.TYPE
|
|
|
|
def test_parse_resp_body_item(self):
|
|
body = self.rest_client._parse_resp(json.dumps(self.item_expected))
|
|
self.assertEqual(self.item_expected, body)
|
|
|
|
def test_parse_resp_body_list(self):
|
|
body = self.rest_client._parse_resp(json.dumps(self.list_expected))
|
|
self.assertEqual(self.list_expected["body_list"], body)
|
|
|
|
def test_parse_resp_body_dict(self):
|
|
body = self.rest_client._parse_resp(json.dumps(self.dict_expected))
|
|
self.assertEqual(self.dict_expected["body_dict"], body)
|
|
|
|
def test_parse_resp_two_top_keys(self):
|
|
dict_two_keys = self.dict_expected.copy()
|
|
dict_two_keys.update({"second_key": ""})
|
|
body = self.rest_client._parse_resp(json.dumps(dict_two_keys))
|
|
self.assertEqual(dict_two_keys, body)
|
|
|
|
def test_parse_resp_one_top_key_without_list_or_dict(self):
|
|
data = {"one_top_key": "not_list_or_dict_value"}
|
|
body = self.rest_client._parse_resp(json.dumps(data))
|
|
self.assertEqual(data, body)
|
|
|
|
|
|
class TestRestClientErrorCheckerJSON(base.TestCase):
|
|
c_type = "application/json"
|
|
|
|
def set_data(self, r_code, enc=None, r_body=None, absolute_limit=True):
|
|
if enc is None:
|
|
enc = self.c_type
|
|
resp_dict = {'status': r_code, 'content-type': enc}
|
|
resp_body = {'resp_body': 'fake_resp_body'}
|
|
|
|
if absolute_limit is False:
|
|
resp_dict.update({'retry-after': 120})
|
|
resp_body.update({'overLimit': {'message': 'fake_message'}})
|
|
resp = httplib2.Response(resp_dict)
|
|
data = {
|
|
"method": "fake_method",
|
|
"url": "fake_url",
|
|
"headers": "fake_headers",
|
|
"body": "fake_body",
|
|
"resp": resp,
|
|
"resp_body": json.dumps(resp_body)
|
|
}
|
|
if r_body is not None:
|
|
data.update({"resp_body": r_body})
|
|
return data
|
|
|
|
def setUp(self):
|
|
super(TestRestClientErrorCheckerJSON, self).setUp()
|
|
self.rest_client = rest_client.RestClient(
|
|
fake_auth_provider.FakeAuthProvider(), None, None)
|
|
|
|
def test_response_less_than_400(self):
|
|
self.rest_client._error_checker(**self.set_data("399"))
|
|
|
|
def test_response_400(self):
|
|
self.assertRaises(exceptions.BadRequest,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("400"))
|
|
|
|
def test_response_401(self):
|
|
self.assertRaises(exceptions.Unauthorized,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("401"))
|
|
|
|
def test_response_403(self):
|
|
self.assertRaises(exceptions.Forbidden,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("403"))
|
|
|
|
def test_response_404(self):
|
|
self.assertRaises(exceptions.NotFound,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("404"))
|
|
|
|
def test_response_409(self):
|
|
self.assertRaises(exceptions.Conflict,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("409"))
|
|
|
|
def test_response_413(self):
|
|
self.assertRaises(exceptions.OverLimit,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("413"))
|
|
|
|
def test_response_413_without_absolute_limit(self):
|
|
self.assertRaises(exceptions.RateLimitExceeded,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("413", absolute_limit=False))
|
|
|
|
def test_response_415(self):
|
|
self.assertRaises(exceptions.InvalidContentType,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("415"))
|
|
|
|
def test_response_422(self):
|
|
self.assertRaises(exceptions.UnprocessableEntity,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("422"))
|
|
|
|
def test_response_500_with_text(self):
|
|
# _parse_resp is expected to return 'str'
|
|
self.assertRaises(exceptions.ServerFault,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("500"))
|
|
|
|
def test_response_501_with_text(self):
|
|
self.assertRaises(exceptions.NotImplemented,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("501"))
|
|
|
|
def test_response_400_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self.assertRaises(exceptions.BadRequest,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("400", r_body=r_body))
|
|
|
|
if self.c_type == 'application/json':
|
|
expected = {"err": "fake_resp_body"}
|
|
else:
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_401_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self.assertRaises(exceptions.Unauthorized,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("401", r_body=r_body))
|
|
|
|
if self.c_type == 'application/json':
|
|
expected = {"err": "fake_resp_body"}
|
|
else:
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_403_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self.assertRaises(exceptions.Forbidden,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("403", r_body=r_body))
|
|
|
|
if self.c_type == 'application/json':
|
|
expected = {"err": "fake_resp_body"}
|
|
else:
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_404_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self.assertRaises(exceptions.NotFound,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("404", r_body=r_body))
|
|
|
|
if self.c_type == 'application/json':
|
|
expected = {"err": "fake_resp_body"}
|
|
else:
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_404_with_invalid_dict(self):
|
|
r_body = '{"foo": "bar"]'
|
|
e = self.assertRaises(exceptions.NotFound,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("404", r_body=r_body))
|
|
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_409_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self.assertRaises(exceptions.Conflict,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("409", r_body=r_body))
|
|
|
|
if self.c_type == 'application/json':
|
|
expected = {"err": "fake_resp_body"}
|
|
else:
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_500_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self.assertRaises(exceptions.ServerFault,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("500", r_body=r_body))
|
|
|
|
if self.c_type == 'application/json':
|
|
expected = {"err": "fake_resp_body"}
|
|
else:
|
|
expected = r_body
|
|
self.assertEqual(expected, e.resp_body)
|
|
|
|
def test_response_501_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
self.assertRaises(exceptions.NotImplemented,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("501", r_body=r_body))
|
|
|
|
def test_response_bigger_than_400(self):
|
|
# Any response code, that bigger than 400, and not in
|
|
# (401, 403, 404, 409, 413, 422, 500, 501)
|
|
self.assertRaises(exceptions.UnexpectedResponseCode,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("402"))
|
|
|
|
|
|
class TestRestClientErrorCheckerTEXT(TestRestClientErrorCheckerJSON):
|
|
c_type = "text/plain"
|
|
|
|
def test_fake_content_type(self):
|
|
# This test is required only in one exemplar
|
|
# Any response code, that bigger than 400, and not in
|
|
# (401, 403, 404, 409, 413, 422, 500, 501)
|
|
self.assertRaises(exceptions.InvalidContentType,
|
|
self.rest_client._error_checker,
|
|
**self.set_data("405", enc="fake_enc"))
|
|
|
|
def test_response_413_without_absolute_limit(self):
|
|
# Skip this test because rest_client cannot get overLimit message
|
|
# from text body.
|
|
pass
|
|
|
|
|
|
class TestRestClientUtils(BaseRestClientTestClass):
|
|
|
|
def _is_resource_deleted(self, resource_id):
|
|
if not isinstance(self.retry_pass, int):
|
|
return False
|
|
if self.retry_count >= self.retry_pass:
|
|
return True
|
|
self.retry_count = self.retry_count + 1
|
|
return False
|
|
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2()
|
|
super(TestRestClientUtils, self).setUp()
|
|
self.retry_count = 0
|
|
self.retry_pass = None
|
|
self.original_deleted_method = self.rest_client.is_resource_deleted
|
|
self.rest_client.is_resource_deleted = self._is_resource_deleted
|
|
|
|
def test_wait_for_resource_deletion(self):
|
|
self.retry_pass = 2
|
|
# Ensure timeout long enough for loop execution to hit retry count
|
|
self.rest_client.build_timeout = 500
|
|
sleep_mock = self.patch('time.sleep')
|
|
self.rest_client.wait_for_resource_deletion('1234')
|
|
self.assertEqual(len(sleep_mock.mock_calls), 2)
|
|
|
|
def test_wait_for_resource_deletion_not_deleted(self):
|
|
self.patch('time.sleep')
|
|
# Set timeout to be very quick to force exception faster
|
|
self.rest_client.build_timeout = 1
|
|
self.assertRaises(exceptions.TimeoutException,
|
|
self.rest_client.wait_for_resource_deletion,
|
|
'1234')
|
|
|
|
def test_wait_for_deletion_with_unimplemented_deleted_method(self):
|
|
self.rest_client.is_resource_deleted = self.original_deleted_method
|
|
self.assertRaises(NotImplementedError,
|
|
self.rest_client.wait_for_resource_deletion,
|
|
'1234')
|
|
|
|
def test_get_versions(self):
|
|
self.rest_client._parse_resp = lambda x: [{'id': 'v1'}, {'id': 'v2'}]
|
|
actual_resp, actual_versions = self.rest_client.get_versions()
|
|
self.assertEqual(['v1', 'v2'], list(actual_versions))
|
|
|
|
def test__str__(self):
|
|
def get_token():
|
|
return "deadbeef"
|
|
|
|
self.fake_auth_provider.get_token = get_token
|
|
self.assertIsNotNone(str(self.rest_client))
|
|
|
|
|
|
class TestProperties(BaseRestClientTestClass):
|
|
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2()
|
|
super(TestProperties, self).setUp()
|
|
creds_dict = {
|
|
'username': 'test-user',
|
|
'user_id': 'test-user_id',
|
|
'tenant_name': 'test-tenant_name',
|
|
'tenant_id': 'test-tenant_id',
|
|
'password': 'test-password'
|
|
}
|
|
self.rest_client = rest_client.RestClient(
|
|
fake_auth_provider.FakeAuthProvider(creds_dict=creds_dict),
|
|
None, None)
|
|
|
|
def test_properties(self):
|
|
self.assertEqual('test-user', self.rest_client.user)
|
|
self.assertEqual('test-user_id', self.rest_client.user_id)
|
|
self.assertEqual('test-tenant_name', self.rest_client.tenant_name)
|
|
self.assertEqual('test-tenant_id', self.rest_client.tenant_id)
|
|
self.assertEqual('test-password', self.rest_client.password)
|
|
|
|
self.rest_client.api_version = 'v1'
|
|
expected = {'api_version': 'v1',
|
|
'endpoint_type': 'publicURL',
|
|
'region': None,
|
|
'service': None,
|
|
'skip_path': True}
|
|
self.rest_client.skip_path()
|
|
self.assertEqual(expected, self.rest_client.filters)
|
|
|
|
self.rest_client.reset_path()
|
|
self.rest_client.api_version = 'v1'
|
|
expected = {'api_version': 'v1',
|
|
'endpoint_type': 'publicURL',
|
|
'region': None,
|
|
'service': None}
|
|
self.assertEqual(expected, self.rest_client.filters)
|
|
|
|
|
|
class TestExpectedSuccess(BaseRestClientTestClass):
|
|
|
|
def setUp(self):
|
|
self.fake_http = fake_http.fake_httplib2()
|
|
super(TestExpectedSuccess, self).setUp()
|
|
|
|
def test_expected_succes_int_match(self):
|
|
expected_code = 202
|
|
read_code = 202
|
|
resp = self.rest_client.expected_success(expected_code, read_code)
|
|
# Assert None resp on success
|
|
self.assertFalse(resp)
|
|
|
|
def test_expected_succes_int_no_match(self):
|
|
expected_code = 204
|
|
read_code = 202
|
|
self.assertRaises(exceptions.InvalidHttpSuccessCode,
|
|
self.rest_client.expected_success,
|
|
expected_code, read_code)
|
|
|
|
def test_expected_succes_list_match(self):
|
|
expected_code = [202, 204]
|
|
read_code = 202
|
|
resp = self.rest_client.expected_success(expected_code, read_code)
|
|
# Assert None resp on success
|
|
self.assertFalse(resp)
|
|
|
|
def test_expected_succes_list_no_match(self):
|
|
expected_code = [202, 204]
|
|
read_code = 200
|
|
self.assertRaises(exceptions.InvalidHttpSuccessCode,
|
|
self.rest_client.expected_success,
|
|
expected_code, read_code)
|
|
|
|
def test_non_success_expected_int(self):
|
|
expected_code = 404
|
|
read_code = 202
|
|
self.assertRaises(AssertionError, self.rest_client.expected_success,
|
|
expected_code, read_code)
|
|
|
|
def test_non_success_expected_list(self):
|
|
expected_code = [404, 202]
|
|
read_code = 202
|
|
self.assertRaises(AssertionError, self.rest_client.expected_success,
|
|
expected_code, read_code)
|
|
|
|
|
|
class TestResponseBody(base.TestCase):
|
|
|
|
def test_str(self):
|
|
response = {'status': 200}
|
|
body = {'key1': 'value1'}
|
|
actual = rest_client.ResponseBody(response, body)
|
|
self.assertEqual("response: %s\nBody: %s" % (response, body),
|
|
str(actual))
|
|
|
|
|
|
class TestResponseBodyData(base.TestCase):
|
|
|
|
def test_str(self):
|
|
response = {'status': 200}
|
|
data = 'data1'
|
|
actual = rest_client.ResponseBodyData(response, data)
|
|
self.assertEqual("response: %s\nBody: %s" % (response, data),
|
|
str(actual))
|
|
|
|
|
|
class TestResponseBodyList(base.TestCase):
|
|
|
|
def test_str(self):
|
|
response = {'status': 200}
|
|
body = ['value1', 'value2', 'value3']
|
|
actual = rest_client.ResponseBodyList(response, body)
|
|
self.assertEqual("response: %s\nBody: %s" % (response, body),
|
|
str(actual))
|
|
|
|
|
|
class TestJSONSchemaValidationBase(base.TestCase):
|
|
|
|
class Response(dict):
|
|
|
|
def __getattr__(self, attr):
|
|
return self[attr]
|
|
|
|
def __setattr__(self, attr, value):
|
|
self[attr] = value
|
|
|
|
def setUp(self):
|
|
super(TestJSONSchemaValidationBase, self).setUp()
|
|
self.fake_auth_provider = fake_auth_provider.FakeAuthProvider()
|
|
self.rest_client = rest_client.RestClient(
|
|
self.fake_auth_provider, None, None)
|
|
|
|
def _test_validate_pass(self, schema, resp_body, status=200):
|
|
resp = self.Response()
|
|
resp.status = status
|
|
self.rest_client.validate_response(schema, resp, resp_body)
|
|
|
|
def _test_validate_fail(self, schema, resp_body, status=200,
|
|
error_msg="HTTP response body is invalid"):
|
|
resp = self.Response()
|
|
resp.status = status
|
|
ex = self.assertRaises(exceptions.InvalidHTTPResponseBody,
|
|
self.rest_client.validate_response,
|
|
schema, resp, resp_body)
|
|
self.assertIn(error_msg, ex._error_string)
|
|
|
|
|
|
class TestRestClientJSONSchemaValidation(TestJSONSchemaValidationBase):
|
|
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'integer',
|
|
},
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
|
|
def test_validate_pass(self):
|
|
body = {'foo': 12}
|
|
self._test_validate_pass(self.schema, body)
|
|
|
|
def test_validate_not_http_success_code(self):
|
|
schema = {
|
|
'status_code': [200]
|
|
}
|
|
body = {}
|
|
self._test_validate_pass(schema, body, status=400)
|
|
|
|
def test_validate_multiple_allowed_type(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': ['integer', 'string'],
|
|
},
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 12}
|
|
self._test_validate_pass(schema, body)
|
|
body = {'foo': '12'}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
def test_validate_enable_additional_property_pass(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {'type': 'integer'}
|
|
},
|
|
'additionalProperties': True,
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 12, 'foo2': 'foo2value'}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
def test_validate_disable_additional_property_pass(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {'type': 'integer'}
|
|
},
|
|
'additionalProperties': False,
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 12}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
def test_validate_disable_additional_property_fail(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {'type': 'integer'}
|
|
},
|
|
'additionalProperties': False,
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 12, 'foo2': 'foo2value'}
|
|
self._test_validate_fail(schema, body)
|
|
|
|
def test_validate_wrong_status_code(self):
|
|
schema = {
|
|
'status_code': [202]
|
|
}
|
|
body = {}
|
|
resp = self.Response()
|
|
resp.status = 200
|
|
ex = self.assertRaises(exceptions.InvalidHttpSuccessCode,
|
|
self.rest_client.validate_response,
|
|
schema, resp, body)
|
|
self.assertIn("Unexpected http success status code", ex._error_string)
|
|
|
|
def test_validate_wrong_attribute_type(self):
|
|
body = {'foo': 1.2}
|
|
self._test_validate_fail(self.schema, body)
|
|
|
|
def test_validate_unexpected_response_body(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
}
|
|
body = {'foo': 12}
|
|
self._test_validate_fail(
|
|
schema, body,
|
|
error_msg="HTTP response body should not exist")
|
|
|
|
def test_validate_missing_response_body(self):
|
|
body = {}
|
|
self._test_validate_fail(self.schema, body)
|
|
|
|
def test_validate_missing_required_attribute(self):
|
|
body = {'notfoo': 12}
|
|
self._test_validate_fail(self.schema, body)
|
|
|
|
def test_validate_response_body_not_list(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'list_items': {
|
|
'type': 'array',
|
|
'items': {'foo': {'type': 'integer'}}
|
|
}
|
|
},
|
|
'required': ['list_items'],
|
|
}
|
|
}
|
|
body = {'foo': 12}
|
|
self._test_validate_fail(schema, body)
|
|
|
|
def test_validate_response_body_list_pass(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'list_items': {
|
|
'type': 'array',
|
|
'items': {'foo': {'type': 'integer'}}
|
|
}
|
|
},
|
|
'required': ['list_items'],
|
|
}
|
|
}
|
|
body = {'list_items': [{'foo': 12}, {'foo': 10}]}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
|
|
class TestRestClientJSONHeaderSchemaValidation(TestJSONSchemaValidationBase):
|
|
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_header': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {'type': 'integer'}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
|
|
def test_validate_header_schema_pass(self):
|
|
resp_body = {}
|
|
resp = self.Response()
|
|
resp.status = 200
|
|
resp.foo = 12
|
|
self.rest_client.validate_response(self.schema, resp, resp_body)
|
|
|
|
def test_validate_header_schema_fail(self):
|
|
resp_body = {}
|
|
resp = self.Response()
|
|
resp.status = 200
|
|
resp.foo = 1.2
|
|
ex = self.assertRaises(exceptions.InvalidHTTPResponseHeader,
|
|
self.rest_client.validate_response,
|
|
self.schema, resp, resp_body)
|
|
self.assertIn("HTTP response header is invalid", ex._error_string)
|
|
|
|
|
|
class TestRestClientJSONSchemaFormatValidation(TestJSONSchemaValidationBase):
|
|
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'format': 'email'
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
|
|
def test_validate_format_pass(self):
|
|
body = {'foo': 'example@example.com'}
|
|
self._test_validate_pass(self.schema, body)
|
|
|
|
def test_validate_format_fail(self):
|
|
body = {'foo': 'wrong_email'}
|
|
self._test_validate_fail(self.schema, body)
|
|
|
|
def test_validate_formats_in_oneOf_pass(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'oneOf': [
|
|
{'format': 'ipv4'},
|
|
{'format': 'ipv6'}
|
|
]
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': '10.0.0.0'}
|
|
self._test_validate_pass(schema, body)
|
|
body = {'foo': 'FE80:0000:0000:0000:0202:B3FF:FE1E:8329'}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
def test_validate_formats_in_oneOf_fail_both_match(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'oneOf': [
|
|
{'format': 'ipv4'},
|
|
{'format': 'ipv4'}
|
|
]
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': '10.0.0.0'}
|
|
self._test_validate_fail(schema, body)
|
|
|
|
def test_validate_formats_in_oneOf_fail_no_match(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'oneOf': [
|
|
{'format': 'ipv4'},
|
|
{'format': 'ipv6'}
|
|
]
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 'wrong_ip_format'}
|
|
self._test_validate_fail(schema, body)
|
|
|
|
def test_validate_formats_in_anyOf_pass(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'anyOf': [
|
|
{'format': 'ipv4'},
|
|
{'format': 'ipv6'}
|
|
]
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': '10.0.0.0'}
|
|
self._test_validate_pass(schema, body)
|
|
body = {'foo': 'FE80:0000:0000:0000:0202:B3FF:FE1E:8329'}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
def test_validate_formats_in_anyOf_pass_both_match(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'anyOf': [
|
|
{'format': 'ipv4'},
|
|
{'format': 'ipv4'}
|
|
]
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': '10.0.0.0'}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
def test_validate_formats_in_anyOf_fail_no_match(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'anyOf': [
|
|
{'format': 'ipv4'},
|
|
{'format': 'ipv6'}
|
|
]
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 'wrong_ip_format'}
|
|
self._test_validate_fail(schema, body)
|
|
|
|
def test_validate_formats_pass_for_unknow_format(self):
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'string',
|
|
'format': 'UNKNOWN'
|
|
}
|
|
},
|
|
'required': ['foo']
|
|
}
|
|
}
|
|
body = {'foo': 'example@example.com'}
|
|
self._test_validate_pass(schema, body)
|
|
|
|
|
|
class TestRestClientJSONSchemaValidatorVersion(TestJSONSchemaValidationBase):
|
|
|
|
schema = {
|
|
'status_code': [200],
|
|
'response_body': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'foo': {'type': 'string'}
|
|
}
|
|
}
|
|
}
|
|
|
|
def test_current_json_schema_validator_version(self):
|
|
with mockpatch.PatchObject(jsonschema.Draft4Validator,
|
|
"check_schema") as chk_schema:
|
|
body = {'foo': 'test'}
|
|
self._test_validate_pass(self.schema, body)
|
|
chk_schema.mock.assert_called_once_with(
|
|
self.schema['response_body'])
|