summaryrefslogtreecommitdiff
path: root/troveclient/tests
diff options
context:
space:
mode:
Diffstat (limited to 'troveclient/tests')
-rw-r--r--troveclient/tests/__init__.py0
-rw-r--r--troveclient/tests/test_accounts.py84
-rw-r--r--troveclient/tests/test_auth.py414
-rw-r--r--troveclient/tests/test_base.py447
-rw-r--r--troveclient/tests/test_client.py322
-rw-r--r--troveclient/tests/test_common.py395
-rw-r--r--troveclient/tests/test_instances.py176
-rw-r--r--troveclient/tests/test_limits.py79
-rw-r--r--troveclient/tests/test_management.py144
-rw-r--r--troveclient/tests/test_secgroups.py102
-rw-r--r--troveclient/tests/test_users.py126
-rw-r--r--troveclient/tests/test_utils.py41
-rw-r--r--troveclient/tests/test_xml.py241
13 files changed, 2571 insertions, 0 deletions
diff --git a/troveclient/tests/__init__.py b/troveclient/tests/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/troveclient/tests/__init__.py
diff --git a/troveclient/tests/test_accounts.py b/troveclient/tests/test_accounts.py
new file mode 100644
index 0000000..e2716fa
--- /dev/null
+++ b/troveclient/tests/test_accounts.py
@@ -0,0 +1,84 @@
+from testtools import TestCase
+from mock import Mock
+
+from troveclient import accounts
+from troveclient 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_))
diff --git a/troveclient/tests/test_auth.py b/troveclient/tests/test_auth.py
new file mode 100644
index 0000000..28cecb9
--- /dev/null
+++ b/troveclient/tests/test_auth.py
@@ -0,0 +1,414 @@
+import contextlib
+
+from testtools import TestCase
+from troveclient import auth
+from mock import Mock
+
+from troveclient import exceptions
+
+"""
+Unit tests for the classes and functions in auth.py.
+"""
+
+
+def check_url_none(test_case, auth_class):
+ # url is None, it must throw exception
+ authObj = auth_class(url=None, type=auth_class, client=None,
+ username=None, password=None, tenant=None)
+ try:
+ authObj.authenticate()
+ test_case.fail("AuthUrlNotGiven exception expected")
+ except exceptions.AuthUrlNotGiven:
+ pass
+
+
+class AuthenticatorTest(TestCase):
+
+ def setUp(self):
+ super(AuthenticatorTest, self).setUp()
+ self.orig_load = auth.ServiceCatalog._load
+ self.orig__init = auth.ServiceCatalog.__init__
+
+ def tearDown(self):
+ super(AuthenticatorTest, self).tearDown()
+ auth.ServiceCatalog._load = self.orig_load
+ auth.ServiceCatalog.__init__ = self.orig__init
+
+ def test_get_authenticator_cls(self):
+ class_list = (auth.KeyStoneV2Authenticator,
+ auth.RaxAuthenticator,
+ auth.Auth1_1,
+ auth.FakeAuth)
+
+ for c in class_list:
+ self.assertEqual(c, auth.get_authenticator_cls(c))
+
+ class_names = {"keystone": auth.KeyStoneV2Authenticator,
+ "rax": auth.RaxAuthenticator,
+ "auth1.1": auth.Auth1_1,
+ "fake": auth.FakeAuth}
+
+ for cn in class_names.keys():
+ self.assertEqual(class_names[cn], auth.get_authenticator_cls(cn))
+
+ cls_or_name = "_unknown_"
+ self.assertRaises(ValueError, auth.get_authenticator_cls, cls_or_name)
+
+ def test__authenticate(self):
+ authObj = auth.Authenticator(Mock(), auth.KeyStoneV2Authenticator,
+ Mock(), Mock(), Mock(), Mock())
+ # test response code 200
+ resp = Mock()
+ resp.status = 200
+ body = "test_body"
+
+ auth.ServiceCatalog._load = Mock(return_value=1)
+ authObj.client._time_request = Mock(return_value=(resp, body))
+
+ sc = authObj._authenticate(Mock(), Mock())
+ self.assertEqual(body, sc.catalog)
+
+ # test AmbiguousEndpoints exception
+ auth.ServiceCatalog.__init__ = \
+ Mock(side_effect=exceptions.AmbiguousEndpoints)
+ self.assertRaises(exceptions.AmbiguousEndpoints,
+ authObj._authenticate, Mock(), Mock())
+
+ # test handling KeyError and raising AuthorizationFailure exception
+ auth.ServiceCatalog.__init__ = Mock(side_effect=KeyError)
+ self.assertRaises(exceptions.AuthorizationFailure,
+ authObj._authenticate, Mock(), Mock())
+
+ # test EndpointNotFound exception
+ mock = Mock(side_effect=exceptions.EndpointNotFound)
+ auth.ServiceCatalog.__init__ = mock
+ self.assertRaises(exceptions.EndpointNotFound,
+ authObj._authenticate, Mock(), Mock())
+ mock.side_effect = None
+
+ # test response code 305
+ resp.__getitem__ = Mock(return_value='loc')
+ resp.status = 305
+ body = "test_body"
+ authObj.client._time_request = Mock(return_value=(resp, body))
+
+ l = authObj._authenticate(Mock(), Mock())
+ self.assertEqual('loc', l)
+
+ # test any response code other than 200 and 305
+ resp.status = 404
+ exceptions.from_response = Mock(side_effect=ValueError)
+ self.assertRaises(ValueError, authObj._authenticate, Mock(), Mock())
+
+ def test_authenticate(self):
+ authObj = auth.Authenticator(Mock(), auth.KeyStoneV2Authenticator,
+ Mock(), Mock(), Mock(), Mock())
+ self.assertRaises(NotImplementedError, authObj.authenticate)
+
+
+class KeyStoneV2AuthenticatorTest(TestCase):
+
+ def test_authenticate(self):
+ # url is None
+ check_url_none(self, auth.KeyStoneV2Authenticator)
+
+ # url is not None, so it must not throw exception
+ url = "test_url"
+ cls_type = auth.KeyStoneV2Authenticator
+ authObj = auth.KeyStoneV2Authenticator(url=url, type=cls_type,
+ client=None, username=None,
+ password=None, tenant=None)
+
+ def side_effect_func(url):
+ return url
+
+ mock = Mock()
+ mock.side_effect = side_effect_func
+ authObj._v2_auth = mock
+ r = authObj.authenticate()
+ self.assertEqual(url, r)
+
+ def test__v2_auth(self):
+ username = "trove_user"
+ password = "trove_password"
+ tenant = "tenant"
+ cls_type = auth.KeyStoneV2Authenticator
+ authObj = auth.KeyStoneV2Authenticator(url=None, type=cls_type,
+ client=None,
+ username=username,
+ password=password,
+ tenant=tenant)
+
+ def side_effect_func(url, body):
+ return body
+ mock = Mock()
+ mock.side_effect = side_effect_func
+ authObj._authenticate = mock
+ body = authObj._v2_auth(Mock())
+ self.assertEqual(username,
+ body['auth']['passwordCredentials']['username'])
+ self.assertEqual(password,
+ body['auth']['passwordCredentials']['password'])
+ self.assertEqual(tenant, body['auth']['tenantName'])
+
+
+class Auth1_1Test(TestCase):
+
+ def test_authenticate(self):
+ # handle when url is None
+ check_url_none(self, auth.Auth1_1)
+
+ # url is not none
+ username = "trove_user"
+ password = "trove_password"
+ url = "test_url"
+ authObj = auth.Auth1_1(url=url,
+ type=auth.Auth1_1,
+ client=None, username=username,
+ password=password, tenant=None)
+
+ def side_effect_func(auth_url, body, root_key):
+ return auth_url, body, root_key
+
+ mock = Mock()
+ mock.side_effect = side_effect_func
+ authObj._authenticate = mock
+ auth_url, body, root_key = authObj.authenticate()
+
+ self.assertEqual(username, body['credentials']['username'])
+ self.assertEqual(password, body['credentials']['key'])
+ self.assertEqual(auth_url, url)
+ self.assertEqual('auth', root_key)
+
+
+class RaxAuthenticatorTest(TestCase):
+
+ def test_authenticate(self):
+ # url is None
+ check_url_none(self, auth.RaxAuthenticator)
+
+ # url is not None, so it must not throw exception
+ url = "test_url"
+ authObj = auth.RaxAuthenticator(url=url,
+ type=auth.RaxAuthenticator,
+ client=None, username=None,
+ password=None, tenant=None)
+
+ def side_effect_func(url):
+ return url
+
+ mock = Mock()
+ mock.side_effect = side_effect_func
+ authObj._rax_auth = mock
+ r = authObj.authenticate()
+ self.assertEqual(url, r)
+
+ def test__rax_auth(self):
+ username = "trove_user"
+ password = "trove_password"
+ tenant = "tenant"
+ authObj = auth.RaxAuthenticator(url=None,
+ type=auth.RaxAuthenticator,
+ client=None, username=username,
+ password=password, tenant=tenant)
+
+ def side_effect_func(url, body):
+ return body
+
+ mock = Mock()
+ mock.side_effect = side_effect_func
+ authObj._authenticate = mock
+ body = authObj._rax_auth(Mock())
+
+ v = body['auth']['RAX-KSKEY:apiKeyCredentials']['username']
+ self.assertEqual(username, v)
+
+ v = body['auth']['RAX-KSKEY:apiKeyCredentials']['apiKey']
+ self.assertEqual(password, v)
+
+ v = body['auth']['RAX-KSKEY:apiKeyCredentials']['tenantName']
+ self.assertEqual(tenant, v)
+
+
+class FakeAuthTest(TestCase):
+
+ def test_authenticate(self):
+ tenant = "tenant"
+ authObj = auth.FakeAuth(url=None,
+ type=auth.FakeAuth,
+ client=None, username=None,
+ password=None, tenant=tenant)
+
+ fc = authObj.authenticate()
+ public_url = "%s/%s" % ('http://localhost:8779/v1.0', tenant)
+ self.assertEqual(public_url, fc.get_public_url())
+ self.assertEqual(tenant, fc.get_token())
+
+
+class ServiceCatalogTest(TestCase):
+
+ def setUp(self):
+ super(ServiceCatalogTest, self).setUp()
+ self.orig_url_for = auth.ServiceCatalog._url_for
+ self.orig__init__ = auth.ServiceCatalog.__init__
+ auth.ServiceCatalog.__init__ = Mock(return_value=None)
+ self.test_url = "http://localhost:1234/test"
+
+ def tearDown(self):
+ super(ServiceCatalogTest, self).tearDown()
+ auth.ServiceCatalog._url_for = self.orig_url_for
+ auth.ServiceCatalog.__init__ = self.orig__init__
+
+ def test__load(self):
+ url = "random_url"
+ auth.ServiceCatalog._url_for = Mock(return_value=url)
+
+ # when service_url is None
+ scObj = auth.ServiceCatalog()
+ scObj.region = None
+ scObj.service_url = None
+ scObj._load()
+ self.assertEqual(url, scObj.public_url)
+ self.assertEqual(url, scObj.management_url)
+
+ # service url is not None
+ service_url = "service_url"
+ scObj = auth.ServiceCatalog()
+ scObj.region = None
+ scObj.service_url = service_url
+ scObj._load()
+ self.assertEqual(service_url, scObj.public_url)
+ self.assertEqual(service_url, scObj.management_url)
+
+ def test_get_token(self):
+ test_id = "test_id"
+ scObj = auth.ServiceCatalog()
+ scObj.root_key = "root_key"
+ scObj.catalog = dict()
+ scObj.catalog[scObj.root_key] = dict()
+ scObj.catalog[scObj.root_key]['token'] = dict()
+ scObj.catalog[scObj.root_key]['token']['id'] = test_id
+ self.assertEqual(test_id, scObj.get_token())
+
+ def test_get_management_url(self):
+ test_mng_url = "test_management_url"
+ scObj = auth.ServiceCatalog()
+ scObj.management_url = test_mng_url
+ self.assertEqual(test_mng_url, scObj.get_management_url())
+
+ def test_get_public_url(self):
+ test_public_url = "test_public_url"
+ scObj = auth.ServiceCatalog()
+ scObj.public_url = test_public_url
+ self.assertEqual(test_public_url, scObj.get_public_url())
+
+ def test__url_for(self):
+ scObj = auth.ServiceCatalog()
+
+ # case for no endpoint found
+ self.case_no_endpoint_match(scObj)
+
+ # case for empty service catalog
+ self.case_endpoing_with_empty_catalog(scObj)
+
+ # more than one matching endpoints
+ self.case_ambiguous_endpoint(scObj)
+
+ # happy case
+ self.case_unique_endpoint(scObj)
+
+ # testing if-statements in for-loop to iterate services in catalog
+ self.case_iterating_services_in_catalog(scObj)
+
+ def case_no_endpoint_match(self, scObj):
+ # empty endpoint list
+ scObj.catalog = dict()
+ scObj.catalog['endpoints'] = list()
+ self.assertRaises(exceptions.EndpointNotFound, scObj._url_for)
+
+ def side_effect_func_ep(attr):
+ return "test_attr_value"
+
+ # simulating dict
+ endpoint = Mock()
+ mock = Mock()
+ mock.side_effect = side_effect_func_ep
+ endpoint.__getitem__ = mock
+ scObj.catalog['endpoints'].append(endpoint)
+
+ # not-empty list but not matching endpoint
+ filter_value = "not_matching_value"
+ self.assertRaises(exceptions.EndpointNotFound, scObj._url_for,
+ attr="test_attr", filter_value=filter_value)
+
+ filter_value = "test_attr_value" # so that we have an endpoint match
+ scObj.root_key = "access"
+ scObj.catalog[scObj.root_key] = dict()
+ self.assertRaises(exceptions.EndpointNotFound, scObj._url_for,
+ attr="test_attr", filter_value=filter_value)
+
+ def case_endpoing_with_empty_catalog(self, scObj):
+ # first, test with empty catalog, this should pass since
+ # there is already enpoint added
+ scObj.catalog[scObj.root_key]['serviceCatalog'] = list()
+
+ endpoint = scObj.catalog['endpoints'][0]
+ endpoint.get = Mock(return_value=self.test_url)
+ r_url = scObj._url_for(attr="test_attr",
+ filter_value="test_attr_value")
+ self.assertEqual(self.test_url, r_url)
+
+ def case_ambiguous_endpoint(self, scObj):
+ scObj.service_type = "trove"
+ scObj.service_name = "test_service_name"
+
+ def side_effect_func_service(key):
+ if key == "type":
+ return "trove"
+ elif key == "name":
+ return "test_service_name"
+ return None
+
+ mock1 = Mock()
+ mock1.side_effect = side_effect_func_service
+ service1 = Mock()
+ service1.get = mock1
+
+ endpoint2 = {"test_attr": "test_attr_value"}
+ service1.__getitem__ = Mock(return_value=[endpoint2])
+ scObj.catalog[scObj.root_key]['serviceCatalog'] = [service1]
+ self.assertRaises(exceptions.AmbiguousEndpoints, scObj._url_for,
+ attr="test_attr", filter_value="test_attr_value")
+
+ def case_unique_endpoint(self, scObj):
+ # changing the endpoint2 attribute to pass the filter
+ service1 = scObj.catalog[scObj.root_key]['serviceCatalog'][0]
+ endpoint2 = service1[0][0]
+ endpoint2["test_attr"] = "new value not matching filter"
+ r_url = scObj._url_for(attr="test_attr",
+ filter_value="test_attr_value")
+ self.assertEqual(self.test_url, r_url)
+
+ def case_iterating_services_in_catalog(self, scObj):
+ service1 = scObj.catalog[scObj.root_key]['serviceCatalog'][0]
+
+ scObj.catalog = dict()
+ scObj.root_key = "access"
+ scObj.catalog[scObj.root_key] = dict()
+ scObj.service_type = "no_match"
+
+ scObj.catalog[scObj.root_key]['serviceCatalog'] = [service1]
+ self.assertRaises(exceptions.EndpointNotFound, scObj._url_for)
+
+ scObj.service_type = "database"
+ scObj.service_name = "no_match"
+ self.assertRaises(exceptions.EndpointNotFound, scObj._url_for)
+
+ # no endpoints and no 'serviceCatalog' in catalog => raise exception
+ scObj = auth.ServiceCatalog()
+ scObj.catalog = dict()
+ scObj.root_key = "access"
+ scObj.catalog[scObj.root_key] = dict()
+ scObj.catalog[scObj.root_key]['serviceCatalog'] = []
+ self.assertRaises(exceptions.EndpointNotFound, scObj._url_for,
+ attr="test_attr", filter_value="test_attr_value")
diff --git a/troveclient/tests/test_base.py b/troveclient/tests/test_base.py
new file mode 100644
index 0000000..ac7318b
--- /dev/null
+++ b/troveclient/tests/test_base.py
@@ -0,0 +1,447 @@
+import contextlib
+import os
+
+from testtools import TestCase
+from mock import Mock
+
+from troveclient import base
+from troveclient import exceptions
+from troveclient import utils
+
+"""
+Unit tests for base.py
+"""
+
+
+def obj_class(self, res, loaded=True):
+ return res
+
+
+class BaseTest(TestCase):
+
+ def test_getid(self):
+ obj = "test"
+ r = base.getid(obj)
+ self.assertEqual(obj, r)
+
+ test_id = "test_id"
+ obj = Mock()
+ obj.id = test_id
+ r = base.getid(obj)
+ self.assertEqual(test_id, r)
+
+
+class ManagerTest(TestCase):
+
+ def setUp(self):
+ super(ManagerTest, self).setUp()
+ self.orig__init = base.Manager.__init__
+ base.Manager.__init__ = Mock(return_value=None)
+ self.orig_os_makedirs = os.makedirs
+
+ def tearDown(self):
+ super(ManagerTest, self).tearDown()
+ base.Manager.__init__ = self.orig__init
+ os.makedirs = self.orig_os_makedirs
+
+ def test___init__(self):
+ api = Mock()
+ base.Manager.__init__ = self.orig__init
+ manager = base.Manager(api)
+ self.assertEqual(api, manager.api)
+
+ def test_completion_cache(self):
+ manager = base.Manager()
+
+ # handling exceptions
+ mode = "w"
+ cache_type = "unittest"
+ obj_class = Mock
+ with manager.completion_cache(cache_type, obj_class, mode):
+ pass
+
+ os.makedirs = Mock(side_effect=OSError)
+ with manager.completion_cache(cache_type, obj_class, mode):
+ pass
+
+ def test_write_to_completion_cache(self):
+ manager = base.Manager()
+
+ # no cache object, nothing should happen
+ manager.write_to_completion_cache("non-exist", "val")
+
+ def side_effect_func(val):
+ return val
+
+ manager._mock_cache = Mock()
+ manager._mock_cache.write = Mock(return_value=None)
+ manager.write_to_completion_cache("mock", "val")
+ self.assertEqual(1, manager._mock_cache.write.call_count)
+
+ def _get_mock(self):
+ manager = base.Manager()
+ manager.api = Mock()
+ manager.api.client = Mock()
+
+ def side_effect_func(self, body, loaded=True):
+ return body
+
+ manager.resource_class = Mock(side_effect=side_effect_func)
+ return manager
+
+ def test__get_with_response_key_none(self):
+ manager = self._get_mock()
+ url_ = "test-url"
+ body_ = "test-body"
+ resp_ = "test-resp"
+ manager.api.client.get = Mock(return_value=(resp_, body_))
+ r = manager._get(url=url_, response_key=None)
+ self.assertEqual(body_, r)
+
+ def test__get_with_response_key(self):
+ manager = self._get_mock()
+ response_key = "response_key"
+ body_ = {response_key: "test-resp-key-body"}
+ url_ = "test_url_get"
+ manager.api.client.get = Mock(return_value=(url_, body_))
+ r = manager._get(url=url_, response_key=response_key)
+ self.assertEqual(body_[response_key], r)
+
+ def test__create(self):
+ manager = base.Manager()
+ manager.api = Mock()
+ manager.api.client = Mock()
+
+ response_key = "response_key"
+ data_ = "test-data"
+ body_ = {response_key: data_}
+ url_ = "test_url_post"
+ manager.api.client.post = Mock(return_value=(url_, body_))
+
+ return_raw = True
+ r = manager._create(url_, body_, response_key, return_raw)
+ self.assertEqual(data_, r)
+
+ return_raw = False
+
+ @contextlib.contextmanager
+ def completion_cache_mock(*arg, **kwargs):
+ yield
+
+ mock = Mock()
+ mock.side_effect = completion_cache_mock
+ manager.completion_cache = mock
+
+ manager.resource_class = Mock(return_value="test-class")
+ r = manager._create(url_, body_, response_key, return_raw)
+ self.assertEqual("test-class", r)
+
+ def get_mock_mng_api_client(self):
+ manager = base.Manager()
+ manager.api = Mock()
+ manager.api.client = Mock()
+ return manager
+
+ def test__delete(self):
+ resp_ = "test-resp"
+ body_ = "test-body"
+
+ manager = self.get_mock_mng_api_client()
+ manager.api.client.delete = Mock(return_value=(resp_, body_))
+ # _delete just calls api.client.delete, and does nothing
+ # the correctness should be tested in api class
+ manager._delete("test-url")
+ pass
+
+ def test__update(self):
+ resp_ = "test-resp"
+ body_ = "test-body"
+
+ manager = self.get_mock_mng_api_client()
+ manager.api.client.put = Mock(return_value=(resp_, body_))
+ body = manager._update("test-url", body_)
+ self.assertEqual(body_, body)
+
+
+class ManagerListTest(ManagerTest):
+
+ def setUp(self):
+ super(ManagerListTest, self).setUp()
+
+ @contextlib.contextmanager
+ def completion_cache_mock(*arg, **kwargs):
+ yield
+
+ self.manager = base.Manager()
+ self.manager.api = Mock()
+ self.manager.api.client = Mock()
+
+ self.response_key = "response_key"
+ self.data_p = ["p1", "p2"]
+ self.body_p = {self.response_key: self.data_p}
+ self.url_p = "test_url_post"
+ self.manager.api.client.post = Mock(return_value=(self.url_p,
+ self.body_p))
+
+ self.data_g = ["g1", "g2", "g3"]
+ self.body_g = {self.response_key: self.data_g}
+ self.url_g = "test_url_get"
+ self.manager.api.client.get = Mock(return_value=(self.url_g,
+ self.body_g))
+
+ mock = Mock()
+ mock.side_effect = completion_cache_mock
+ self.manager.completion_cache = mock
+
+ def tearDown(self):
+ super(ManagerListTest, self).tearDown()
+
+ def obj_class(self, res, loaded=True):
+ return res
+
+ def test_list_with_body_none(self):
+ body = None
+ l = self.manager._list("url", self.response_key, obj_class, body)
+ self.assertEqual(len(self.data_g), len(l))
+ for i in range(0, len(l)):
+ self.assertEqual(self.data_g[i], l[i])
+
+ def test_list_body_not_none(self):
+ body = "something"
+ l = self.manager._list("url", self.response_key, obj_class, body)
+ self.assertEqual(len(self.data_p), len(l))
+ for i in range(0, len(l)):
+ self.assertEqual(self.data_p[i], l[i])
+
+ def test_list_key_mapping(self):
+ data_ = {"values": ["p1", "p2"]}
+ body_ = {self.response_key: data_}
+ url_ = "test_url_post"
+ self.manager.api.client.post = Mock(return_value=(url_, body_))
+ l = self.manager._list("url", self.response_key,
+ obj_class, "something")
+ data = data_["values"]
+ self.assertEqual(len(data), len(l))
+ for i in range(0, len(l)):
+ self.assertEqual(data[i], l[i])
+
+ def test_list_without_key_mapping(self):
+ data_ = {"v1": "1", "v2": "2"}
+ body_ = {self.response_key: data_}
+ url_ = "test_url_post"
+ self.manager.api.client.post = Mock(return_value=(url_, body_))
+ l = self.manager._list("url", self.response_key,
+ obj_class, "something")
+ self.assertEqual(len(data_), len(l))
+
+
+class ManagerWithFind(TestCase):
+
+ def setUp(self):
+ super(ManagerWithFind, self).setUp()
+ self.orig__init = base.ManagerWithFind.__init__
+ base.ManagerWithFind.__init__ = Mock(return_value=None)
+ self.manager = base.ManagerWithFind()
+
+ def tearDown(self):
+ super(ManagerWithFind, self).tearDown()
+ base.ManagerWithFind.__init__ = self.orig__init
+
+ def test_find(self):
+ obj1 = Mock()
+ obj1.attr1 = "v1"
+ obj1.attr2 = "v2"
+ obj1.attr3 = "v3"
+
+ obj2 = Mock()
+ obj2.attr1 = "v1"
+ obj2.attr2 = "v2"
+
+ self.manager.list = Mock(return_value=[obj1, obj2])
+ self.manager.resource_class = Mock
+
+ # exactly one match case
+ found = self.manager.find(attr1="v1", attr2="v2", attr3="v3")
+ self.assertEqual(obj1, found)
+
+ # no match case
+ self.assertRaises(exceptions.NotFound, self.manager.find,
+ attr1="v2", attr2="v2", attr3="v3")
+
+ # multiple matches case
+ obj2.attr3 = "v3"
+ self.assertRaises(exceptions.NoUniqueMatch, self.manager.find,
+ attr1="v1", attr2="v2", attr3="v3")
+
+ def test_findall(self):
+ obj1 = Mock()
+ obj1.attr1 = "v1"
+ obj1.attr2 = "v2"
+ obj1.attr3 = "v3"
+
+ obj2 = Mock()
+ obj2.attr1 = "v1"
+ obj2.attr2 = "v2"
+
+ self.manager.list = Mock(return_value=[obj1, obj2])
+
+ found = self.manager.findall(attr1="v1", attr2="v2", attr3="v3")
+ self.assertEqual(1, len(found))
+ self.assertEqual(obj1, found[0])
+
+ found = self.manager.findall(attr1="v2", attr2="v2", attr3="v3")
+ self.assertEqual(0, len(found))
+
+ found = self.manager.findall(attr7="v1", attr2="v2")
+ self.assertEqual(0, len(found))
+
+ def test_list(self):
+ # this method is not yet implemented, exception expected
+ self.assertRaises(NotImplementedError, self.manager.list)
+
+
+class ResourceTest(TestCase):
+
+ def setUp(self):
+ super(ResourceTest, self).setUp()
+ self.orig___init__ = base.Resource.__init__
+
+ def tearDown(self):
+ super(ResourceTest, self).tearDown()
+ base.Resource.__init__ = self.orig___init__
+
+ def test___init__(self):
+ manager = Mock()
+ manager.write_to_completion_cache = Mock(return_value=None)
+
+ info_ = {}
+ robj = base.Resource(manager, info_)
+ self.assertEqual(0, manager.write_to_completion_cache.call_count)
+
+ info_ = {"id": "id-with-less-than-36-char"}
+ robj = base.Resource(manager, info_)
+ self.assertEqual(info_["id"], robj.id)
+ self.assertEqual(0, manager.write_to_completion_cache.call_count)
+
+ id_ = "id-with-36-char-"
+ for i in range(36 - len(id_)):
+ id_ = id_ + "-"
+ info_ = {"id": id_}
+ robj = base.Resource(manager, info_)
+ self.assertEqual(info_["id"], robj.id)
+ self.assertEqual(1, manager.write_to_completion_cache.call_count)
+
+ info_["name"] = "test-human-id"
+ # Resource.HUMAN_ID is False
+ robj = base.Resource(manager, info_)
+ self.assertEqual(info_["id"], robj.id)
+ self.assertEqual(None, robj.human_id)
+ self.assertEqual(2, manager.write_to_completion_cache.call_count)
+
+ # base.Resource.HUMAN_ID = True
+ info_["HUMAN_ID"] = True
+ robj = base.Resource(manager, info_)
+ self.assertEqual(info_["id"], robj.id)
+ self.assertEqual(info_["name"], robj.human_id)
+ self.assertEqual(4, manager.write_to_completion_cache.call_count)
+
+ def test_human_id(self):
+ manager = Mock()
+ manager.write_to_completion_cache = Mock(return_value=None)
+
+ info_ = {"name": "test-human-id"}
+ robj = base.Resource(manager, info_)
+ self.assertEqual(None, robj.human_id)
+
+ info_["HUMAN_ID"] = True
+ robj = base.Resource(manager, info_)
+ self.assertEqual(info_["name"], robj.human_id)
+ robj.name = "new-human-id"
+ self.assertEqual("new-human-id", robj.human_id)
+
+ def get_mock_resource_obj(self):
+ base.Resource.__init__ = Mock(return_value=None)
+ robj = base.Resource()
+ return robj
+
+ def test__add_details(self):
+ robj = self.get_mock_resource_obj()
+ info_ = {"name": "test-human-id", "test_attr": 5}
+ robj._add_details(info_)
+ self.assertEqual(info_["name"], robj.name)
+ self.assertEqual(info_["test_attr"], robj.test_attr)
+
+ def test___getattr__(self):
+ robj = self.get_mock_resource_obj()
+ info_ = {"name": "test-human-id", "test_attr": 5}
+ robj._add_details(info_)
+ self.assertEqual(info_["test_attr"], robj.__getattr__("test_attr"))
+
+ # TODO: looks like causing infinite recursive calls
+ #robj.__getattr__("test_non_exist_attr")
+
+ def test___repr__(self):
+ robj = self.get_mock_resource_obj()
+ info_ = {"name": "test-human-id", "test_attr": 5}
+ robj._add_details(info_)
+
+ expected = "<Resource name=test-human-id, test_attr=5>"
+ self.assertEqual(expected, robj.__repr__())
+
+ def test_get(self):
+ robj = self.get_mock_resource_obj()
+ manager = Mock()
+ manager.get = None
+
+ robj.manager = object()
+ robj.get()
+
+ manager = Mock()
+ robj.manager = Mock()
+
+ robj.id = "id"
+ new = Mock()
+ new._info = {"name": "test-human-id", "test_attr": 5}
+ robj.manager.get = Mock(return_value=new)
+ robj.get()
+ self.assertEqual("test-human-id", robj.name)
+ self.assertEqual(5, robj.test_attr)
+
+ def tes___eq__(self):
+ robj = self.get_mock_resource_obj()
+ other = base.Resource()
+
+ info_ = {"name": "test-human-id", "test_attr": 5}
+ robj._info = info_
+ other._info = {}
+ self.assertNotTrue(robj.__eq__(other))
+
+ robj._info = info_
+ self.assertTrue(robj.__eq__(other))
+
+ robj.id = "rid"
+ other.id = "oid"
+ self.assertNotTrue(robj.__eq__(other))
+
+ other.id = "rid"
+ self.assertTrue(robj.__eq__(other))
+
+ # not instance of the same class
+ other = Mock()
+ self.assertNotTrue(robj.__eq__(other))
+
+ def test_is_loaded(self):
+ robj = self.get_mock_resource_obj()
+ robj._loaded = True
+ self.assertTrue(robj.is_loaded())
+
+ robj._loaded = False
+ self.assertFalse(robj.is_loaded())
+
+ def test_set_loaded(self):
+ robj = self.get_mock_resource_obj()
+ robj.set_loaded(True)
+ self.assertTrue(robj._loaded)
+
+ robj.set_loaded(False)
+ self.assertFalse(robj._loaded)
diff --git a/troveclient/tests/test_client.py b/troveclient/tests/test_client.py
new file mode 100644
index 0000000..263316b
--- /dev/null
+++ b/troveclient/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 troveclient import client
+from troveclient import exceptions
+from troveclient 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 TroveHTTPClientTest(TestCase):
+
+ def setUp(self):
+ super(TroveHTTPClientTest, self).setUp()
+ self.orig__init = client.TroveHTTPClient.__init__
+ client.TroveHTTPClient.__init__ = Mock(return_value=None)
+ self.hc = client.TroveHTTPClient()
+ 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(TroveHTTPClientTest, self).tearDown()
+ client.TroveHTTPClient.__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.TroveHTTPClient.__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.TroveHTTPClient, user,
+ password, tenant, auth_url, service_name)
+
+ hc = client.TroveHTTPClient(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.TroveHTTPClient(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(Exception, 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(Exception,
+ 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.TroveHTTPClient.__init__
+ client.TroveHTTPClient.__init__ = Mock(return_value=None)
+ self.dbaas = client.Dbaas("user", "api-key")
+
+ def tearDown(self):
+ super(DbaasTest, self).tearDown()
+ client.TroveHTTPClient.__init__ = self.orig__init
+
+ def test___init__(self):
+ client.TroveHTTPClient.__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/troveclient/tests/test_common.py b/troveclient/tests/test_common.py
new file mode 100644
index 0000000..e1b488a
--- /dev/null
+++ b/troveclient/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 troveclient import common
+from troveclient 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(Exception,
+ 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('database', co.service_type)
+ self.assertEqual('trove', 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/troveclient/tests/test_instances.py b/troveclient/tests/test_instances.py
new file mode 100644
index 0000000..05e4019
--- /dev/null
+++ b/troveclient/tests/test_instances.py
@@ -0,0 +1,176 @@
+from testtools import TestCase
+from mock import Mock
+
+from troveclient import instances
+from troveclient 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(Exception, 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/troveclient/tests/test_limits.py b/troveclient/tests/test_limits.py
new file mode 100644
index 0000000..d40b646
--- /dev/null
+++ b/troveclient/tests/test_limits.py
@@ -0,0 +1,79 @@
+from testtools import TestCase
+from mock import Mock
+from troveclient import limits
+
+
+class LimitsTest(TestCase):
+ """
+ This class tests the calling code for the Limits API
+ """
+
+ def setUp(self):
+ super(LimitsTest, self).setUp()
+ self.limits = limits.Limits(Mock())
+ self.limits.api.client = Mock()
+
+ def tearDown(self):
+ super(LimitsTest, self).tearDown()
+
+ def test_list(self):
+ resp = Mock()
+ resp.status = 200
+ body = {"limits":
+ [
+ {'maxTotalInstances': 55,
+ 'verb': 'ABSOLUTE',
+ 'maxTotalVolumes': 100},
+ {'regex': '.*',
+ 'nextAvailable': '2011-07-21T18:17:06Z',
+ 'uri': '*',
+ 'value': 10,
+ 'verb': 'POST',
+ 'remaining': 2, 'unit': 'MINUTE'},
+ {'regex': '.*',
+ 'nextAvailable': '2011-07-21T18:17:06Z',
+ 'uri': '*',
+ 'value': 10,
+ 'verb': 'PUT',
+ 'remaining': 2,
+ 'unit': 'MINUTE'},
+ {'regex': '.*',
+ 'nextAvailable': '2011-07-21T18:17:06Z',
+ 'uri': '*',
+ 'value': 10,
+ 'verb': 'DELETE',
+ 'remaining': 2,
+ 'unit': 'MINUTE'},
+ {'regex': '.*',
+ 'nextAvailable': '2011-07-21T18:17:06Z',
+ 'uri': '*',
+ 'value': 10,
+ 'verb': 'GET',
+ 'remaining': 2, 'unit': 'MINUTE'}]}
+ response = (resp, body)
+
+ mock_get = Mock(return_value=response)
+ self.limits.api.client.get = mock_get
+ self.assertIsNotNone(self.limits.list())
+ mock_get.assert_called_once_with("/limits")
+
+ def test_list_errors(self):
+ status_list = [400, 401, 403, 404, 408, 409, 413, 500, 501]
+ for status_code in status_list:
+ self._check_error_response(status_code)
+
+ def _check_error_response(self, status_code):
+ RESPONSE_KEY = "limits"
+
+ resp = Mock()
+ resp.status = status_code
+ body = {RESPONSE_KEY: {
+ 'absolute': {},
+ 'rate': [
+ {'limit': []
+ }]}}
+ response = (resp, body)
+
+ mock_get = Mock(return_value=response)
+ self.limits.api.client.get = mock_get
+ self.assertRaises(Exception, self.limits.list)
diff --git a/troveclient/tests/test_management.py b/troveclient/tests/test_management.py
new file mode 100644
index 0000000..c04e216
--- /dev/null
+++ b/troveclient/tests/test_management.py
@@ -0,0 +1,144 @@
+from testtools import TestCase
+from mock import Mock
+
+from troveclient import management
+from troveclient 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(Exception, 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_migrate_to_host(self):
+ hostname = 'hostname2'
+ self._mock_action()
+ self.management.migrate(1, host=hostname)
+ self.assertEqual(1, self.management._action.call_count)
+ self.assertEqual({'migrate': {'host': hostname}}, 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_)
+
+ def test_reset_task_status(self):
+ self._mock_action()
+ self.management.reset_task_status(1)
+ self.assertEqual(1, self.management._action.call_count)
+ self.assertEqual({'reset-task-status': {}}, self.body_)
diff --git a/troveclient/tests/test_secgroups.py b/troveclient/tests/test_secgroups.py
new file mode 100644
index 0000000..779d01e
--- /dev/null
+++ b/troveclient/tests/test_secgroups.py
@@ -0,0 +1,102 @@
+from testtools import TestCase
+from mock import Mock
+
+from troveclient import security_groups
+from troveclient import base
+
+"""
+Unit tests for security_groups.py
+"""
+
+
+class SecGroupTest(TestCase):
+
+ def setUp(self):
+ super(SecGroupTest, self).setUp()
+ self.orig__init = security_groups.SecurityGroup.__init__
+ security_groups.SecurityGroup.__init__ = Mock(return_value=None)
+ self.security_group = security_groups.SecurityGroup()
+ self.security_groups = security_groups.SecurityGroups(1)
+
+ def tearDown(self):
+ super(SecGroupTest, self).tearDown()
+ security_groups.SecurityGroup.__init__ = self.orig__init
+
+ def test___repr__(self):
+ self.security_group.name = "security_group-1"
+ self.assertEqual('<SecurityGroup: security_group-1>',
+ self.security_group.__repr__())
+
+ def test_list(self):
+ sec_group_list = ['secgroup1', 'secgroup2']
+ self.security_groups.list = Mock(return_value=sec_group_list)
+ self.assertEqual(sec_group_list, self.security_groups.list())
+
+ def test_get(self):
+ def side_effect_func(path, inst):
+ return path, inst
+
+ self.security_groups._get = Mock(side_effect=side_effect_func)
+ self.security_group.id = 1
+ self.assertEqual(('/security-groups/1', 'security_group'),
+ self.security_groups.get(self.security_group))
+
+
+class SecGroupRuleTest(TestCase):
+
+ def setUp(self):
+ super(SecGroupRuleTest, self).setUp()
+ self.orig__init = security_groups.SecurityGroupRule.__init__
+ security_groups.SecurityGroupRule.__init__ = Mock(return_value=None)
+ security_groups.SecurityGroupRules.__init__ = Mock(return_value=None)
+ self.security_group_rule = security_groups.SecurityGroupRule()
+ self.security_group_rules = security_groups.SecurityGroupRules()
+
+ def tearDown(self):
+ super(SecGroupRuleTest, self).tearDown()
+ security_groups.SecurityGroupRule.__init__ = self.orig__init
+
+ def test___repr__(self):
+ self.security_group_rule.group_id = 1
+ self.security_group_rule.protocol = "tcp"
+ self.security_group_rule.from_port = 80
+ self.security_group_rule.to_port = 80
+ self.security_group_rule.cidr = "0.0.0.0//0"
+ representation = \
+ "<SecurityGroupRule: ( \
+ Security Group id: %d, \
+ Protocol: %s, \
+ From_Port: %d, \
+ To_Port: %d, \
+ CIDR: %s )>" % (1, "tcp", 80, 80, "0.0.0.0//0")
+
+ self.assertEqual(representation,
+ self.security_group_rule.__repr__())
+
+ def test_create(self):
+ def side_effect_func(path, body, inst):
+ return path, body, inst
+
+ self.security_group_rules._create = Mock(side_effect=side_effect_func)
+ p, b, i = self.security_group_rules.create(1, "tcp",
+ 80, 80, "0.0.0.0//0")
+ self.assertEqual("/security-group-rules", p)
+ self.assertEqual("security_group_rule", i)
+ self.assertEqual(1, b["security_group_rule"]["group_id"])
+ self.assertEqual("tcp", b["security_group_rule"]["protocol"])
+ self.assertEqual(80, b["security_group_rule"]["from_port"])
+ self.assertEqual(80, b["security_group_rule"]["to_port"])
+ self.assertEqual("0.0.0.0//0", b["security_group_rule"]["cidr"])
+
+ def test_delete(self):
+ resp = Mock()
+ resp.status = 200
+ body = None
+ self.security_group_rules.api = Mock()
+ self.security_group_rules.api.client = Mock()
+ self.security_group_rules.api.client.delete = \
+ Mock(return_value=(resp, body))
+ self.security_group_rules.delete(self.id)
+ resp.status = 500
+ self.assertRaises(Exception, self.security_group_rules.delete,
+ self.id)
diff --git a/troveclient/tests/test_users.py b/troveclient/tests/test_users.py
new file mode 100644
index 0000000..0fc32f6
--- /dev/null
+++ b/troveclient/tests/test_users.py
@@ -0,0 +1,126 @@
+from testtools import TestCase
+from mock import Mock
+
+from troveclient import users
+from troveclient 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 _build_fake_user(self, name, hostname=None, password=None,
+ databases=None):
+ return {'name': name,
+ 'password': password if password else 'password',
+ 'host': hostname,
+ 'databases': databases if databases else [],
+ }
+
+ def test_create(self):
+ self.users.api.client.post = self._get_mock_method()
+ self._resp.status = 200
+ user = self._build_fake_user('user1')
+
+ self.users.create(23, [user])
+ self.assertEqual('/instances/23/users', self._url)
+ self.assertEqual({"users": [user]}, self._body)
+
+ # Even if host isn't supplied originally,
+ # the default is supplied.
+ del user['host']
+ self.users.create(23, [user])
+ self.assertEqual('/instances/23/users', self._url)
+ user['host'] = '%'
+ self.assertEqual({"users": [user]}, self._body)
+
+ # If host is supplied, of course it's put into the body.
+ user['host'] = '127.0.0.1'
+ self.users.create(23, [user])
+ self.assertEqual({"users": [user]}, self._body)
+
+ # Make sure that response of 400 is recognized as an error.
+ user['host'] = '%'
+ self._resp.status = 400
+ self.assertRaises(Exception, self.users.create, 12, [user])
+
+ 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))
diff --git a/troveclient/tests/test_utils.py b/troveclient/tests/test_utils.py
new file mode 100644
index 0000000..12ee9d0
--- /dev/null
+++ b/troveclient/tests/test_utils.py
@@ -0,0 +1,41 @@
+import os
+from testtools import TestCase
+from troveclient import utils
+from troveclient import versions
+
+
+class UtilsTest(TestCase):
+
+ def test_add_hookable_mixin(self):
+ def func():
+ pass
+
+ hook_type = "hook_type"
+ mixin = utils.HookableMixin()
+ mixin.add_hook(hook_type, func)
+ self.assertTrue(hook_type in mixin._hooks_map)
+ self.assertTrue(func in mixin._hooks_map[hook_type])
+
+ def test_run_hookable_mixin(self):
+ def func():
+ pass
+
+ hook_type = "hook_type"
+ mixin = utils.HookableMixin()
+ mixin.add_hook(hook_type, func)
+ mixin.run_hooks(hook_type)
+
+ def test_environment(self):
+ self.assertEqual('', utils.env())
+ self.assertEqual('passing', utils.env(default='passing'))
+
+ os.environ['test_abc'] = 'passing'
+ self.assertEqual('passing', utils.env('test_abc'))
+ self.assertEqual('', utils.env('test_abcd'))
+
+ def test_slugify(self):
+ import unicodedata
+
+ self.assertEqual('not_unicode', utils.slugify('not_unicode'))
+ self.assertEqual('unicode', utils.slugify(unicode('unicode')))
+ self.assertEqual('slugify-test', utils.slugify('SLUGIFY% test!'))
diff --git a/troveclient/tests/test_xml.py b/troveclient/tests/test_xml.py
new file mode 100644
index 0000000..96490ce
--- /dev/null
+++ b/troveclient/tests/test_xml.py
@@ -0,0 +1,241 @@
+from testtools import TestCase
+from lxml import etree
+from troveclient import xml
+
+
+class XmlTest(TestCase):
+
+ ELEMENT = '''
+ <instances>
+ <instance>
+ <flavor>
+ <links>
+ </links>
+ <value value="5"/>
+ </flavor>
+ </instance>
+ </instances>
+ '''
+ ROOT = etree.fromstring(ELEMENT)
+
+ JSON = {'instances':
+ {'instances': ['1', '2', '3']}, 'dummy': {'dict': True}}
+
+ def test_element_ancestors_match_list(self):
+ # Test normal operation:
+ self.assertTrue(xml.element_ancestors_match_list(self.ROOT[0][0],
+ ['instance',
+ 'instances']))
+
+ # Test itr_elem is None:
+ self.assertTrue(xml.element_ancestors_match_list(self.ROOT,
+ ['instances']))
+
+ # Test that the first parent element does not match the first list
+ # element:
+ self.assertFalse(xml.element_ancestors_match_list(self.ROOT[0][0],
+ ['instances',
+ 'instance']))
+
+ def test_populate_element_from_dict(self):
+ # Test populate_element_from_dict with a None in the data
+ ele = '''
+ <instance>
+ <volume>
+ <value size="5"/>
+ </volume>
+ </instance>
+ '''
+ rt = etree.fromstring(ele)
+
+ self.assertEqual(None, xml.populate_element_from_dict(rt,
+ {'size': None}))
+
+ def test_element_must_be_list(self):
+ # Test for when name isn't in the dictionary
+ self.assertFalse(xml.element_must_be_list(self.ROOT, "not_in_list"))
+
+ # Test when name is in the dictionary but list is empty
+ self.assertTrue(xml.element_must_be_list(self.ROOT, "accounts"))
+
+ # Test when name is in the dictionary but list is not empty
+ self.assertTrue(xml.element_must_be_list(self.ROOT[0][0][0], "links"))
+
+ def test_element_to_json(self):
+ # Test when element must be list:
+ self.assertEqual([{'flavor': {'links': [], 'value': {'value': '5'}}}],
+ xml.element_to_json("accounts", self.ROOT))
+
+ # Test when element must not be list:
+ exp = {'instance': {'flavor': {'links': [], 'value': {'value': '5'}}}}
+ self.assertEqual(exp, xml.element_to_json("not_in_list", self.ROOT))
+
+ def test_root_element_to_json(self):
+ # Test when element must be list:
+ exp = ([{'flavor': {'links': [], 'value': {'value': '5'}}}], None)
+ self.assertEqual(exp, xml.root_element_to_json("accounts", self.ROOT))
+
+ # Test when element must not be list:
+ exp = {'instance': {'flavor': {'links': [], 'value': {'value': '5'}}}}
+ self.assertEqual((exp, None),
+ xml.root_element_to_json("not_in_list", self.ROOT))
+
+ # Test rootEnabled True:
+ t_element = etree.fromstring('''<rootEnabled> True </rootEnabled>''')
+ self.assertEqual((True, None),
+ xml.root_element_to_json("rootEnabled", t_element))
+
+ # Test rootEnabled False:
+ f_element = etree.fromstring('''<rootEnabled> False </rootEnabled>''')
+ self.assertEqual((False, None),
+ xml.root_element_to_json("rootEnabled", f_element))
+
+ def test_element_to_list(self):
+ # Test w/ no child elements
+ self.assertEqual([], xml.element_to_list(self.ROOT[0][0][0]))
+
+ # Test w/ no child elements and check_for_links = True
+ self.assertEqual(([], None),
+ xml.element_to_list(self.ROOT[0][0][0],
+ check_for_links=True))
+
+ # Test w/ child elements
+ self.assertEqual([{}, {'value': '5'}],
+ xml.element_to_list(self.ROOT[0][0]))
+
+ # Test w/ child elements and check_for_links = True
+ self.assertEqual(([{'value': '5'}], []),
+ xml.element_to_list(self.ROOT[0][0],
+ check_for_links=True))
+
+ def test_element_to_dict(self):
+ # Test when there is not a None
+ exp = {'instance': {'flavor': {'links': [], 'value': {'value': '5'}}}}
+ self.assertEqual(exp, xml.element_to_dict(self.ROOT))
+
+ # Test when there is a None
+ element = '''
+ <server>
+ None
+ </server>
+ '''
+ rt = etree.fromstring(element)
+ self.assertEqual(None, xml.element_to_dict(rt))
+
+ def test_standarize_json(self):
+ xml.standardize_json_lists(self.JSON)
+ self.assertEqual({'instances': ['1', '2', '3'],
+ 'dummy': {'dict': True}}, self.JSON)
+
+ def test_normalize_tag(self):
+ ELEMENT_NS = '''
+ <instances xmlns="http://www.w3.org/1999/xhtml">
+ <instance>
+ <flavor>
+ <links>
+ </links>
+ <value value="5"/>
+ </flavor>
+ </instance>
+ </instances>
+ '''
+ ROOT_NS = etree.fromstring(ELEMENT_NS)
+
+ # Test normalizing without namespace info
+ self.assertEqual('instances', xml.normalize_tag(self.ROOT))
+
+ # Test normalizing with namespace info
+ self.assertEqual('instances', xml.normalize_tag(ROOT_NS))
+
+ def test_create_root_xml_element(self):
+ # Test creating when name is not in REQUEST_AS_LIST
+ element = xml.create_root_xml_element("root", {"root": "value"})
+ exp = '<root xmlns="http://docs.openstack.org/database/api/v1.0" ' \
+ 'root="value"/>'
+ self.assertEqual(exp, etree.tostring(element))
+
+ # Test creating when name is in REQUEST_AS_LIST
+ element = xml.create_root_xml_element("users", [])
+ exp = '<users xmlns="http://docs.openstack.org/database/api/v1.0"/>'
+ self.assertEqual(exp, etree.tostring(element))
+
+ def test_creating_subelements(self):
+ # Test creating a subelement as a dictionary
+ element = xml.create_root_xml_element("root", {"root": 5})
+ xml.create_subelement(element, "subelement", {"subelement": "value"})
+ exp = '<root xmlns="http://docs.openstack.org/database/api/v1.0" ' \
+ 'root="5"><subelement subelement="value"/></root>'
+ self.assertEqual(exp, etree.tostring(element))
+
+ # Test creating a subelement as a list
+ element = xml.create_root_xml_element("root",
+ {"root": {"value": "nested"}})
+ xml.create_subelement(element, "subelement", [{"subelement": "value"}])
+ exp = '<root xmlns="http://docs.openstack.org/database/api/v1.0">' \
+ '<root value="nested"/><subelement><subelement subelement=' \
+ '"value"/></subelement></root>'
+ self.assertEqual(exp, etree.tostring(element))
+
+ # Test creating a subelement as a string (should raise TypeError)
+ element = xml.create_root_xml_element("root", {"root": "value"})
+ try:
+ xml.create_subelement(element, "subelement", ["value"])
+ self.fail("TypeError exception expected")
+ except TypeError:
+ pass
+
+ def test_modify_response_types(self):
+ TYPE_MAP = {
+ "Int": int,
+ "Bool": bool
+ }
+ #Is a string True
+ self.assertEqual(True, xml.modify_response_types('True', TYPE_MAP))
+
+ #Is a string False
+ self.assertEqual(False, xml.modify_response_types('False', TYPE_MAP))
+
+ #Is a dict
+ test_dict = {"Int": "5"}
+ test_dict = xml.modify_response_types(test_dict, TYPE_MAP)
+ self.assertEqual(int, test_dict["Int"].__class__)
+
+ #Is a list
+ test_list = {"a_list": [{"Int": "5"}, {"Str": "A"}]}
+ test_list = xml.modify_response_types(test_list["a_list"], TYPE_MAP)
+ self.assertEqual([{'Int': 5}, {'Str': 'A'}], test_list)
+
+ def test_trovexmlclient(self):
+ from troveclient import exceptions
+
+ client = xml.TroveXmlClient("user", "password", "tenant",
+ "auth_url", "service_name",
+ auth_strategy="fake")
+ request = {'headers': {}}
+
+ # Test morph_request, no body
+ client.morph_request(request)
+ self.assertEqual('application/xml', request['headers']['Accept'])
+ self.assertEqual('application/xml', request['headers']['Content-Type'])
+
+ # Test morph_request, with body
+ request['body'] = {'root': {'test': 'test'}}
+ client.morph_request(request)
+ body = '<root xmlns="http://docs.openstack.org/database/api/v1.0" ' \
+ 'test="test"/>\n'
+ exp = {'body': body,
+ 'headers': {'Content-Type': 'application/xml',
+ 'Accept': 'application/xml'}}
+ self.assertEqual(exp, request)
+
+ # Test morph_response_body
+ request = "<users><links><user href='value'/></links></users>"
+ result = client.morph_response_body(request)
+ self.assertEqual({'users': [], 'links': [{'href': 'value'}]}, result)
+
+ # Test morph_response_body with improper input
+ try:
+ client.morph_response_body("value")
+ self.fail("ResponseFormatError exception expected")
+ except exceptions.ResponseFormatError:
+ pass
View Lorry Controller Status