From 73b435f1b53fc38a55a232a0cc4c9a8b79148f20 Mon Sep 17 00:00:00 2001
From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Wed, 3 Aug 2016 13:22:17 -0400
Subject: Add test helpers to enginefacade

Adds new methods make_new_manager(), patch_factory(),
patch_engine(), to provide support to test fixtures that
wish to generate new managers from an existing one, and to
cross-patch managers, factories and engines into other managers.

Change-Id: I83c6296d992f08117fff1438bc079bbf71db787b
---
 oslo_db/tests/sqlalchemy/test_enginefacade.py | 148 ++++++++++++++++++++++++++
 1 file changed, 148 insertions(+)

(limited to 'oslo_db/tests/sqlalchemy/test_enginefacade.py')

diff --git a/oslo_db/tests/sqlalchemy/test_enginefacade.py b/oslo_db/tests/sqlalchemy/test_enginefacade.py
index 72663d2..fb63856 100644
--- a/oslo_db/tests/sqlalchemy/test_enginefacade.py
+++ b/oslo_db/tests/sqlalchemy/test_enginefacade.py
@@ -1076,6 +1076,154 @@ class MockFacadeTest(oslo_test_base.BaseTestCase):
                     session.execute("test")
 
 
+class PatchFactoryTest(oslo_test_base.BaseTestCase):
+
+    def test_patch_manager(self):
+        normal_mgr = enginefacade.transaction_context()
+        normal_mgr.configure(connection="sqlite:///foo.db")
+        alt_mgr = enginefacade.transaction_context()
+        alt_mgr.configure(connection="sqlite:///bar.db")
+
+        @normal_mgr.writer
+        def go1(context):
+            s1 = context.session
+            self.assertEqual(
+                s1.bind.url, "sqlite:///foo.db")
+            self.assertIs(
+                s1.bind,
+                normal_mgr._factory._writer_engine)
+
+        @normal_mgr.writer
+        def go2(context):
+            s1 = context.session
+
+            self.assertEqual(
+                s1.bind.url,
+                "sqlite:///bar.db")
+
+            self.assertIs(
+                normal_mgr._factory._writer_engine,
+                alt_mgr._factory._writer_engine
+            )
+
+        def create_engine(sql_connection, **kw):
+            return mock.Mock(url=sql_connection)
+
+        with mock.patch(
+                "oslo_db.sqlalchemy.engines.create_engine", create_engine):
+            context = oslo_context.RequestContext()
+            go1(context)
+            reset = normal_mgr.patch_factory(alt_mgr)
+            go2(context)
+            reset()
+            go1(context)
+
+    def test_patch_factory(self):
+        normal_mgr = enginefacade.transaction_context()
+        normal_mgr.configure(connection="sqlite:///foo.db")
+        alt_mgr = enginefacade.transaction_context()
+        alt_mgr.configure(connection="sqlite:///bar.db")
+
+        @normal_mgr.writer
+        def go1(context):
+            s1 = context.session
+            self.assertEqual(
+                s1.bind.url, "sqlite:///foo.db")
+            self.assertIs(
+                s1.bind,
+                normal_mgr._factory._writer_engine)
+
+        @normal_mgr.writer
+        def go2(context):
+            s1 = context.session
+
+            self.assertEqual(
+                s1.bind.url,
+                "sqlite:///bar.db")
+
+            self.assertIs(
+                normal_mgr._factory._writer_engine,
+                alt_mgr._factory._writer_engine
+            )
+
+        def create_engine(sql_connection, **kw):
+            return mock.Mock(url=sql_connection)
+
+        with mock.patch(
+                "oslo_db.sqlalchemy.engines.create_engine", create_engine):
+            context = oslo_context.RequestContext()
+            go1(context)
+            reset = normal_mgr.patch_factory(alt_mgr._factory)
+            go2(context)
+            reset()
+            go1(context)
+
+    def test_patch_engine(self):
+        normal_mgr = enginefacade.transaction_context()
+        normal_mgr.configure(connection="sqlite:///foo.db")
+
+        @normal_mgr.writer
+        def go1(context):
+            s1 = context.session
+            self.assertEqual(
+                s1.bind.url, "sqlite:///foo.db")
+            self.assertIs(
+                s1.bind,
+                normal_mgr._factory._writer_engine)
+
+        @normal_mgr.writer
+        def go2(context):
+            s1 = context.session
+
+            self.assertEqual(
+                s1.bind.url,
+                "sqlite:///bar.db")
+
+        def create_engine(sql_connection, **kw):
+            return mock.Mock(url=sql_connection)
+
+        with mock.patch(
+                "oslo_db.sqlalchemy.engines.create_engine", create_engine):
+            mock_engine = create_engine("sqlite:///bar.db")
+
+            context = oslo_context.RequestContext()
+            go1(context)
+            reset = normal_mgr.patch_engine(mock_engine)
+            go2(context)
+            self.assertIs(
+                normal_mgr._factory._writer_engine, mock_engine)
+            reset()
+            go1(context)
+
+    def test_new_manager_from_config(self):
+        normal_mgr = enginefacade.transaction_context()
+        normal_mgr.configure(
+            connection="sqlite://",
+            sqlite_fk=True,
+            mysql_sql_mode="FOOBAR",
+            max_overflow=38
+        )
+
+        normal_mgr._factory._start()
+
+        copied_mgr = normal_mgr.make_new_manager()
+
+        self.assertTrue(normal_mgr._factory._started)
+        self.assertIsNotNone(normal_mgr._factory._writer_engine)
+
+        self.assertIsNot(copied_mgr._factory, normal_mgr._factory)
+        self.assertFalse(copied_mgr._factory._started)
+        copied_mgr._factory._start()
+        self.assertIsNot(
+            normal_mgr._factory._writer_engine,
+            copied_mgr._factory._writer_engine)
+
+        engine_args = copied_mgr._factory._engine_args_for_conf(None)
+        self.assertTrue(engine_args['sqlite_fk'])
+        self.assertEqual("FOOBAR", engine_args["mysql_sql_mode"])
+        self.assertEqual(38, engine_args["max_overflow"])
+
+
 class SynchronousReaderWSlaveMockFacadeTest(MockFacadeTest):
     synchronous_reader = True
 
-- 
cgit v1.2.1