drop flag

tests/unit/test_database.py

@@ -149,6 +149,49 @@ class DatabaseContextTests(unittest.TestCase):
         )
         cursor.execute("DROP TABLE shifted_seasons_current")
 
+    def rewrite_tracks_table_without_dropped(self, cursor):
+        cursor.execute("ALTER TABLE tracks RENAME TO tracks_current")
+        cursor.execute(
+            """
+            CREATE TABLE tracks (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                track_type INTEGER,
+                "index" INTEGER,
+                source_index INTEGER,
+                pattern_id INTEGER,
+                disposition_flags INTEGER,
+                codec_name VARCHAR,
+                audio_layout INTEGER,
+                FOREIGN KEY(pattern_id) REFERENCES patterns(id) ON DELETE CASCADE
+            )
+            """
+        )
+        cursor.execute(
+            """
+            INSERT INTO tracks (
+                id,
+                track_type,
+                "index",
+                source_index,
+                pattern_id,
+                disposition_flags,
+                codec_name,
+                audio_layout
+            )
+            SELECT
+                id,
+                track_type,
+                "index",
+                source_index,
+                pattern_id,
+                disposition_flags,
+                codec_name,
+                audio_layout
+            FROM tracks_current
+            """
+        )
+        cursor.execute("DROP TABLE tracks_current")
+
     def test_database_context_bootstraps_new_database_with_current_version(self):
         with patch("ffx.database.Base.metadata.create_all", wraps=Base.metadata.create_all) as mocked_create_all:
             context = databaseContext(str(self.database_path))
@@ -321,6 +364,41 @@ class DatabaseContextTests(unittest.TestCase):
         mocked_confirm.assert_not_called()
         mocked_echo.assert_not_called()
 
+    def test_database_context_repairs_current_track_schema_without_version_bump(self):
+        context = databaseContext(str(self.database_path))
+        context["engine"].dispose()
+
+        connection = sqlite3.connect(self.database_path)
+        try:
+            cursor = connection.cursor()
+            cursor.execute("PRAGMA foreign_keys=OFF")
+            self.rewrite_tracks_table_without_dropped(cursor)
+            connection.commit()
+        finally:
+            connection.close()
+
+        with patch("ffx.database.click.confirm") as mocked_confirm, patch(
+            "ffx.database.click.echo"
+        ) as mocked_echo:
+            reopened_context = databaseContext(str(self.database_path))
+        try:
+            self.assertEqual(DATABASE_VERSION, getDatabaseVersion(reopened_context))
+
+            connection = sqlite3.connect(self.database_path)
+            try:
+                column_names = {
+                    row[1]
+                    for row in connection.execute("PRAGMA table_info(tracks)").fetchall()
+                }
+                self.assertIn("dropped", column_names)
+            finally:
+                connection.close()
+        finally:
+            reopened_context["engine"].dispose()
+
+        mocked_confirm.assert_not_called()
+        mocked_echo.assert_not_called()
+
 
 if __name__ == "__main__":
     unittest.main()