ff

tests/unit/test_database.py

@@ -7,6 +7,8 @@ import tempfile
 import unittest
 from unittest.mock import patch
 
+import click
+
 
 SRC_ROOT = Path(__file__).resolve().parents[2] / "src"
 
@@ -165,9 +167,16 @@ class DatabaseContextTests(unittest.TestCase):
         finally:
             connection.close()
 
-        reopened_context = databaseContext(str(self.database_path))
+        with patch("ffx.database.click.confirm", return_value=True) 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))
+            mocked_confirm.assert_called_once()
+
+            backup_path = Path(f"{self.database_path}.v2-to-v3.bak")
+            self.assertTrue(backup_path.exists())
 
             Session = reopened_context["session"]
             session = Session()
@@ -188,6 +197,42 @@ class DatabaseContextTests(unittest.TestCase):
         finally:
             reopened_context["engine"].dispose()
 
+        echoedLines = [call.args[0] for call in mocked_echo.call_args_list]
+        self.assertIn("Database migration required.", echoedLines)
+        self.assertIn("Current version: 2", echoedLines)
+        self.assertIn(f"Target version: {DATABASE_VERSION}", echoedLines)
+        self.assertIn(
+            "  2 -> 3: ffx.model.migration.step_2_3 [present]",
+            echoedLines,
+        )
+
+    def test_database_context_aborts_migration_when_confirmation_is_declined(self):
+        context = databaseContext(str(self.database_path))
+        try:
+            Session = context["session"]
+            session = Session()
+            try:
+                version_row = (
+                    session.query(Property)
+                    .filter(Property.key == DATABASE_VERSION_KEY)
+                    .first()
+                )
+                version_row.value = "2"
+                session.commit()
+            finally:
+                session.close()
+        finally:
+            context["engine"].dispose()
+
+        with patch("ffx.database.click.confirm", return_value=False), patch(
+            "ffx.database.click.echo"
+        ):
+            with self.assertRaises(click.ClickException) as raisedContext:
+                databaseContext(str(self.database_path))
+
+        self.assertEqual("Database migration aborted by user.", str(raisedContext.exception))
+        self.assertFalse(Path(f"{self.database_path}.v2-to-v3.bak").exists())
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_migration.py

@@ -11,10 +11,24 @@ if str(SRC_ROOT) not in sys.path:
     sys.path.insert(0, str(SRC_ROOT))
 
 
-from ffx.model.migration import DatabaseVersionException, loadMigrationStep, migrateDatabase  # noqa: E402
+from ffx.model.migration import (  # noqa: E402
+    DatabaseVersionException,
+    getMigrationPlan,
+    loadMigrationStep,
+    migrateDatabase,
+)
 
 
 class MigrationTests(unittest.TestCase):
+    def test_get_migration_plan_lists_known_step_with_module_presence(self):
+        migrationPlan = getMigrationPlan(2, 3)
+
+        self.assertEqual(1, len(migrationPlan))
+        self.assertEqual(2, migrationPlan[0].versionFrom)
+        self.assertEqual(3, migrationPlan[0].versionTo)
+        self.assertEqual("ffx.model.migration.step_2_3", migrationPlan[0].moduleName)
+        self.assertTrue(migrationPlan[0].modulePresent)
+
     def test_load_migration_step_returns_known_step(self):
         migrationStep = loadMigrationStep(2, 3)
 
@@ -31,4 +45,3 @@ class MigrationTests(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
-