Add get_non_innodb_tables() to utils

Add a helper function to utils for detection of tables which aren't
using InnoDB storage engine in MySQL (something we want to enforce in
consuming projects).

Change-Id: I22889e23b025fc32fe76e4086cd3cd3d5bd0b771

1 files changed, 44 insertions, 0 deletions

diff --git a/tests/sqlalchemy/test_utils.py b/tests/sqlalchemy/test_utils.py
index 88c3817..a4cd3f2 100644
--- a/tests/sqlalchemy/test_utils.py
+++ b/tests/sqlalchemy/test_utils.py
@@ -1094,3 +1094,47 @@ class TestDialectFunctionDispatcher(test_base.BaseTestCase):
             "Return value not allowed for multiple filtered function",
             str(exc)
         )
+
+
+class TestGetInnoDBTables(db_test_base.MySQLOpportunisticTestCase):
+
+    def test_all_tables_use_innodb(self):
+        self.engine.execute("CREATE TABLE customers "
+                            "(a INT, b CHAR (20), INDEX (a)) ENGINE=InnoDB")
+        self.assertEqual([], utils.get_non_innodb_tables(self.engine))
+
+    def test_all_tables_use_innodb_false(self):
+        self.engine.execute("CREATE TABLE employee "
+                            "(i INT) ENGINE=MEMORY")
+        self.assertEqual(['employee'],
+                         utils.get_non_innodb_tables(self.engine))
+
+    def test_skip_tables_use_default_value(self):
+        self.engine.execute("CREATE TABLE migrate_version "
+                            "(i INT) ENGINE=MEMORY")
+        self.assertEqual([],
+                         utils.get_non_innodb_tables(self.engine))
+
+    def test_skip_tables_use_passed_value(self):
+        self.engine.execute("CREATE TABLE some_table "
+                            "(i INT) ENGINE=MEMORY")
+        self.assertEqual([],
+                         utils.get_non_innodb_tables(
+                             self.engine, skip_tables=('some_table',)))
+
+    def test_skip_tables_use_empty_list(self):
+        self.engine.execute("CREATE TABLE some_table_3 "
+                            "(i INT) ENGINE=MEMORY")
+        self.assertEqual(['some_table_3'],
+                         utils.get_non_innodb_tables(
+                         self.engine, skip_tables=()))
+
+    def test_skip_tables_use_several_values(self):
+        self.engine.execute("CREATE TABLE some_table_1 "
+                            "(i INT) ENGINE=MEMORY")
+        self.engine.execute("CREATE TABLE some_table_2 "
+                            "(i INT) ENGINE=MEMORY")
+        self.assertEqual([],
+                         utils.get_non_innodb_tables(
+                             self.engine,
+                             skip_tables=('some_table_1', 'some_table_2')))