diff --git a/tests/test_base.py b/tests/test_base.py new file mode 100644 index 00000000..a9ab5cdc --- /dev/null +++ b/tests/test_base.py @@ -0,0 +1,447 @@ +import contextlib +import os + +from testtools import TestCase +from mock import Mock + +from reddwarfclient import base +from reddwarfclient import exceptions +from reddwarfclient 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 = "" + 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)