From 66a4889ee96452962a95e5458429911dc29a841a Mon Sep 17 00:00:00 2001 From: Darla Ahlert Date: Fri, 13 Jan 2017 13:59:16 -0600 Subject: [PATCH] Unit testing for db.sqlalchemy.api.py Unit tests for db/sqlalchemy/api.py Change-Id: I28f5d7d997e150e0269cf08e2039265e13c83ec4 Co-Authored-By: Jin Li --- gluon/db/sqlalchemy/api.py | 1 - gluon/tests/db/test_api_connection.py | 203 +++++++++++++++++++++++ gluon/tests/db/test_db_sqlalchemy_api.py | 131 +++++++++++++++ 3 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 gluon/tests/db/test_api_connection.py create mode 100644 gluon/tests/db/test_db_sqlalchemy_api.py diff --git a/gluon/db/sqlalchemy/api.py b/gluon/db/sqlalchemy/api.py index e749928..2cd264f 100644 --- a/gluon/db/sqlalchemy/api.py +++ b/gluon/db/sqlalchemy/api.py @@ -106,7 +106,6 @@ class Connection(api.Connection): def _add_filters(self, query, filters): if filters is None: filters = {} - for (key, value) in six.iteritems(filters): query = query.filter_by(**{key: value}) diff --git a/gluon/tests/db/test_api_connection.py b/gluon/tests/db/test_api_connection.py new file mode 100644 index 0000000..fa7c701 --- /dev/null +++ b/gluon/tests/db/test_api_connection.py @@ -0,0 +1,203 @@ +# Copyright 2016, AT&T +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import sqlalchemy.orm.exc + +from gluon.common import exception +from gluon.db.sqlalchemy import api +from gluon.db.sqlalchemy import models as sql_models +from gluon.tests import base +from mock import patch +from oslo_db import exception as db_exc + + +class ConnectionTestCase(base.TestCase): + + def setUp(self): + super(ConnectionTestCase, self).setUp() + + def raiseDBDuplicateEntryException(self): + raise db_exc.DBDuplicateEntry + + """ + test create() + """ + + """ + # can't figure this error out...is this a bug?? + + exception as e.__dict__ = {'value': None, 'inner_exception': None, + 'cause': None, 'columns': []} + + File "gluon/db/sqlalchemy/api.py", line 106, in create + key=e.__dict__['columns'][0], + IndexError: list index out of range + + # exception thrown + def test_create_duplicate_entry(self): + connection = api.Connection() + mock_model = mock.Mock() + mock_model.__name__ = "fake_name" + print("mock_model = " + str(mock_model)) + + mock_values = mock.Mock() + mock_obj = mock_model() + print("mock obj before = " + str(mock_obj)) + + mock_obj.update.return_value = mock_obj + print("mock_obj after update = " + str(mock_obj)) + + mock_obj.save = self.raiseDBDuplicateEntryException + print("mock_obj after save = " + str(mock_obj)) + + self.assertRaises(exception.AlreadyExists, connection.create, + mock_model, mock_values) + """ + + # no exception + def test_create(self): + connection = api.Connection() + mock_model = mock.Mock() + mock_model.__name__ = "fake_name" + mock_values = mock.Mock() + mock_obj = mock_model() + mock_obj.update.return_value = mock_obj + mock_obj.save.return_value = mock_obj + + observed = connection.create(mock_model, mock_values) + + mock_obj.update.assert_called_once() + mock_obj.save.assert_called_once() + self.assertEqual(mock_obj, observed) + + """ + test _add_filters() + """ + def test_add_filters(self): + connection = api.Connection() + mock_filters = {'key': 'value'} + mock_query = mock.Mock() + mock_query.filter_by.return_value = mock_query + + observed = connection._add_filters(mock_query, mock_filters) + + mock_query.filter_by.assert_called_once() + self.assertEqual(mock_query, observed) + + """ + test get_list() + """ + @mock.patch('gluon.db.sqlalchemy.api.model_query') + @mock.patch('gluon.db.sqlalchemy.api.Connection._add_filters') + @mock.patch('gluon.db.sqlalchemy.api._paginate_query') + def test_get_list(self, mock_paginate_query, mock_add_filters, + mock_model_query): + connection = api.Connection() + mock_query = mock.Mock() + mock_query_2 = mock.Mock() + mock_model = mock.Mock() + mock_model_query.return_value = mock_query + mock_add_filters.return_value = mock_query_2 + mock_paginate_query.return_value = mock_query_2 + + observed = connection.get_list(mock_model, columns=None, + filters=None, limit=None, + marker=None, sort_key=None, + sort_dir=None, failed=None, + period=None) + + mock_model_query.assert_called_once() + mock_add_filters.assert_called_once() + mock_paginate_query.assert_called_once() + self.assertEqual(mock_query_2, observed) + + def raiseNoResultFoundException(self): + raise sqlalchemy.orm.exc.NoResultFound + + """ + test get_by_uuid() + """ + # exception thrown + @mock.patch('gluon.db.sqlalchemy.api.model_query') + def test_get_by_uuid_noResultFound(self, mock_model_query): + connection = api.Connection() + mock_model = mock.Mock() + mock_model.__name__ = "fake_name" + mock_uuid = mock.Mock() + mock_query = mock.Mock() + mock_model_query.return_value = mock_query + mock_query.filter_by.return_value = mock_query + mock_query.one = self.raiseNoResultFoundException + self.assertRaises(exception.NotFound, + connection.get_by_uuid, mock_model, + mock_uuid) + + # no exception thrown + @mock.patch('gluon.db.sqlalchemy.api.model_query') + def test_get_by_uuid(self, mock_model_query): + connection = api.Connection() + mock_model = mock.Mock() + mock_uuid = mock.Mock() + mock_query = mock.Mock() + mock_model_query.return_value = mock_query + mock_query.filter_by.return_value = mock_query + mock_query.one.return_value = mock_query + + observed = connection.get_by_uuid(mock_model, mock_uuid) + + mock_model_query.assert_called_once() + mock_query.filter_by.assert_called_once() + self.assertEqual(mock_query, observed) + + """ + test get_by_primary_key() + """ + # exption thrown + @mock.patch('gluon.db.sqlalchemy.api.model_query') + def test_get_by_primary_key_noResultFound(self, mock_model_query): + connection = api.Connection() + mock_pk_type = "fake_pk" + mock_model = mock.Mock() + mock_model.__name__ = "fake_name" + mock_key = "fake_key" + mock_query = mock.Mock() + mock_model.get_primary_key.return_value = mock_pk_type + mock_model_query.return_value = mock_query + mock_query.filter_by.return_value = mock_query + mock_query.one = self.raiseNoResultFoundException + self.assertRaises(exception.NotFound, + connection.get_by_primary_key, + mock_model, mock_key) + + @mock.patch('gluon.db.sqlalchemy.api.model_query') + def test_get_by_primary_key(self, mock_model_query): + connection = api.Connection() + mock_pk_type = "fake_pk" + mock_key = "fake_key" + mock_query = mock.Mock() + mock_model = mock.Mock() + mock_key = mock.Mock() + mock_model.get_primary_key.return_value = mock_pk_type + mock_model_query.return_value = mock_query + mock_query.filter_by.return_value = mock_query + mock_query.one.return_value = mock_query + observed = connection.get_by_primary_key(mock_model, mock_key) + + mock_model.get_primary_key.assert_called_once() + mock_model_query.assert_called_once() + mock_query.filter_by.assert_called_once() + mock_query.one.assert_called_once() + + self.assertEqual(mock_query, observed) diff --git a/gluon/tests/db/test_db_sqlalchemy_api.py b/gluon/tests/db/test_db_sqlalchemy_api.py new file mode 100644 index 0000000..1ee630d --- /dev/null +++ b/gluon/tests/db/test_db_sqlalchemy_api.py @@ -0,0 +1,131 @@ +# Copyright 2016, AT&T +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock + +from gluon.db.sqlalchemy import api +from gluon.tests import base +from mock import patch + + +class ApiTestCase(base.TestCase): + + def setUp(self): + super(ApiTestCase, self).setUp() + + """ + test _create_facade_lazily() + """ + @mock.patch('gluon.db.sqlalchemy.api.db_session') + def test_create_facade_lazily(self, mock_db_session): + mock_return = mock.Mock() + mock_db_session.EngineFacade.from_config.return_value = mock_return + api._FACADE = None + + api._create_facade_lazily() + + mock_db_session.EngineFacade.from_config.assert_called_with(api.CONF) + self.assertIsNotNone(api._FACADE) + self.assertEqual(api._FACADE, mock_return) + + """ + test get_engine + """ + @mock.patch('gluon.db.sqlalchemy.api._create_facade_lazily') + def test_get_engine(self, mock_create_facade_lazily): + re = mock.Mock() + mock_create_facade_lazily.return_value.get_engine.return_value = re + + observed = api.get_engine() + + mock_create_facade_lazily.assert_called_once() + self.assertEqual(re, observed) + + """ + test get_session + """ + @mock.patch('gluon.db.sqlalchemy.api._create_facade_lazily') + def test_get_session(self, mock_create_facade_lazily): + re = mock.Mock() + mock_create_facade_lazily.return_value.get_session.return_value = re + + observed = api.get_session() + + mock_create_facade_lazily.assert_called_once() + self.assertEqual(re, observed) + + """ + test get_backend + """ + @mock.patch('gluon.db.sqlalchemy.api.Connection') + def test_get_backend(self, mock_Connection): + api.get_backend() + mock_Connection.assert_called_once() + + """ + test model_query + """ + def test_model_query_with_session(self): + model = mock.Mock() + session = mock.Mock() + query = mock.Mock() + session.query.return_value = query + observed = api.model_query(model, session=session) + session.query.assert_called_once() + self.assertEqual(query, observed) + + @mock.patch('gluon.db.sqlalchemy.api.get_session') + def test_model_query_no_session(self, mock_get_session): + model = mock.Mock() + query = mock.Mock() + session = mock.Mock() + mock_get_session.return_value = session + session.query.return_value = query + observed = api.model_query(model) + mock_get_session.assert_called_once() + session.query.assert_called_once() + self.assertEqual(query, observed) + + """ + test _paginate_query + """ + @mock.patch('gluon.db.sqlalchemy.api.model_query') + @mock.patch('gluon.db.sqlalchemy.api.db_utils.paginate_query') + def test_paginate_query(self, mock_paginate_query, mock_model_query): + query_1 = mock.Mock() + mock_model_query.return_value = query_1 + + mock_model = mock.Mock() + sample_sort_keys = "foo" + mock_model.get_primary_key.return_value = sample_sort_keys + + sample_sort_key = 'bar' + + query_2 = mock.Mock() + query_2.all.return_value = "query result" + mock_paginate_query.return_value = query_2 + + observed = api._paginate_query(mock_model, limit=None, marker=None, + sort_key=sample_sort_key, + sort_dir=None, query=None) + + mock_model_query.assert_called_once() + mock_model.get_primary_key.assert_called_once() + mock_paginate_query.assert_called_once_with(query_1, + mock_model, + None, + ['bar', 'foo'], + marker=None, + sort_dir=None) + self.assertEqual("query result", observed)