f93048334c
This commit adds support to RestClientException objects to store the response body as a local attr. Additionally it modifies all the uses of the subclassed RestClientExceptions in the base rest client to store the response body on each error type. This is done in order to check the content in some occasion that additional error details are coming back from a service in the body. It will be useful to have the resp_body passed into the exception so that it can be used in additional checks. Change-Id: Ie1bb259e7e683081f0ad127617acc8deb98467c0
645 lines
24 KiB
Python
645 lines
24 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))
|