diff options
Diffstat (limited to 'reddwarfclient/tests/test_client.py')
| -rw-r--r-- | reddwarfclient/tests/test_client.py | 322 |
1 files changed, 0 insertions, 322 deletions
diff --git a/reddwarfclient/tests/test_client.py b/reddwarfclient/tests/test_client.py deleted file mode 100644 index d52331f..0000000 --- a/reddwarfclient/tests/test_client.py +++ /dev/null @@ -1,322 +0,0 @@ -import contextlib -import os -import logging -import httplib2 -import time - -from testtools import TestCase -from mock import Mock - -from reddwarfclient import client -from reddwarfclient import exceptions -from reddwarfclient import utils - -""" -Unit tests for client.py -""" - - -class ClientTest(TestCase): - - def test_log_to_streamhandler(self): - client.log_to_streamhandler() - self.assertTrue(client._logger.level == logging.DEBUG) - - -class ReddwarfHTTPClientTest(TestCase): - - def setUp(self): - super(ReddwarfHTTPClientTest, self).setUp() - self.orig__init = client.ReddwarfHTTPClient.__init__ - client.ReddwarfHTTPClient.__init__ = Mock(return_value=None) - self.hc = client.ReddwarfHTTPClient() - self.hc.auth_token = "test-auth-token" - self.hc.service_url = "test-service-url/" - self.hc.tenant = "test-tenant" - - self.__debug_lines = list() - - self.orig_client__logger = client._logger - client._logger = Mock() - - self.orig_time = time.time - self.orig_htttp_request = httplib2.Http.request - - def tearDown(self): - super(ReddwarfHTTPClientTest, self).tearDown() - client.ReddwarfHTTPClient.__init__ = self.orig__init - client._logger = self.orig_client__logger - time.time = self.orig_time - httplib2.Http.request = self.orig_htttp_request - - def side_effect_func_for_moc_debug(self, s, *args): - self.__debug_lines.append(s) - - def test___init__(self): - client.ReddwarfHTTPClient.__init__ = self.orig__init - - user = "test-user" - password = "test-password" - tenant = "test-tenant" - auth_url = "http://test-auth-url/" - service_name = None - - # when there is no auth_strategy provided - self.assertRaises(ValueError, client.ReddwarfHTTPClient, user, - password, tenant, auth_url, service_name) - - hc = client.ReddwarfHTTPClient(user, password, tenant, auth_url, - service_name, auth_strategy="fake") - self.assertEqual("http://test-auth-url", hc.auth_url) - - # auth_url is none - hc = client.ReddwarfHTTPClient(user, password, tenant, None, - service_name, auth_strategy="fake") - self.assertEqual(None, hc.auth_url) - - def test_get_timings(self): - self.hc.times = ["item1", "item2"] - self.assertEqual(2, len(self.hc.get_timings())) - self.assertEqual("item1", self.hc.get_timings()[0]) - self.assertEqual("item2", self.hc.get_timings()[1]) - - def test_http_log(self): - self.hc.simple_log = Mock(return_value=None) - self.hc.pretty_log = Mock(return_value=None) - - client.RDC_PP = False - self.hc.http_log(None, None, None, None) - self.assertEqual(1, self.hc.simple_log.call_count) - - client.RDC_PP = True - self.hc.http_log(None, None, None, None) - self.assertEqual(1, self.hc.pretty_log.call_count) - - def test_simple_log(self): - client._logger.isEnabledFor = Mock(return_value=False) - self.hc.simple_log(None, None, None, None) - self.assertEqual(0, len(self.__debug_lines)) - - client._logger.isEnabledFor = Mock(return_value=True) - se = self.side_effect_func_for_moc_debug - client._logger.debug = Mock(side_effect=se) - self.hc.simple_log(['item1', 'GET', 'item3', 'POST', 'item5'], - {'headers': {'e1': 'e1-v', 'e2': 'e2-v'}, - 'body': 'body'}, None, None) - self.assertEqual(3, len(self.__debug_lines)) - self.assertTrue(self.__debug_lines[0].startswith('REQ: curl -i')) - self.assertTrue(self.__debug_lines[1].startswith('REQ BODY:')) - self.assertTrue(self.__debug_lines[2].startswith('RESP:')) - - def test_pretty_log(self): - client._logger.isEnabledFor = Mock(return_value=False) - self.hc.pretty_log(None, None, None, None) - self.assertEqual(0, len(self.__debug_lines)) - - client._logger.isEnabledFor = Mock(return_value=True) - se = self.side_effect_func_for_moc_debug - client._logger.debug = Mock(side_effect=se) - self.hc.pretty_log(['item1', 'GET', 'item3', 'POST', 'item5'], - {'headers': {'e1': 'e1-v', 'e2': 'e2-v'}, - 'body': 'body'}, None, None) - self.assertEqual(5, len(self.__debug_lines)) - self.assertTrue(self.__debug_lines[0].startswith('REQUEST:')) - self.assertTrue(self.__debug_lines[1].startswith('curl -i')) - self.assertTrue(self.__debug_lines[2].startswith('BODY:')) - self.assertTrue(self.__debug_lines[3].startswith('RESPONSE HEADERS:')) - self.assertTrue(self.__debug_lines[4].startswith('RESPONSE BODY')) - - # no body case - self.__debug_lines = list() - self.hc.pretty_log(['item1', 'GET', 'item3', 'POST', 'item5'], - {'headers': {'e1': 'e1-v', 'e2': 'e2-v'}}, - None, None) - self.assertEqual(4, len(self.__debug_lines)) - self.assertTrue(self.__debug_lines[0].startswith('REQUEST:')) - self.assertTrue(self.__debug_lines[1].startswith('curl -i')) - self.assertTrue(self.__debug_lines[2].startswith('RESPONSE HEADERS:')) - self.assertTrue(self.__debug_lines[3].startswith('RESPONSE BODY')) - - def test_request(self): - self.hc.USER_AGENT = "user-agent" - resp = Mock() - body = Mock() - resp.status = 200 - httplib2.Http.request = Mock(return_value=(resp, body)) - self.hc.morph_response_body = Mock(return_value=body) - r, b = self.hc.request() - self.assertEqual(resp, r) - self.assertEqual(body, b) - self.assertEqual((resp, body), self.hc.last_response) - - httplib2.Http.request = Mock(return_value=(resp, None)) - r, b = self.hc.request() - self.assertEqual(resp, r) - self.assertEqual(None, b) - - status_list = [400, 401, 403, 404, 408, 409, 413, 500, 501] - for status in status_list: - resp.status = status - self.assertRaises(ValueError, self.hc.request) - - exception = exceptions.ResponseFormatError - self.hc.morph_response_body = Mock(side_effect=exception) - self.assertRaises(Exception, self.hc.request) - - def test_raise_error_from_status(self): - resp = Mock() - resp.status = 200 - self.hc.raise_error_from_status(resp, Mock()) - - status_list = [400, 401, 403, 404, 408, 409, 413, 500, 501] - for status in status_list: - resp.status = status - self.assertRaises(ValueError, - self.hc.raise_error_from_status, resp, Mock()) - - def test_morph_request(self): - kwargs = dict() - kwargs['headers'] = dict() - kwargs['body'] = ['body', {'item1': 'value1'}] - self.hc.morph_request(kwargs) - expected = {'body': '["body", {"item1": "value1"}]', - 'headers': {'Content-Type': 'application/json', - 'Accept': 'application/json'}} - self.assertEqual(expected, kwargs) - - def test_morph_response_body(self): - body_string = '["body", {"item1": "value1"}]' - expected = ['body', {'item1': 'value1'}] - self.assertEqual(expected, self.hc.morph_response_body(body_string)) - body_string = '["body", {"item1": }]' - self.assertRaises(exceptions.ResponseFormatError, - self.hc.morph_response_body, body_string) - - def test__time_request(self): - self.__time = 0 - - def side_effect_func(): - self.__time = self.__time + 1 - return self.__time - - time.time = Mock(side_effect=side_effect_func) - self.hc.request = Mock(return_value=("mock-response", "mock-body")) - self.hc.times = list() - resp, body = self.hc._time_request("test-url", "Get") - self.assertEqual(("mock-response", "mock-body"), (resp, body)) - self.assertEqual([('Get test-url', 1, 2)], self.hc.times) - - def mock_time_request_func(self): - def side_effect_func(url, method, **kwargs): - return url, method - self.hc._time_request = Mock(side_effect=side_effect_func) - - def test__cs_request(self): - self.mock_time_request_func() - resp, body = self.hc._cs_request("test-url", "GET") - self.assertEqual(('test-service-url/test-url', 'GET'), (resp, body)) - - self.hc.authenticate = Mock(side_effect=ValueError) - self.hc.auth_token = None - self.hc.service_url = None - self.assertRaises(ValueError, self.hc._cs_request, "test-url", "GET") - - self.hc.authenticate = Mock(return_value=None) - self.hc.service_url = "test-service-url/" - - def side_effect_func_time_req(url, method, **kwargs): - raise exceptions.Unauthorized(None) - - self.hc._time_request = Mock(side_effect=side_effect_func_time_req) - self.assertRaises(exceptions.Unauthorized, - self.hc._cs_request, "test-url", "GET") - - def test_get(self): - self.mock_time_request_func() - resp, body = self.hc.get("test-url") - self.assertEqual(("test-service-url/test-url", "GET"), (resp, body)) - - def test_post(self): - self.mock_time_request_func() - resp, body = self.hc.post("test-url") - self.assertEqual(("test-service-url/test-url", "POST"), (resp, body)) - - def test_put(self): - self.mock_time_request_func() - resp, body = self.hc.put("test-url") - self.assertEqual(("test-service-url/test-url", "PUT"), (resp, body)) - - def test_delete(self): - self.mock_time_request_func() - resp, body = self.hc.delete("test-url") - self.assertEqual(("test-service-url/test-url", "DELETE"), (resp, body)) - - def test_authenticate(self): - self.hc.authenticator = Mock() - catalog = Mock() - catalog.get_public_url = Mock(return_value="public-url") - catalog.get_management_url = Mock(return_value="mng-url") - catalog.get_token = Mock(return_value="test-token") - - self.__auth_calls = [] - - def side_effect_func(token, url): - self.__auth_calls = [token, url] - - self.hc.authenticate_with_token = Mock(side_effect=side_effect_func) - self.hc.authenticator.authenticate = Mock(return_value=catalog) - self.hc.endpoint_type = "publicURL" - self.hc.authenticate() - self.assertEqual(["test-token", None], - self.__auth_calls) - - self.__auth_calls = [] - self.hc.service_url = None - self.hc.authenticate() - self.assertEqual(["test-token", "public-url"], self.__auth_calls) - - self.__auth_calls = [] - self.hc.endpoint_type = "adminURL" - self.hc.authenticate() - self.assertEqual(["test-token", "mng-url"], self.__auth_calls) - - def test_authenticate_with_token(self): - self.hc.service_url = None - self.assertRaises(exceptions.ServiceUrlNotGiven, - self.hc.authenticate_with_token, "token", None) - self.hc.authenticate_with_token("token", "test-url") - self.assertEqual("test-url", self.hc.service_url) - self.assertEqual("token", self.hc.auth_token) - - -class DbaasTest(TestCase): - - def setUp(self): - super(DbaasTest, self).setUp() - self.orig__init = client.ReddwarfHTTPClient.__init__ - client.ReddwarfHTTPClient.__init__ = Mock(return_value=None) - self.dbaas = client.Dbaas("user", "api-key") - - def tearDown(self): - super(DbaasTest, self).tearDown() - client.ReddwarfHTTPClient.__init__ = self.orig__init - - def test___init__(self): - client.ReddwarfHTTPClient.__init__ = Mock(return_value=None) - self.assertNotEqual(None, self.dbaas.mgmt) - - def test_set_management_url(self): - self.dbaas.set_management_url("test-management-url") - self.assertEqual("test-management-url", - self.dbaas.client.management_url) - - def test_get_timings(self): - __timings = {'start': 1, 'end': 2} - self.dbaas.client.get_timings = Mock(return_value=__timings) - self.assertEqual(__timings, self.dbaas.get_timings()) - - def test_authenticate(self): - mock_auth = Mock(return_value=None) - self.dbaas.client.authenticate = mock_auth - self.dbaas.authenticate() - self.assertEqual(1, mock_auth.call_count) |
