diff options
Diffstat (limited to 'tempest_lib/tests/test_rest_client.py')
| -rw-r--r-- | tempest_lib/tests/test_rest_client.py | 1065 |
1 files changed, 0 insertions, 1065 deletions
diff --git a/tempest_lib/tests/test_rest_client.py b/tempest_lib/tests/test_rest_client.py deleted file mode 100644 index 452cef5..0000000 --- a/tempest_lib/tests/test_rest_client.py +++ /dev/null @@ -1,1065 +0,0 @@ -# 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], - }} - null_dict = {} - - 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) - - def test_parse_nullable_dict(self): - body = self.rest_client._parse_resp(json.dumps(self.null_dict)) - self.assertEqual(self.null_dict, 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_410(self): - self._test_error_checker(exceptions.Gone, self.set_data("410")) - - 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_410_with_dict(self): - r_body = '{"resp_body": {"err": "fake_resp_body"}}' - e = self._test_error_checker(exceptions.Gone, - self.set_data("410", 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_410_with_invalid_dict(self): - r_body = '{"foo": "bar"]' - e = self._test_error_checker(exceptions.Gone, - self.set_data("410", 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']) |