# 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.rest_client = rest_client.RestClient( fake_auth_provider.FakeAuthProvider(), 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): if enc is None: enc = self.c_type resp_dict = {'status': r_code, 'content-type': enc} resp = httplib2.Response(resp_dict) data = { "method": "fake_method", "url": "fake_url", "headers": "fake_headers", "body": "fake_body", "resp": resp, "resp_body": '{"resp_body": "fake_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.Unauthorized, 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_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_500_with_dict(self): r_body = '{"resp_body": {"err": "fake_resp_body"}}' self.assertRaises(exceptions.ServerFault, self.rest_client._error_checker, **self.set_data("500", r_body=r_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")) 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') 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)