Commits

Ronny Pfannschmidt committed 758a3d8

detection of failed migrations

Comments (0)

Files changed (3)

 def push_migration(connection, migration, first):
     print('migration', migration.name)
     _prepare_migration(connection, migration, first)
-    connection.execute(migration.sql)
-    _record_migration_result(connection, migration, first)
+    try:
+        connection.executescript(migration.sql)
+    except connection.Error as error:
+        print(' ', error)
+    else:
+        _record_migration_result(connection, migration, first)
+
 
 
 def migration_state(connection):
     items = list(c)
     if items:
         return dict(connection.execute("""
-            select name, checksum
+            select
+                name,
+                case
+                    when completed = 1
+                    then checksum
+                    else ':failed to complete'
+                end
             from micromigrate_migrations
         """))
 
 
 def pick_next_doable(migrations):
     names = set(x.name for x in migrations)
-
     migrations = [
         mig for mig in migrations
-        if mig.after is None or not (mig.after - names)
+        if mig.after is None or not (names & mig.after)
     ]
     return migrations[0]
 

testing/conftest.py

+import pytest
+
+
+@pytest.mark.tryfirst
+def pytest_runtest_makereport(item, __multicall__):
+    report = __multicall__.execute()
+    conn = getattr(item, '_conn', None)
+    if conn is not None:
+        report.sections.append(('db', '\n'.join(conn.iterdump())))
+    return report
+
+

testing/test_miromigrate.py

 import pytest
 import micromigrate as mm
 
-
 @pytest.fixture
 def plain_conn(request):
     import sqlite3
     conn = sqlite3.connect(':memory:')
-    #XXX: dump
+    request._pyfuncitem._conn = conn
     return conn
 
 
+@pytest.fixture
+def conn(plain_conn):
+    mm.migrate(plain_conn, [])
+    return plain_conn
+
+
 def print_db(conn):
     print('\n'.join(conn.iterdump()))
 
     assert mm.migration_state(plain_conn) == {
         mm.initial_migration.name: mm.initial_migration.checksum,
     }
+
+
+def test_boken_transaction(conn):
+    state = mm.migration_state(conn)
+    print('state', sorted(state))
+    migration = mm.parse_migration(u"""
+        -- migration broke
+        -- after micromigrate:enable
+        create table foo(name unique);
+        insert into foo values ('a');
+        insert into foo values ('a');
+        """)
+    print(migration, migration.after)
+    mm.migrate(conn, [migration])
+    state = mm.migration_state(conn)
+    assert state[migration.name] == ':failed to complete'