adding unit tests for common.py client.py management.py users.py instances.py accounts.py
implements blueprint reddwarfclient-unit-tests Change-Id: Icad276aed20795377ed43b89b963339f314bdd3e
This commit is contained in:
denizdemir
parent
28c8f2aa20
commit
7df04ca15a
|
|
@ -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('<Account: account-1>', 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_))
|
||||
|
|
@ -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)
|
||||
|
|
@ -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"))
|
||||
|
|
@ -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('<Instance: instance-1>', 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)
|
||||
|
|
@ -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: Instance 1 enabled at ct by tu>',
|
||||
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_)
|
||||
|
|
@ -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('<User: user-1>', 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))
|
||||
Loading…
Reference in New Issue