diff --git a/tests/test_accounts.py b/tests/test_accounts.py new file mode 100644 index 00000000..ec76f134 --- /dev/null +++ b/tests/test_accounts.py @@ -0,0 +1,84 @@ +from testtools import TestCase +from mock import Mock + +from reddwarfclient import accounts +from reddwarfclient import base + +""" +Unit tests for accounts.py +""" + + +class AccountTest(TestCase): + + def setUp(self): + super(AccountTest, self).setUp() + self.orig__init = accounts.Account.__init__ + accounts.Account.__init__ = Mock(return_value=None) + self.account = accounts.Account() + + def tearDown(self): + super(AccountTest, self).tearDown() + accounts.Account.__init__ = self.orig__init + + def test___repr__(self): + self.account.name = "account-1" + self.assertEqual('', self.account.__repr__()) + + +class AccountsTest(TestCase): + + def setUp(self): + super(AccountsTest, self).setUp() + self.orig__init = accounts.Accounts.__init__ + accounts.Accounts.__init__ = Mock(return_value=None) + self.accounts = accounts.Accounts() + self.accounts.api = Mock() + self.accounts.api.client = Mock() + + def tearDown(self): + super(AccountsTest, self).tearDown() + accounts.Accounts.__init__ = self.orig__init + + def test__list(self): + def side_effect_func(self, val): + return val + + self.accounts.resource_class = Mock(side_effect=side_effect_func) + key_ = 'key' + body_ = {key_: "test-value"} + self.accounts.api.client.get = Mock(return_value=('resp', body_)) + self.assertEqual("test-value", self.accounts._list('url', key_)) + + self.accounts.api.client.get = Mock(return_value=('resp', None)) + self.assertRaises(Exception, self.accounts._list, 'url', None) + + def test_index(self): + resp = Mock() + resp.status = 400 + body = {"Accounts": {}} + self.accounts.api.client.get = Mock(return_value=(resp, body)) + self.assertRaises(Exception, self.accounts.index) + resp.status = 200 + self.assertTrue(isinstance(self.accounts.index(), base.Resource)) + self.accounts.api.client.get = Mock(return_value=(resp, None)) + self.assertRaises(Exception, self.accounts.index) + + def test_show(self): + def side_effect_func(acct_name, acct): + return acct_name, acct + + account_ = Mock() + account_.name = "test-account" + self.accounts._list = Mock(side_effect=side_effect_func) + self.assertEqual(('/mgmt/accounts/test-account', 'account'), + self.accounts.show(account_)) + + def test__get_account_name(self): + account_ = 'account with no name' + self.assertEqual(account_, + accounts.Accounts._get_account_name(account_)) + account_ = Mock() + account_.name = "account-name" + self.assertEqual("account-name", + accounts.Accounts._get_account_name(account_)) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 00000000..d52331ff --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,322 @@ +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) diff --git a/tests/test_common.py b/tests/test_common.py new file mode 100644 index 00000000..d090f97c --- /dev/null +++ b/tests/test_common.py @@ -0,0 +1,395 @@ +import sys +import optparse +import json +import collections + +from testtools import TestCase +from mock import Mock + +from reddwarfclient import common +from reddwarfclient import client + +""" + unit tests for common.py +""" + + +class CommonTest(TestCase): + + def setUp(self): + super(CommonTest, self).setUp() + self.orig_sys_exit = sys.exit + sys.exit = Mock(return_value=None) + + def tearDown(self): + super(CommonTest, self).tearDown() + sys.exit = self.orig_sys_exit + + def test_methods_of(self): + class DummyClass: + def dummyMethod(self): + print("just for test") + + obj = DummyClass() + result = common.methods_of(obj) + self.assertEqual(1, len(result)) + method = result['dummyMethod'] + self.assertIsNotNone(method) + + def test_check_for_exceptions(self): + status = [400, 422, 500] + for s in status: + resp = Mock() + resp.status = s + self.assertRaises(ValueError, + common.check_for_exceptions, resp, "body") + + # a no-exception case + resp = Mock() + resp.status = 200 + common.check_for_exceptions(resp, "body") + + def test_print_actions(self): + cmd = "test-cmd" + actions = {"test": "test action", "help": "help action"} + common.print_actions(cmd, actions) + pass + + def test_print_commands(self): + commands = {"cmd-1": "cmd 1", "cmd-2": "cmd 2"} + common.print_commands(commands) + pass + + def test_limit_url(self): + url_ = "test-url" + limit_ = None + marker_ = None + self.assertEqual(url_, common.limit_url(url_)) + + limit_ = "test-limit" + marker_ = "test-marker" + expected = "test-url?marker=test-marker&limit=test-limit" + self.assertEqual(expected, + common.limit_url(url_, limit=limit_, marker=marker_)) + + +class CliOptionsTest(TestCase): + + def check_default_options(self, co): + self.assertEqual(None, co.username) + self.assertEqual(None, co.apikey) + self.assertEqual(None, co.tenant_id) + self.assertEqual(None, co.auth_url) + self.assertEqual('keystone', co.auth_type) + self.assertEqual('reddwarf', co.service_type) + self.assertEqual('Reddwarf', co.service_name) + self.assertEqual('RegionOne', co.region) + self.assertEqual(None, co.service_url) + self.assertFalse(co.insecure) + self.assertFalse(co.verbose) + self.assertFalse(co.debug) + self.assertEqual(None, co.token) + self.assertEqual(None, co.xml) + + def check_option(self, oparser, option_name): + option = oparser.get_option("--%s" % option_name) + self.assertNotEqual(None, option) + if option_name in common.CliOptions.DEFAULT_VALUES: + self.assertEqual(common.CliOptions.DEFAULT_VALUES[option_name], + option.default) + + def test___init__(self): + co = common.CliOptions() + self.check_default_options(co) + + def test_deafult(self): + co = common.CliOptions.default() + self.check_default_options(co) + + def test_load_from_file(self): + co = common.CliOptions.load_from_file() + self.check_default_options(co) + + def test_create_optparser(self): + option_names = ["verbose", "debug", "auth_url", "username", "apikey", + "tenant_id", "auth_type", "service_type", + "service_name", "service_type", "service_name", + "service_url", "region", "insecure", "token", + "xml", "secure", "json", "terse", "hide-debug"] + + oparser = common.CliOptions.create_optparser(True) + for option_name in option_names: + self.check_option(oparser, option_name) + + oparser = common.CliOptions.create_optparser(False) + for option_name in option_names: + self.check_option(oparser, option_name) + + +class ArgumentRequiredTest(TestCase): + + def setUp(self): + super(ArgumentRequiredTest, self).setUp() + self.param = "test-param" + self.arg_req = common.ArgumentRequired(self.param) + + def test___init__(self): + self.assertEqual(self.param, self.arg_req.param) + + def test___str__(self): + expected = 'Argument "--%s" required.' % self.param + self.assertEqual(expected, self.arg_req.__str__()) + + +class CommandsBaseTest(TestCase): + + def setUp(self): + super(CommandsBaseTest, self).setUp() + self.orig_sys_exit = sys.exit + sys.exit = Mock(return_value=None) + parser = common.CliOptions().create_optparser(False) + self.cmd_base = common.CommandsBase(parser) + + def tearDown(self): + super(CommandsBaseTest, self).tearDown() + sys.exit = self.orig_sys_exit + + def test___init__(self): + self.assertNotEqual(None, self.cmd_base) + + def test__get_client(self): + client.log_to_streamhandler = Mock(return_value=None) + expected = Mock() + client.Dbaas = Mock(return_value=expected) + + self.cmd_base.xml = Mock() + self.cmd_base.verbose = False + r = self.cmd_base._get_client() + self.assertEqual(expected, r) + + self.cmd_base.xml = None + self.cmd_base.verbose = True + r = self.cmd_base._get_client() + self.assertEqual(expected, r) + + # test debug true + self.cmd_base.debug = True + client.Dbaas = Mock(side_effect=ValueError) + self.assertRaises(ValueError, self.cmd_base._get_client) + + def test__safe_exec(self): + func = Mock(return_value="test") + self.cmd_base.debug = True + r = self.cmd_base._safe_exec(func) + self.assertEqual("test", r) + + self.cmd_base.debug = False + r = self.cmd_base._safe_exec(func) + self.assertEqual("test", r) + + func = Mock(side_effect=ValueError) # an arbitrary exception + r = self.cmd_base._safe_exec(func) + self.assertEqual(None, r) + + def test__prepare_parser(self): + parser = optparse.OptionParser() + common.CommandsBase.params = ["test_1", "test_2"] + self.cmd_base._prepare_parser(parser) + option = parser.get_option("--%s" % "test_1") + self.assertNotEqual(None, option) + option = parser.get_option("--%s" % "test_2") + self.assertNotEqual(None, option) + + def test__parse_options(self): + parser = optparse.OptionParser() + parser.add_option("--%s" % "test_1", default="test_1v") + parser.add_option("--%s" % "test_2", default="test_2v") + self.cmd_base._parse_options(parser) + self.assertEqual("test_1v", self.cmd_base.test_1) + self.assertEqual("test_2v", self.cmd_base.test_2) + + def test__require(self): + self.assertRaises(common.ArgumentRequired, + self.cmd_base._require, "attr_1") + self.cmd_base.attr_1 = None + self.assertRaises(common.ArgumentRequired, + self.cmd_base._require, "attr_1") + self.cmd_base.attr_1 = "attr_v1" + self.cmd_base._require("attr_1") + + def test__make_list(self): + self.assertRaises(AttributeError, self.cmd_base._make_list, "attr1") + self.cmd_base.attr1 = "v1,v2" + self.cmd_base._make_list("attr1") + self.assertEqual(["v1", "v2"], self.cmd_base.attr1) + self.cmd_base.attr1 = ["v3"] + self.cmd_base._make_list("attr1") + self.assertEqual(["v3"], self.cmd_base.attr1) + + def test__pretty_print(self): + func = Mock(return_value=None) + self.cmd_base.verbose = True + self.assertEqual(None, self.cmd_base._pretty_print(func)) + self.cmd_base.verbose = False + self.assertEqual(None, self.cmd_base._pretty_print(func)) + + def test__dumps(self): + json.dumps = Mock(return_value="test-dump") + self.assertEqual("test-dump", self.cmd_base._dumps("item")) + + def test__pretty_list(self): + func = Mock(return_value=None) + self.cmd_base.verbose = True + self.assertEqual(None, self.cmd_base._pretty_list(func)) + self.cmd_base.verbose = False + self.assertEqual(None, self.cmd_base._pretty_list(func)) + item = Mock(return_value="test") + item._info = "info" + func = Mock(return_value=[item]) + self.assertEqual(None, self.cmd_base._pretty_list(func)) + + def test__pretty_paged(self): + self.cmd_base.limit = "5" + func = Mock(return_value=None) + self.cmd_base.verbose = True + self.assertEqual(None, self.cmd_base._pretty_paged(func)) + + self.cmd_base.verbose = False + + class MockIterable(collections.Iterable): + links = ["item"] + count = 1 + + def __iter__(self): + return ["item1"] + + def __len__(self): + return count + + ret = MockIterable() + func = Mock(return_value=ret) + self.assertEqual(None, self.cmd_base._pretty_paged(func)) + + ret.count = 0 + self.assertEqual(None, self.cmd_base._pretty_paged(func)) + + func = Mock(side_effect=ValueError) + self.assertEqual(None, self.cmd_base._pretty_paged(func)) + self.cmd_base.debug = True + self.cmd_base.marker = Mock() + self.assertRaises(ValueError, self.cmd_base._pretty_paged, func) + + +class AuthTest(TestCase): + + def setUp(self): + super(AuthTest, self).setUp() + self.orig_sys_exit = sys.exit + sys.exit = Mock(return_value=None) + self.parser = common.CliOptions().create_optparser(False) + self.auth = common.Auth(self.parser) + + def tearDown(self): + super(AuthTest, self).tearDown() + sys.exit = self.orig_sys_exit + + def test___init__(self): + self.assertEqual(None, self.auth.dbaas) + self.assertEqual(None, self.auth.apikey) + + def test_login(self): + self.auth.username = "username" + self.auth.apikey = "apikey" + self.auth.tenant_id = "tenant_id" + self.auth.auth_url = "auth_url" + dbaas = Mock() + dbaas.authenticate = Mock(return_value=None) + dbaas.client = Mock() + dbaas.client.auth_token = Mock() + dbaas.client.service_url = Mock() + self.auth._get_client = Mock(return_value=dbaas) + self.auth.login() + + self.auth.debug = True + self.auth._get_client = Mock(side_effect=ValueError) + self.assertRaises(ValueError, self.auth.login) + + self.auth.debug = False + self.auth.login() + + +class AuthedCommandsBaseTest(TestCase): + + def setUp(self): + super(AuthedCommandsBaseTest, self).setUp() + self.orig_sys_exit = sys.exit + sys.exit = Mock(return_value=None) + + def tearDown(self): + super(AuthedCommandsBaseTest, self).tearDown() + sys.exit = self.orig_sys_exit + + def test___init__(self): + parser = common.CliOptions().create_optparser(False) + common.AuthedCommandsBase.debug = True + dbaas = Mock() + dbaas.authenticate = Mock(return_value=None) + dbaas.client = Mock() + dbaas.client.auth_token = Mock() + dbaas.client.service_url = Mock() + dbaas.client.authenticate_with_token = Mock() + common.AuthedCommandsBase._get_client = Mock(return_value=dbaas) + authed_cmd = common.AuthedCommandsBase(parser) + + +class PaginatedTest(TestCase): + + def setUp(self): + super(PaginatedTest, self).setUp() + self.items_ = ["item1", "item2"] + self.next_marker_ = "next-marker" + self.links_ = ["link1", "link2"] + self.pgn = common.Paginated(self.items_, self.next_marker_, + self.links_) + + def tearDown(self): + super(PaginatedTest, self).tearDown() + + def test___init__(self): + self.assertEqual(self.items_, self.pgn.items) + self.assertEqual(self.next_marker_, self.pgn.next) + self.assertEqual(self.links_, self.pgn.links) + + def test___len__(self): + self.assertEqual(len(self.items_), self.pgn.__len__()) + + def test___iter__(self): + itr_expected = self.items_.__iter__() + itr = self.pgn.__iter__() + self.assertEqual(itr_expected.next(), itr.next()) + self.assertEqual(itr_expected.next(), itr.next()) + self.assertRaises(StopIteration, itr_expected.next) + self.assertRaises(StopIteration, itr.next) + + def test___getitem__(self): + self.assertEqual(self.items_[0], self.pgn.__getitem__(0)) + + def test___setitem__(self): + self.pgn.__setitem__(0, "new-item") + self.assertEqual("new-item", self.pgn.items[0]) + + def test___delitem(self): + del self.pgn[0] + self.assertEqual(1, self.pgn.__len__()) + + def test___reversed__(self): + itr = self.pgn.__reversed__() + expected = ["item2", "item1"] + self.assertEqual("item2", itr.next()) + self.assertEqual("item1", itr.next()) + self.assertRaises(StopIteration, itr.next) + + def test___contains__(self): + self.assertTrue(self.pgn.__contains__("item1")) + self.assertTrue(self.pgn.__contains__("item2")) + self.assertFalse(self.pgn.__contains__("item3")) diff --git a/tests/test_instances.py b/tests/test_instances.py new file mode 100644 index 00000000..2131d269 --- /dev/null +++ b/tests/test_instances.py @@ -0,0 +1,176 @@ +from testtools import TestCase +from mock import Mock + +from reddwarfclient import instances +from reddwarfclient import base + +""" +Unit tests for instances.py +""" + + +class InstanceTest(TestCase): + + def setUp(self): + super(InstanceTest, self).setUp() + self.orig__init = instances.Instance.__init__ + instances.Instance.__init__ = Mock(return_value=None) + self.instance = instances.Instance() + self.instance.manager = Mock() + + def tearDown(self): + super(InstanceTest, self).tearDown() + instances.Instance.__init__ = self.orig__init + + def test___repr__(self): + self.instance.name = "instance-1" + self.assertEqual('', self.instance.__repr__()) + + def test_list_databases(self): + db_list = ['database1', 'database2'] + self.instance.manager.databases = Mock() + self.instance.manager.databases.list = Mock(return_value=db_list) + self.assertEqual(db_list, self.instance.list_databases()) + + def test_delete(self): + db_delete_mock = Mock(return_value=None) + self.instance.manager.delete = db_delete_mock + self.instance.delete() + self.assertEqual(1, db_delete_mock.call_count) + + def test_restart(self): + db_restart_mock = Mock(return_value=None) + self.instance.manager.restart = db_restart_mock + self.instance.id = 1 + self.instance.restart() + self.assertEqual(1, db_restart_mock.call_count) + + +class InstancesTest(TestCase): + + def setUp(self): + super(InstancesTest, self).setUp() + self.orig__init = instances.Instances.__init__ + instances.Instances.__init__ = Mock(return_value=None) + self.instances = instances.Instances() + self.instances.api = Mock() + self.instances.api.client = Mock() + self.instances.resource_class = Mock(return_value="instance-1") + + self.orig_base_getid = base.getid + base.getid = Mock(return_value="instance1") + + def tearDown(self): + super(InstancesTest, self).tearDown() + instances.Instances.__init__ = self.orig__init + base.getid = self.orig_base_getid + + def test_create(self): + def side_effect_func(path, body, inst): + return path, body, inst + + self.instances._create = Mock(side_effect=side_effect_func) + p, b, i = self.instances.create("test-name", 103, "test-volume", + ['db1', 'db2'], ['u1', 'u2']) + self.assertEqual("/instances", p) + self.assertEqual("instance", i) + self.assertEqual(['db1', 'db2'], b["instance"]["databases"]) + self.assertEqual(['u1', 'u2'], b["instance"]["users"]) + self.assertEqual("test-name", b["instance"]["name"]) + self.assertEqual("test-volume", b["instance"]["volume"]) + self.assertEqual(103, b["instance"]["flavorRef"]) + + def test__list(self): + self.instances.api.client.get = Mock(return_value=('resp', None)) + self.assertRaises(Exception, self.instances._list, "url", None) + + body = Mock() + body.get = Mock(return_value=[{'href': 'http://test.net/test_file', + 'rel': 'next'}]) + body.__getitem__ = Mock(return_value='instance1') + #self.instances.resource_class = Mock(return_value="instance-1") + self.instances.api.client.get = Mock(return_value=('resp', body)) + _expected = [{'href': 'http://test.net/test_file', 'rel': 'next'}] + self.assertEqual(_expected, self.instances._list("url", None).links) + + def test_list(self): + def side_effect_func(path, inst, limit, marker): + return path, inst, limit, marker + + self.instances._list = Mock(side_effect=side_effect_func) + limit_ = "test-limit" + marker_ = "test-marker" + expected = ("/instances", "instances", limit_, marker_) + self.assertEqual(expected, self.instances.list(limit_, marker_)) + + def test_get(self): + def side_effect_func(path, inst): + return path, inst + + self.instances._get = Mock(side_effect=side_effect_func) + self.assertEqual(('/instances/instance1', 'instance'), + self.instances.get(1)) + + def test_delete(self): + resp = Mock() + resp.status = 200 + body = None + self.instances.api.client.delete = Mock(return_value=(resp, body)) + self.instances.delete('instance1') + resp.status = 500 + self.assertRaises(ValueError, self.instances.delete, 'instance1') + + def test__action(self): + body = Mock() + resp = Mock() + resp.status = 200 + self.instances.api.client.post = Mock(return_value=(resp, body)) + self.assertEqual('instance-1', self.instances._action(1, body)) + + self.instances.api.client.post = Mock(return_value=(resp, None)) + self.assertEqual(None, self.instances._action(1, body)) + + def _set_action_mock(self): + def side_effect_func(instance_id, body): + self._instance_id = instance_id + self._body = body + + self._instance_id = None + self._body = None + self.instances._action = Mock(side_effect=side_effect_func) + + def test_resize_volume(self): + self._set_action_mock() + self.instances.resize_volume(152, 512) + self.assertEqual(152, self._instance_id) + self.assertEqual({"resize": {"volume": {"size": 512}}}, self._body) + + def test_resize_instance(self): + self._set_action_mock() + self.instances.resize_instance(4725, 103) + self.assertEqual(4725, self._instance_id) + self.assertEqual({"resize": {"flavorRef": 103}}, self._body) + + def test_restart(self): + self._set_action_mock() + self.instances.restart(253) + self.assertEqual(253, self._instance_id) + self.assertEqual({'restart': {}}, self._body) + + def test_reset_password(self): + self._set_action_mock() + self.instances.reset_password(634) + self.assertEqual(634, self._instance_id) + self.assertEqual({'reset-password': {}}, self._body) + + +class InstanceStatusTest(TestCase): + + def test_constants(self): + self.assertEqual("ACTIVE", instances.InstanceStatus.ACTIVE) + self.assertEqual("BLOCKED", instances.InstanceStatus.BLOCKED) + self.assertEqual("BUILD", instances.InstanceStatus.BUILD) + self.assertEqual("FAILED", instances.InstanceStatus.FAILED) + self.assertEqual("REBOOT", instances.InstanceStatus.REBOOT) + self.assertEqual("RESIZE", instances.InstanceStatus.RESIZE) + self.assertEqual("SHUTDOWN", instances.InstanceStatus.SHUTDOWN) diff --git a/tests/test_management.py b/tests/test_management.py new file mode 100644 index 00000000..c0b53360 --- /dev/null +++ b/tests/test_management.py @@ -0,0 +1,131 @@ +from testtools import TestCase +from mock import Mock + +from reddwarfclient import management +from reddwarfclient import base + +""" +Unit tests for management.py +""" + + +class RootHistoryTest(TestCase): + + def setUp(self): + super(RootHistoryTest, self).setUp() + self.orig__init = management.RootHistory.__init__ + management.RootHistory.__init__ = Mock(return_value=None) + + def tearDown(self): + super(RootHistoryTest, self).tearDown() + management.RootHistory.__init__ = self.orig__init + + def test___repr__(self): + root_history = management.RootHistory() + root_history.id = "1" + root_history.created = "ct" + root_history.user = "tu" + self.assertEqual('', + root_history.__repr__()) + + +class ManagementTest(TestCase): + + def setUp(self): + super(ManagementTest, self).setUp() + self.orig__init = management.Management.__init__ + management.Management.__init__ = Mock(return_value=None) + self.management = management.Management() + self.management.api = Mock() + self.management.api.client = Mock() + + self.orig_hist__init = management.RootHistory.__init__ + self.orig_base_getid = base.getid + base.getid = Mock(return_value="instance1") + + def tearDown(self): + super(ManagementTest, self).tearDown() + management.Management.__init__ = self.orig__init + management.RootHistory.__init__ = self.orig_hist__init + base.getid = self.orig_base_getid + + def test__list(self): + self.management.api.client.get = Mock(return_value=('resp', None)) + self.assertRaises(Exception, self.management._list, "url", None) + + body = Mock() + body.get = Mock(return_value=[{'href': 'http://test.net/test_file', + 'rel': 'next'}]) + body.__getitem__ = Mock(return_value='instance1') + self.management.resource_class = Mock(return_value="instance-1") + self.management.api.client.get = Mock(return_value=('resp', body)) + _expected = [{'href': 'http://test.net/test_file', 'rel': 'next'}] + self.assertEqual(_expected, self.management._list("url", None).links) + + def test_show(self): + def side_effect_func(path, instance): + return path, instance + self.management._get = Mock(side_effect=side_effect_func) + p, i = self.management.show(1) + self.assertEqual(('/mgmt/instances/instance1', 'instance'), (p, i)) + + def test_index(self): + def side_effect_func(url, name, limit, marker): + return url + + self.management._list = Mock(side_effect=side_effect_func) + self.assertEqual('/mgmt/instances?deleted=true', + self.management.index(deleted=True)) + self.assertEqual('/mgmt/instances?deleted=false', + self.management.index(deleted=False)) + + def test_root_enabled_history(self): + self.management.api.client.get = Mock(return_value=('resp', None)) + self.assertRaises(Exception, + self.management.root_enabled_history, "instance") + body = {'root_history': 'rh'} + self.management.api.client.get = Mock(return_value=('resp', body)) + management.RootHistory.__init__ = Mock(return_value=None) + rh = self.management.root_enabled_history("instance") + self.assertTrue(isinstance(rh, management.RootHistory)) + + def test__action(self): + resp = Mock() + self.management.api.client.post = Mock(return_value=(resp, 'body')) + resp.status = 200 + self.management._action(1, 'body') + self.assertEqual(1, self.management.api.client.post.call_count) + resp.status = 400 + self.assertRaises(ValueError, self.management._action, 1, 'body') + self.assertEqual(2, self.management.api.client.post.call_count) + + def _mock_action(self): + self.body_ = "" + + def side_effect_func(instance_id, body): + self.body_ = body + self.management._action = Mock(side_effect=side_effect_func) + + def test_stop(self): + self._mock_action() + self.management.stop(1) + self.assertEqual(1, self.management._action.call_count) + self.assertEqual({'stop': {}}, self.body_) + + def test_reboot(self): + self._mock_action() + self.management.reboot(1) + self.assertEqual(1, self.management._action.call_count) + self.assertEqual({'reboot': {}}, self.body_) + + def test_migrate(self): + self._mock_action() + self.management.migrate(1) + self.assertEqual(1, self.management._action.call_count) + self.assertEqual({'migrate': {}}, self.body_) + + def test_update(self): + self._mock_action() + self.management.update(1) + self.assertEqual(1, self.management._action.call_count) + self.assertEqual({'update': {}}, self.body_) diff --git a/tests/test_users.py b/tests/test_users.py new file mode 100644 index 00000000..266d9fd5 --- /dev/null +++ b/tests/test_users.py @@ -0,0 +1,100 @@ +from testtools import TestCase +from mock import Mock + +from reddwarfclient import users +from reddwarfclient import base + +""" +Unit tests for users.py +""" + + +class UserTest(TestCase): + + def setUp(self): + super(UserTest, self).setUp() + self.orig__init = users.User.__init__ + users.User.__init__ = Mock(return_value=None) + self.user = users.User() + + def tearDown(self): + super(UserTest, self).tearDown() + users.User.__init__ = self.orig__init + + def test___repr__(self): + self.user.name = "user-1" + self.assertEqual('', self.user.__repr__()) + + +class UsersTest(TestCase): + + def setUp(self): + super(UsersTest, self).setUp() + self.orig__init = users.Users.__init__ + users.Users.__init__ = Mock(return_value=None) + self.users = users.Users() + self.users.api = Mock() + self.users.api.client = Mock() + + self.orig_base_getid = base.getid + base.getid = Mock(return_value="instance1") + + def tearDown(self): + super(UsersTest, self).tearDown() + users.Users.__init__ = self.orig__init + base.getid = self.orig_base_getid + + def _get_mock_method(self): + self._resp = Mock() + self._body = None + self._url = None + + def side_effect_func(url, body=None): + self._body = body + self._url = url + return (self._resp, body) + + return Mock(side_effect=side_effect_func) + + def test_create(self): + self.users.api.client.post = self._get_mock_method() + self._resp.status = 200 + self.users.create(23, 'user1') + self.assertEqual('/instances/23/users', self._url) + self.assertEqual({"users": 'user1'}, self._body) + self._resp.status = 400 + self.assertRaises(Exception, self.users.create, 12, ['user1']) + + def test_delete(self): + self.users.api.client.delete = self._get_mock_method() + self._resp.status = 200 + self.users.delete(27, 'user1') + self.assertEqual('/instances/27/users/user1', self._url) + self._resp.status = 400 + self.assertRaises(Exception, self.users.delete, 34, 'user1') + + def test__list(self): + def side_effect_func(self, val): + return val + + key = 'key' + body = Mock() + body.get = Mock(return_value=[{'href': 'http://test.net/test_file', + 'rel': 'next'}]) + body.__getitem__ = Mock(return_value=["test-value"]) + + resp = Mock() + resp.status = 200 + self.users.resource_class = Mock(side_effect=side_effect_func) + self.users.api.client.get = Mock(return_value=(resp, body)) + self.assertEqual(["test-value"], self.users._list('url', key).items) + + self.users.api.client.get = Mock(return_value=(resp, None)) + self.assertRaises(Exception, self.users._list, 'url', None) + + def test_list(self): + def side_effect_func(path, user, limit, marker): + return path + + self.users._list = Mock(side_effect=side_effect_func) + self.assertEqual('/instances/instance1/users', self.users.list(1))