Add raise unit test for get and add tuple in sqlalchemy
Change-Id: I8c9f05a370c42fe4b547b45b53535dbb51bb9cd6
This commit is contained in:
parent
4326fd75ba
commit
1977c7e9d3
|
@ -159,10 +159,8 @@ def delete_tuple(tablename, user_id, tuple_id, project_id=None):
|
||||||
'{0} not found in Table {1}'.
|
'{0} not found in Table {1}'.
|
||||||
format(tuple_id, tablename))
|
format(tuple_id, tablename))
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
session.close()
|
|
||||||
raise freezer_api_exc.StorageEngineError(
|
raise freezer_api_exc.StorageEngineError(
|
||||||
message='Mysql operation failed {0}'.format(e))
|
message='Mysql operation failed {0}'.format(e))
|
||||||
session.close()
|
|
||||||
return tuple_id
|
return tuple_id
|
||||||
|
|
||||||
|
|
||||||
|
@ -186,12 +184,9 @@ def add_tuple(tuple):
|
||||||
try:
|
try:
|
||||||
tuple.save(session=session)
|
tuple.save(session=session)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
session.close()
|
|
||||||
raise freezer_api_exc.StorageEngineError(
|
raise freezer_api_exc.StorageEngineError(
|
||||||
message='Mysql operation failed {0}'.format(e))
|
message='Mysql operation failed {0}'.format(e))
|
||||||
|
|
||||||
session.close()
|
|
||||||
|
|
||||||
|
|
||||||
def update_tuple(tablename, user_id, tuple_id, tuple_values, project_id=None):
|
def update_tuple(tablename, user_id, tuple_id, tuple_values, project_id=None):
|
||||||
|
|
||||||
|
|
|
@ -18,6 +18,8 @@
|
||||||
"""Tests for manipulating job via the DB API"""
|
"""Tests for manipulating job via the DB API"""
|
||||||
|
|
||||||
import copy
|
import copy
|
||||||
|
import mock
|
||||||
|
from mock import patch
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
|
|
||||||
from freezer_api.common import exceptions as freezer_api_exc
|
from freezer_api.common import exceptions as freezer_api_exc
|
||||||
|
@ -89,3 +91,58 @@ class ApiTestCase(base.DbTestCase):
|
||||||
raise freezer_api_exc.StorageEngineError(
|
raise freezer_api_exc.StorageEngineError(
|
||||||
message='sqlalchemy operation failed {0}'.format(e))
|
message='sqlalchemy operation failed {0}'.format(e))
|
||||||
session.close()
|
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)
|
||||||
|
|
|
@ -18,6 +18,8 @@
|
||||||
"""Tests for manipulating job via the DB API"""
|
"""Tests for manipulating job via the DB API"""
|
||||||
|
|
||||||
import copy
|
import copy
|
||||||
|
import mock
|
||||||
|
from mock import patch
|
||||||
|
|
||||||
from freezer_api.common import exceptions as freezer_api_exc
|
from freezer_api.common import exceptions as freezer_api_exc
|
||||||
from freezer_api.db.sqlalchemy import api
|
from freezer_api.db.sqlalchemy import api
|
||||||
|
@ -89,3 +91,66 @@ class ApiTestCase(base.DbTestCase):
|
||||||
raise freezer_api_exc.StorageEngineError(
|
raise freezer_api_exc.StorageEngineError(
|
||||||
message='sqlalchemy operation failed {0}'.format(e))
|
message='sqlalchemy operation failed {0}'.format(e))
|
||||||
session.close()
|
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)
|
||||||
|
|
Loading…
Reference in New Issue