Add unit test for action_manager and session_manager

Change-Id: Icbd505f50d882078df1e2f26bc539cd52c111d5c
This commit is contained in:
gengchc2 2018-12-26 16:47:35 -08:00
parent 81198a782a
commit 0f7c37c776
1 changed files with 358 additions and 0 deletions

View File

@ -1141,3 +1141,361 @@ class TestElasticSearchEngine_job(unittest.TestCase, ElasticSearchDB):
job_id=common.fake_job_0_job_id,
doc=common.get_fake_job_0())
self.assertEqual(3, res)
class TestElasticSearchEngine_action(unittest.TestCase, ElasticSearchDB):
@patch('freezer_api.storage.elasticv2.logging')
@patch('freezer_api.storage.elasticv2.elasticsearch')
def setUp(self, mock_elasticsearch, mock_logging):
backend = 'elasticsearch'
grp = cfg.OptGroup(backend)
CONF.register_group(grp)
CONF.register_opts(self._ES_OPTS, group=backend)
mock_elasticsearch.Elasticsearch.return_value = mock.Mock()
kwargs = {'hosts': 'http://elasticservaddr:1997'}
self.eng = elastic.ElasticSearchEngineV2(backend=backend)
self.eng.init(index='freezer', **kwargs)
self.eng.action_manager = mock.Mock()
def test_get_action_userid_and_action_id_return_doc(self):
self.eng.action_manager.get.return_value = common.get_fake_action_0()
res = self.eng.get_action(project_id='tecs',
user_id=common.fake_action_0['user_id'],
action_id=common.fake_action_0['action_id'])
self.assertEqual(common.fake_action_0, res)
self.eng.action_manager.get.assert_called_with(
project_id='tecs',
user_id=common.fake_action_0['user_id'],
doc_id=common.fake_action_0['action_id'])
def test_get_action_userid_and_action_id_return_none(self):
self.eng.action_manager.get.return_value = None
res = self.eng.get_action(project_id='tecs',
user_id=common.fake_action_0['user_id'],
action_id=common.fake_action_0['action_id'])
self.assertIsNone(res)
self.eng.action_manager.get.assert_called_with(
project_id='tecs',
user_id=common.fake_action_0['user_id'],
doc_id=common.fake_action_0['action_id'])
def test_get_action_with_userid_and_search_return_list(self):
self.eng.action_manager.search.return_value = \
[common.fake_action_0, common.fake_action_0]
my_search = {'match': [{'some_field': 'some text'},
{'description': 'some other text'}]}
res = self.eng.search_action(project_id='tecs',
user_id=common.fake_action_0['user_id'],
offset=6, limit=15,
search=my_search)
self.assertEqual([common.fake_action_0, common.fake_action_0], res)
self.eng.action_manager.search.assert_called_with(
project_id='tecs',
user_id=common.fake_action_0['user_id'],
search=my_search,
limit=15, offset=6)
def test_get_action_with_userid_and_search_return_empty_list(self):
self.eng.action_manager.search.return_value = []
my_search = {'match': [{'some_field': 'some text'},
{'description': 'some other text'}]}
res = self.eng.search_action(project_id='tecs',
user_id=common.fake_action_0['user_id'],
offset=6, limit=15,
search=my_search)
self.assertEqual([], res)
self.eng.action_manager.search.assert_called_with(
project_id='tecs',
user_id=common.fake_action_0['user_id'],
search=my_search,
limit=15, offset=6)
@patch('freezer_api.common.elasticv2_utils.ActionDoc')
def test_add_action_ok(self, mock_actiondoc):
mock_actiondoc.create.return_value = common.get_fake_action_0()
self.eng.action_manager.insert.return_value = (True, 1)
res = self.eng.add_action(project_id='tecs',
user_id=common.fake_action_0['user_id'],
doc=common.get_fake_action_0())
self.assertEqual(common.fake_action_0['action_id'], res)
self.eng.action_manager.insert.assert_called_with(
common.fake_action_0, common.fake_action_0['action_id']
)
def test_add_action_raises_StorageEngineError_when_manager_insert_raises(
self):
self.eng.action_manager.get.return_value = None
self.eng.action_manager.insert.side_effect = (
exceptions.StorageEngineError('regular test failure')
)
self.assertRaises(exceptions.StorageEngineError, self.eng.add_action,
project_id='tecs',
user_id=common.fake_action_0['user_id'],
doc=common.get_fake_action_0())
def test_delete_action_ok(self):
self.eng.action_manager.delete.return_value = common.fake_action_0[
'action_id']
res = self.eng.delete_action(
project_id='tecs',
user_id=common.fake_action_0['action_id'],
action_id=common.fake_action_0['action_id'])
self.assertEqual(common.fake_action_0['action_id'], res)
def test_delete_client_raises_StorageEngineError_when_es_delete_raises(
self):
self.eng.action_manager.delete.side_effect = (
exceptions.StorageEngineError()
)
self.assertRaises(exceptions.StorageEngineError,
self.eng.delete_action,
project_id='tecs',
user_id=common.fake_action_0['action_id'],
action_id=common.fake_action_0['action_id'])
def test_update_action_raises_DocumentNotFound_when_doc_not_exists(self):
self.eng.action_manager.get.side_effect = exceptions.DocumentNotFound(
'regular test failure')
patch = {'action_id': 'black_milk'}
self.assertRaises(exceptions.DocumentNotFound, self.eng.update_action,
project_id='tecs',
user_id=common.fake_action_0['action_id'],
action_id=common.fake_action_0['action_id'],
patch_doc=patch)
def test_update_action_raises_DocumentNotFound_when_update_raises(
self):
self.eng.action_manager.get.return_value = common.get_fake_action_0()
patch = {'action_id': 'black_milk'}
self.eng.action_manager.update.side_effect = (
exceptions.DocumentNotFound('regular test failure')
)
self.assertRaises(exceptions.DocumentNotFound, self.eng.update_action,
project_id='tecs',
user_id=common.fake_action_0['action_id'],
action_id=common.fake_action_0['action_id'],
patch_doc=patch)
def test_update_action_returns_new_doc_version(self):
self.eng.action_manager.get.return_value = common.get_fake_action_0()
patch = {'action_id': 'group_four'}
self.eng.action_manager.update.return_value = 11
res = self.eng.update_action(
project_id='tecs',
user_id=common.fake_action_0['action_id'],
action_id=common.fake_action_0['action_id'],
patch_doc=patch)
self.assertEqual(11, res)
def test_replace_action_raises_AccessForbidden_when_action_manager_raises(
self):
self.eng.action_manager.get.side_effect = exceptions.AccessForbidden(
'regular test failure')
self.eng.action_manager.insert.return_value = (True, 3)
self.assertRaises(exceptions.AccessForbidden, self.eng.replace_action,
project_id='tecs',
user_id=common.fake_action_0['action_id'],
action_id=common.fake_action_0['action_id'],
doc=common.get_fake_action_0())
# def test_replace_action_returns_ok_when_doc_is_new(self):
# self.eng.action_manager.get.side_effect = (
# exceptions.DocumentNotFound('regular test failure')
# )
# self.eng.action_manager.insert.return_value = (True, 1)
# res = self.eng.replace_action(
# user_id=common.fake_action_0['action_id'],
# action_id=common.fake_action_0['action_id'],
# doc=common.get_fake_action_0())
# self.assertEqual(res, 1)
#
# def test_replace_action_returns_version_1_when_doc_is_overwritten(
# self):
# self.eng.action_manager.get.return_value = common.get_fake_action_0()
# self.eng.action_manager.insert.return_value = (False, 3)
# res = self.eng.replace_action(
# user_id=common.fake_action_0['action_id'],
# action_id=common.fake_action_0['action_id'],
# doc=common.get_fake_action_0())
# self.assertEqual(res, 3)
class TestElasticSearchEngine_session(unittest.TestCase, ElasticSearchDB):
@patch('freezer_api.storage.elasticv2.logging')
@patch('freezer_api.storage.elasticv2.elasticsearch')
def setUp(self, mock_elasticsearch, mock_logging):
backend = 'elasticsearch'
grp = cfg.OptGroup(backend)
CONF.register_group(grp)
CONF.register_opts(self._ES_OPTS, group=backend)
mock_elasticsearch.Elasticsearch.return_value = mock.Mock()
kwargs = {'hosts': 'http://elasticservaddr:1997'}
self.eng = elastic.ElasticSearchEngineV2(backend=backend)
self.eng.init(index='freezer', **kwargs)
self.eng.session_manager = mock.Mock()
def test_get_session_userid_and_session_id_return_doc(self):
self.eng.session_manager.get.return_value = common.get_fake_session_0()
res = self.eng.get_session(project_id='tecs',
user_id=common.fake_session_0['user_id'],
session_id=common.fake_session_0[
'session_id'])
self.assertEqual(common.fake_session_0, res)
self.eng.session_manager.get.assert_called_with(
project_id='tecs',
user_id=common.fake_session_0['user_id'],
doc_id=common.fake_session_0['session_id'])
def test_get_session_userid_and_session_id_return_none(self):
self.eng.session_manager.get.return_value = None
res = self.eng.get_session(project_id='tecs',
user_id=common.fake_session_0['user_id'],
session_id=common.fake_session_0[
'session_id'])
self.assertIsNone(res)
self.eng.session_manager.get.assert_called_with(
project_id='tecs',
user_id=common.fake_session_0['user_id'],
doc_id=common.fake_session_0['session_id'])
def test_get_session_with_userid_and_search_return_list(self):
self.eng.session_manager.search.return_value = \
[common.fake_session_0, common.fake_session_0]
my_search = {'match': [{'some_field': 'some text'},
{'description': 'some other text'}]}
res = self.eng.search_session(
project_id='tecs',
user_id=common.fake_session_0['user_id'],
offset=6, limit=15,
search=my_search)
self.assertEqual([common.fake_session_0, common.fake_session_0], res)
self.eng.session_manager.search.assert_called_with(
project_id='tecs',
user_id=common.fake_session_0['user_id'],
search=my_search,
limit=15, offset=6)
def test_get_session_with_userid_and_search_return_empty_list(self):
self.eng.session_manager.search.return_value = []
my_search = {'match': [{'some_field': 'some text'},
{'description': 'some other text'}]}
res = self.eng.search_session(
project_id='tecs',
user_id=common.fake_session_0['user_id'],
offset=6, limit=15,
search=my_search)
self.assertEqual([], res)
self.eng.session_manager.search.assert_called_with(
project_id='tecs',
user_id=common.fake_session_0['user_id'],
search=my_search,
limit=15, offset=6)
@patch('freezer_api.common.elasticv2_utils.SessionDoc')
def test_add_session_ok(self, mock_sessiondoc):
mock_sessiondoc.create.return_value = common.get_fake_session_0()
self.eng.session_manager.insert.return_value = (True, 1)
res = self.eng.add_session(project_id='tecs',
user_id=common.fake_session_0['user_id'],
doc=common.get_fake_session_0())
self.assertEqual(common.fake_session_0['session_id'], res)
self.eng.session_manager.insert.assert_called_with(
common.fake_session_0,
common.fake_session_0['session_id'])
def test_add_session_raises_StorageEngineError_when_manager_insert_raises(
self):
self.eng.session_manager.get.return_value = None
self.eng.session_manager.insert.side_effect = (
exceptions.StorageEngineError('regular test failure')
)
self.assertRaises(exceptions.StorageEngineError, self.eng.add_session,
project_id='tecs',
user_id=common.fake_session_0['user_id'],
doc=common.get_fake_session_0())
def test_delete_session_ok(self):
self.eng.session_manager.delete.return_value = common.fake_session_0[
'session_id']
res = self.eng.delete_session(
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'])
self.assertEqual(common.fake_session_0['session_id'], res)
def test_delete_client_raises_StorageEngineError_when_es_delete_raises(
self):
self.eng.session_manager.delete.side_effect = (
exceptions.StorageEngineError()
)
self.assertRaises(exceptions.StorageEngineError,
self.eng.delete_session,
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'])
def test_update_session_raises_DocumentNotFound_when_doc_not_exists(self):
self.eng.session_manager.get.side_effect = exceptions.DocumentNotFound(
'regular test failure')
patch = {'session_id': 'black_milk'}
self.assertRaises(exceptions.DocumentNotFound, self.eng.update_session,
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'],
patch_doc=patch)
def test_update_session_raises_DocumentNotFound_when_update_raises(
self):
self.eng.session_manager.get.return_value = common.get_fake_session_0()
patch = {'session_id': 'black_milk'}
self.eng.session_manager.update.side_effect = (
exceptions.DocumentNotFound('regular test failure')
)
self.assertRaises(exceptions.DocumentNotFound, self.eng.update_session,
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'],
patch_doc=patch)
def test_update_session_returns_new_doc_version(self):
self.eng.session_manager.get.return_value = common.get_fake_session_0()
patch = {'session_id': 'group_four'}
self.eng.session_manager.update.return_value = 11
res = self.eng.update_session(
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'],
patch_doc=patch)
self.assertEqual(11, res)
def test_replace_session_raises_AccessForbidden_when_session_manager_raise(
self):
self.eng.session_manager.get.side_effect = exceptions.AccessForbidden(
'regular test failure')
self.eng.session_manager.insert.return_value = (True, 3)
self.assertRaises(exceptions.AccessForbidden, self.eng.replace_session,
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'],
doc=common.get_fake_session_0())
def test_replace_session_returns_ok_when_doc_is_new(self):
self.eng.session_manager.get.side_effect = exceptions.DocumentNotFound(
'regular test failure')
self.eng.session_manager.insert.return_value = (True, 1)
res = self.eng.replace_session(
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'],
doc=common.get_fake_session_0())
self.assertEqual(1, res)
def test_replace_session_returns_version_1_when_doc_is_overwritten(self):
self.eng.session_manager.get.return_value = common.get_fake_session_0()
self.eng.session_manager.insert.return_value = (False, 3)
res = self.eng.replace_session(
project_id='tecs',
user_id=common.fake_session_0['session_id'],
session_id=common.fake_session_0['session_id'],
doc=common.get_fake_session_0())
self.assertEqual(3, res)