9a58df8abe
We are trying to add "get API version" test for Nova API on Temepst side, but validate_response() doesn't work at all for the API because the API returns HTTP300. This patch makes validate_response() work for HTTP30X for the test. Change-Id: If2ce6c9c9e2fde554b4e44dc99ec70d087c3f682
1039 lines
36 KiB
Python
1039 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 copy
|
|
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_error_checker(self, exception_type, data):
|
|
e = self.assertRaises(exception_type,
|
|
self.rest_client._error_checker,
|
|
**data)
|
|
self.assertEqual(e.resp, data['resp'])
|
|
self.assertTrue(hasattr(e, 'resp_body'))
|
|
return e
|
|
|
|
def test_response_400(self):
|
|
self._test_error_checker(exceptions.BadRequest, self.set_data("400"))
|
|
|
|
def test_response_401(self):
|
|
self._test_error_checker(exceptions.Unauthorized, self.set_data("401"))
|
|
|
|
def test_response_403(self):
|
|
self._test_error_checker(exceptions.Forbidden, self.set_data("403"))
|
|
|
|
def test_response_404(self):
|
|
self._test_error_checker(exceptions.NotFound, self.set_data("404"))
|
|
|
|
def test_response_409(self):
|
|
self._test_error_checker(exceptions.Conflict, self.set_data("409"))
|
|
|
|
def test_response_413(self):
|
|
self._test_error_checker(exceptions.OverLimit, self.set_data("413"))
|
|
|
|
def test_response_413_without_absolute_limit(self):
|
|
self._test_error_checker(exceptions.RateLimitExceeded,
|
|
self.set_data("413", absolute_limit=False))
|
|
|
|
def test_response_415(self):
|
|
self._test_error_checker(exceptions.InvalidContentType,
|
|
self.set_data("415"))
|
|
|
|
def test_response_422(self):
|
|
self._test_error_checker(exceptions.UnprocessableEntity,
|
|
self.set_data("422"))
|
|
|
|
def test_response_500_with_text(self):
|
|
# _parse_resp is expected to return 'str'
|
|
self._test_error_checker(exceptions.ServerFault, self.set_data("500"))
|
|
|
|
def test_response_501_with_text(self):
|
|
self._test_error_checker(exceptions.NotImplemented,
|
|
self.set_data("501"))
|
|
|
|
def test_response_400_with_dict(self):
|
|
r_body = '{"resp_body": {"err": "fake_resp_body"}}'
|
|
e = self._test_error_checker(exceptions.BadRequest,
|
|
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._test_error_checker(exceptions.Unauthorized,
|
|
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._test_error_checker(exceptions.Forbidden,
|
|
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._test_error_checker(exceptions.NotFound,
|
|
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._test_error_checker(exceptions.NotFound,
|
|
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._test_error_checker(exceptions.Conflict,
|
|
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._test_error_checker(exceptions.ServerFault,
|
|
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._test_error_checker(exceptions.NotImplemented,
|
|
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._test_error_checker(exceptions.UnexpectedResponseCode,
|
|
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._test_error_checker(exceptions.UnexpectedContentType,
|
|
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_with_http_success_code(self):
|
|
body = {'foo': 12}
|
|
self._test_validate_pass(self.schema, body, status=200)
|
|
|
|
def test_validate_pass_with_http_redirect_code(self):
|
|
body = {'foo': 12}
|
|
schema = copy.deepcopy(self.schema)
|
|
schema['status_code'] = 300
|
|
self._test_validate_pass(schema, body, status=300)
|
|
|
|
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'])
|