Add rudimentary migration testing

We start at version 0 and apply each migration one at a time.

bp db-migration-testing

Change-Id: If098d9baeca4c58e19b23fac69bcb48a3d6c072a

1 files changed, 38 insertions, 0 deletions

diff --git a/designate/tests/test_storage/test_sqlalchemy.py b/designate/tests/test_storage/test_sqlalchemy.py
index 1810bc69..859cc72f 100644
--- a/designate/tests/test_storage/test_sqlalchemy.py
+++ b/designate/tests/test_storage/test_sqlalchemy.py
@@ -13,10 +13,18 @@
 # 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 os
+import tempfile
+from migrate.versioning import api as versioning_api
+from migrate.versioning import repository
+import sqlalchemy
 from designate.openstack.common import log as logging
 from designate.tests.test_storage import StorageTestCase
 
 LOG = logging.getLogger(__name__)
+REPOSITORY = os.path.abspath(os.path.join(os.path.dirname(__file__), '..',
+                                          '..', 'storage', 'impl_sqlalchemy',
+                                          'migrate_repo'))
 
 
 class SqlalchemyStorageTest(StorageTestCase):
@@ -26,3 +34,33 @@ class SqlalchemyStorageTest(StorageTestCase):
         self.config(database_connection='sqlite://',
                     group='storage:sqlalchemy')
         super(SqlalchemyStorageTest, self).setUp()
+
+        self.REPOSITORY = repository.Repository(REPOSITORY)
+
+    # Migration Test Stuff
+    def _init_database(self, url):
+        LOG.debug('Building Engine')
+        engine = sqlalchemy.create_engine(url)
+
+        LOG.debug('Initializing database')
+        versioning_api.version_control(engine, repository=self.REPOSITORY)
+
+        return engine
+
+    def _migrate_up(self, engine, version):
+        versioning_api.upgrade(engine, repository=self.REPOSITORY,
+                               version=version)
+
+    def _migrate_down(self, engine, version):
+        versioning_api.downgrade(engine, repository=self.REPOSITORY,
+                                 version=version)
+
+    def test_migrations_walk(self):
+        # Init the Database
+        engine = self._init_database("sqlite:///%s" % tempfile.mktemp())
+
+        LOG.debug('Latest version is %s' % self.REPOSITORY.latest)
+
+        for version in range(1, self.REPOSITORY.latest):
+            LOG.debug('Upgrading to %s' % version)
+            self._migrate_up(engine, version)