# 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'])