From 1977c7e9d300ff23e766b6d21cb4e24e1c5a570f Mon Sep 17 00:00:00 2001 From: gengchc2 Date: Mon, 24 Dec 2018 18:45:28 -0800 Subject: [PATCH] Add raise unit test for get and add tuple in sqlalchemy Change-Id: I8c9f05a370c42fe4b547b45b53535dbb51bb9cd6 --- freezer_api/db/sqlalchemy/api.py | 5 -- .../unit/sqlalchemy/v1/test_api_common.py | 57 ++++++++++++++++ .../unit/sqlalchemy/v2/test_api_common.py | 65 +++++++++++++++++++ 3 files changed, 122 insertions(+), 5 deletions(-) diff --git a/freezer_api/db/sqlalchemy/api.py b/freezer_api/db/sqlalchemy/api.py index 42089b0c..591acc90 100644 --- a/freezer_api/db/sqlalchemy/api.py +++ b/freezer_api/db/sqlalchemy/api.py @@ -159,10 +159,8 @@ def delete_tuple(tablename, user_id, tuple_id, project_id=None): '{0} not found in Table {1}'. format(tuple_id, tablename)) except Exception as e: - session.close() raise freezer_api_exc.StorageEngineError( message='Mysql operation failed {0}'.format(e)) - session.close() return tuple_id @@ -186,12 +184,9 @@ def add_tuple(tuple): try: tuple.save(session=session) except Exception as e: - session.close() raise freezer_api_exc.StorageEngineError( message='Mysql operation failed {0}'.format(e)) - session.close() - def update_tuple(tablename, user_id, tuple_id, tuple_values, project_id=None): diff --git a/freezer_api/tests/unit/sqlalchemy/v1/test_api_common.py b/freezer_api/tests/unit/sqlalchemy/v1/test_api_common.py index d297450b..54e0b048 100644 --- a/freezer_api/tests/unit/sqlalchemy/v1/test_api_common.py +++ b/freezer_api/tests/unit/sqlalchemy/v1/test_api_common.py @@ -18,6 +18,8 @@ """Tests for manipulating job via the DB API""" import copy +import mock +from mock import patch from oslo_config import cfg from freezer_api.common import exceptions as freezer_api_exc @@ -89,3 +91,58 @@ class ApiTestCase(base.DbTestCase): raise freezer_api_exc.StorageEngineError( message='sqlalchemy operation failed {0}'.format(e)) session.close() + + @ patch('oslo_db.sqlalchemy.utils.model_query') + def test_raises_delete_tuple(self, mock_model_query): + mock_model_query.side_effect = Exception('regular test failure') + self.assertRaises(freezer_api_exc.StorageEngineError, + api.delete_tuple, models.Job, self.fake_user_id, + self.fake_job_id) + + def test_delete_tuple(self): + job_doc1 = copy.deepcopy(self.fake_job_0) + job_doc2 = copy.deepcopy(self.fake_job_0) + self.dbapi.add_job(user_id=self.fake_user_id, + doc=job_doc1) + job_id2 = self.dbapi.add_job(user_id=self.fake_user_id, + doc=job_doc2) + + result = self.dbapi.delete_job(user_id=self.fake_user_id, + job_id=job_id2) + session = api.get_db_session() + with session.begin(): + try: + query = api.model_query(session, models.Job, + read_deleted='no') + query = query.filter_by(user_id=self.fake_user_id) + result = query.all() + self.assertEqual(len(result), 1) + + query = api.model_query(session, models.Job, + read_deleted='only') + query = query.filter_by(user_id=self.fake_user_id) + result = query.all() + self.assertEqual(len(result), 1) + + query = api.model_query(session, models.Job, + read_deleted='yes') + query = query.filter_by(user_id=self.fake_user_id) + result = query.all() + self.assertEqual(len(result), 2) + except Exception as e: + raise freezer_api_exc.StorageEngineError( + message='sqlalchemy operation failed {0}'.format(e)) + session.close() + + @patch('oslo_db.sqlalchemy.utils.model_query') + def test_raises_get_tuple(self, mock_model_query): + mock_model_query.side_effect = Exception('regular test failure') + self.assertRaises(freezer_api_exc.StorageEngineError, + api.get_tuple, models.Job, self.fake_user_id, + self.fake_job_id) + + def test_raises_add_tuple(self): + mock_tuple = mock.MagicMock() + mock_tuple.save.side_effect = Exception('regular test failure') + self.assertRaises(freezer_api_exc.StorageEngineError, + api.add_tuple, mock_tuple) diff --git a/freezer_api/tests/unit/sqlalchemy/v2/test_api_common.py b/freezer_api/tests/unit/sqlalchemy/v2/test_api_common.py index 0b8c6f0f..9c78fc6a 100644 --- a/freezer_api/tests/unit/sqlalchemy/v2/test_api_common.py +++ b/freezer_api/tests/unit/sqlalchemy/v2/test_api_common.py @@ -18,6 +18,8 @@ """Tests for manipulating job via the DB API""" import copy +import mock +from mock import patch from freezer_api.common import exceptions as freezer_api_exc from freezer_api.db.sqlalchemy import api @@ -89,3 +91,66 @@ class ApiTestCase(base.DbTestCase): raise freezer_api_exc.StorageEngineError( message='sqlalchemy operation failed {0}'.format(e)) session.close() + + @patch('oslo_db.sqlalchemy.utils.model_query') + def test_raises_delete_tuple(self, mock_model_query): + mock_model_query.side_effect = Exception('regular test failure') + self.assertRaises(freezer_api_exc.StorageEngineError, + api.delete_tuple, models.Job, self.fake_user_id, + self.fake_job_id, + project_id=self.fake_project_id) + + def test_delete_tuple(self): + job_doc1 = copy.deepcopy(self.fake_job_0) + job_doc2 = copy.deepcopy(self.fake_job_0) + self.dbapi.add_job(user_id=self.fake_user_id, + doc=job_doc1, + project_id=self.fake_project_id) + job_id2 = self.dbapi.add_job(user_id=self.fake_user_id, + doc=job_doc2, + project_id=self.fake_project_id) + + result = self.dbapi.delete_job(user_id=self.fake_user_id, + job_id=job_id2, + project_id=self.fake_project_id) + session = api.get_db_session() + with session.begin(): + try: + query = api.model_query(session, models.Job, + read_deleted='no', + project_id=self.fake_project_id) + query = query.filter_by(user_id=self.fake_user_id) + result = query.all() + self.assertEqual(len(result), 1) + + query = api.model_query(session, models.Job, + read_deleted='only', + project_id=self.fake_project_id) + query = query.filter_by(user_id=self.fake_user_id) + result = query.all() + self.assertEqual(len(result), 1) + + query = api.model_query(session, models.Job, + read_deleted='yes', + project_id=self.fake_project_id) + query = query.filter_by(user_id=self.fake_user_id) + result = query.all() + self.assertEqual(len(result), 2) + except Exception as e: + raise freezer_api_exc.StorageEngineError( + message='sqlalchemy operation failed {0}'.format(e)) + session.close() + + @patch('oslo_db.sqlalchemy.utils.model_query') + def test_raises_get_tuple(self, mock_model_query): + mock_model_query.side_effect = Exception('regular test failure') + self.assertRaises(freezer_api_exc.StorageEngineError, + api.get_tuple, models.Job, self.fake_user_id, + self.fake_job_id, + project_id=self.fake_project_id) + + def test_raises_add_tuple(self): + mock_tuple = mock.MagicMock() + mock_tuple.save.side_effect = Exception('regular test failure') + self.assertRaises(freezer_api_exc.StorageEngineError, + api.add_tuple, mock_tuple)