Merge "Use oslo database code"

6 files changed, 10 insertions, 135 deletions

diff --git a/nova/tests/baremetal/db/test_bm_interface.py b/nova/tests/baremetal/db/test_bm_interface.py
index 9f051ac9b..32beb1ce0 100644
--- a/nova/tests/baremetal/db/test_bm_interface.py
+++ b/nova/tests/baremetal/db/test_bm_interface.py
@@ -18,6 +18,7 @@ Bare-metal DB testcase for BareMetalInterface
 """
 
 from nova import exception
+from nova.openstack.common.db.sqlalchemy import session as db_session
 from nova.tests.baremetal.db import base
 from nova.virt.baremetal import db
 
@@ -27,7 +28,7 @@ class BareMetalInterfaceTestCase(base.BMDBTestCase):
     def test_unique_address(self):
         pif1_id = db.bm_interface_create(self.context, 1, '11:11:11:11:11:11',
                                          '0x1', 1)
-        self.assertRaises(exception.DBError,
+        self.assertRaises(db_session.DBError,
                           db.bm_interface_create,
                           self.context, 2, '11:11:11:11:11:11', '0x2', 2)
         # succeed after delete pif1
diff --git a/nova/tests/baremetal/db/test_bm_pxe_ip.py b/nova/tests/baremetal/db/test_bm_pxe_ip.py
index 9a93b46ad..9820f3af0 100644
--- a/nova/tests/baremetal/db/test_bm_pxe_ip.py
+++ b/nova/tests/baremetal/db/test_bm_pxe_ip.py
@@ -18,6 +18,7 @@ Bare-metal DB testcase for BareMetalPxeIp
 """
 
 from nova import exception
+from nova.openstack.common.db.sqlalchemy import session as db_session
 from nova.tests.baremetal.db import base
 from nova.tests.baremetal.db import utils
 from nova.virt.baremetal import db
@@ -50,14 +51,14 @@ class BareMetalPxeIpTestCase(base.BMDBTestCase):
         # address duplicates
         i = utils.new_bm_pxe_ip(address='10.1.1.1',
                                 server_address='10.1.1.201')
-        self.assertRaises(exception.DBError,
+        self.assertRaises(db_session.DBError,
                           db.bm_pxe_ip_create_direct,
                           self.context, i)
 
         # server_address duplicates
         i = utils.new_bm_pxe_ip(address='10.1.1.3',
                                 server_address='10.1.1.101')
-        self.assertRaises(exception.DBError,
+        self.assertRaises(db_session.DBError,
                           db.bm_pxe_ip_create_direct,
                           self.context, i)
 
diff --git a/nova/tests/baremetal/test_pxe.py b/nova/tests/baremetal/test_pxe.py
index 09f1079bf..e50462b0e 100644
--- a/nova/tests/baremetal/test_pxe.py
+++ b/nova/tests/baremetal/test_pxe.py
@@ -25,6 +25,7 @@ from testtools import matchers
 
 from nova import exception
 from nova.openstack.common import cfg
+from nova.openstack.common.db.sqlalchemy import session as db_session
 from nova.tests.baremetal.db import base as bm_db_base
 from nova.tests.baremetal.db import utils as bm_db_utils
 from nova.tests.image import fake as fake_image
@@ -521,7 +522,7 @@ class PXEPublicMethodsTestCase(BareMetalPXETestCase):
                 AndRaise(exception.NovaException)
         bm_utils.unlink_without_raise(pxe_path)
         self.driver._collect_mac_addresses(self.context, self.node).\
-                AndRaise(exception.DBError)
+                AndRaise(db_session.DBError)
         bm_utils.rmtree_without_raise(
                 os.path.join(CONF.baremetal.tftp_root, 'fake-uuid'))
         self.mox.ReplayAll()
diff --git a/nova/tests/network/test_manager.py b/nova/tests/network/test_manager.py
index 48183010f..9d467bcb1 100644
--- a/nova/tests/network/test_manager.py
+++ b/nova/tests/network/test_manager.py
@@ -29,6 +29,7 @@ from nova.network import linux_net
 from nova.network import manager as network_manager
 from nova.network import model as net_model
 from nova.openstack.common import cfg
+from nova.openstack.common.db.sqlalchemy import session as db_session
 from nova.openstack.common import importutils
 from nova.openstack.common import log as logging
 from nova.openstack.common import rpc
@@ -2046,7 +2047,7 @@ class FloatingIPTestCase(test.TestCase):
         # address column, so fake the collision-avoidance here
         def fake_vif_save(vif):
             if vif.address == crash_test_dummy_vif['address']:
-                raise exception.DBError("If you're smart, you'll retry!")
+                raise db_session.DBError("If you're smart, you'll retry!")
         self.stubs.Set(models.VirtualInterface, 'save', fake_vif_save)
 
         # Attempt to add another and make sure that both MACs are consumed
diff --git a/nova/tests/test_instance_types.py b/nova/tests/test_instance_types.py
index 9a42bac0c..0829df2c6 100644
--- a/nova/tests/test_instance_types.py
+++ b/nova/tests/test_instance_types.py
@@ -21,8 +21,8 @@ from nova.compute import instance_types
 from nova import context
 from nova import db
 from nova.db.sqlalchemy import models
-from nova.db.sqlalchemy import session as sql_session
 from nova import exception
+from nova.openstack.common.db.sqlalchemy import session as sql_session
 from nova.openstack.common import log as logging
 from nova import test
 
diff --git a/nova/tests/test_sqlalchemy.py b/nova/tests/test_sqlalchemy.py
deleted file mode 100644
index 5c7f4450b..000000000
--- a/nova/tests/test_sqlalchemy.py
+++ /dev/null
@@ -1,129 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-# Copyright (c) 2012 Rackspace Hosting
-# All Rights Reserved.
-#
-#    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.
-
-"""Unit tests for SQLAlchemy specific code."""
-
-from eventlet import db_pool
-try:
-    import MySQLdb
-except ImportError:
-    MySQLdb = None
-
-from sqlalchemy import Column, MetaData, Table, UniqueConstraint
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy import DateTime, Integer
-
-from nova import context
-from nova.db.sqlalchemy import models
-from nova.db.sqlalchemy import session
-from nova import exception
-from nova import test
-
-
-class DbPoolTestCase(test.TestCase):
-    def setUp(self):
-        super(DbPoolTestCase, self).setUp()
-        self.flags(sql_dbpool_enable=True)
-        self.user_id = 'fake'
-        self.project_id = 'fake'
-        self.context = context.RequestContext(self.user_id, self.project_id)
-        if not MySQLdb:
-            self.skipTest("Unable to test due to lack of MySQLdb")
-
-    def test_db_pool_option(self):
-        self.flags(sql_idle_timeout=11, sql_min_pool_size=21,
-                sql_max_pool_size=42)
-
-        info = {}
-
-        class FakeConnectionPool(db_pool.ConnectionPool):
-            def __init__(self, mod_name, **kwargs):
-                info['module'] = mod_name
-                info['kwargs'] = kwargs
-                super(FakeConnectionPool, self).__init__(mod_name,
-                        **kwargs)
-
-            def connect(self, *args, **kwargs):
-                raise test.TestingException()
-
-        self.stubs.Set(db_pool, 'ConnectionPool',
-                FakeConnectionPool)
-
-        sql_connection = 'mysql://user:pass@127.0.0.1/nova'
-        self.assertRaises(test.TestingException, session.create_engine,
-                sql_connection)
-
-        self.assertEqual(info['module'], MySQLdb)
-        self.assertEqual(info['kwargs']['max_idle'], 11)
-        self.assertEqual(info['kwargs']['min_size'], 21)
-        self.assertEqual(info['kwargs']['max_size'], 42)
-
-
-BASE = declarative_base()
-_TABLE_NAME = '__tmp__test__tmp__'
-
-
-class TmpTable(BASE, models.NovaBase):
-    __tablename__ = _TABLE_NAME
-    id = Column(Integer, primary_key=True)
-    foo = Column(Integer)
-
-
-class SessionErrorWrapperTestCase(test.TestCase):
-    def setUp(self):
-        super(SessionErrorWrapperTestCase, self).setUp()
-        meta = MetaData()
-        meta.bind = session.get_engine()
-        test_table = Table(_TABLE_NAME, meta,
-                           Column('id', Integer, primary_key=True,
-                                  nullable=False),
-                           Column('deleted', Integer, default=0),
-                           Column('deleted_at', DateTime),
-                           Column('updated_at', DateTime),
-                           Column('created_at', DateTime),
-                           Column('foo', Integer),
-                           UniqueConstraint('foo', name='uniq_foo'))
-        test_table.create()
-
-    def tearDown(self):
-        super(SessionErrorWrapperTestCase, self).tearDown()
-        meta = MetaData()
-        meta.bind = session.get_engine()
-        test_table = Table(_TABLE_NAME, meta, autoload=True)
-        test_table.drop()
-
-    def test_flush_wrapper(self):
-        tbl = TmpTable()
-        tbl.update({'foo': 10})
-        tbl.save()
-
-        tbl2 = TmpTable()
-        tbl2.update({'foo': 10})
-        self.assertRaises(exception.DBDuplicateEntry, tbl2.save)
-
-    def test_execute_wrapper(self):
-        _session = session.get_session()
-        with _session.begin():
-            for i in [10, 20]:
-                tbl = TmpTable()
-                tbl.update({'foo': i})
-                tbl.save(session=_session)
-
-            method = _session.query(TmpTable).\
-                            filter_by(foo=10).\
-                            update
-            self.assertRaises(exception.DBDuplicateEntry,
-                              method, {'foo': 20})