f46b6c99e0
In json schema, "format" keyword is defined to allow inter-operable
semantic validation for a fixed subset of values.
For example - ip address which comes as 'string' can be further checked
with ipv4 or ipv6 format.
ip_address = {
'type': 'string',
'oneOf': [
{'format': 'ipv4'},
{'format': 'ipv6'}
]
}
Json schema does not validate 'format' correctly if FormatChecker is not
passed during validation. Without FormatChecker, it validate all format as true.
To validate 'format' correctly FormatChecker has to be passed while doing schema
validation.
This commit fix this issue and adds unit tests for that.
Closes-Bug:1460975
Change-Id: Iad6ec9644eb1ec976b4a47f7dec127bfaecfd1fa
1030 lines
36 KiB
Python
1030 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
|
|
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)
|