added basic tests for piccolo migrations forwards / backwards entrypoints

Author: dantownsend

piccolo.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+# This is used for running Piccolo commands on the example project within
+# tests.
+export PICCOLO_CONF="tests.postgres_conf"
+python -m piccolo.main $@

piccolo/apps/migrations/commands/backwards.py

@@ -5,9 +5,12 @@
 
 
 class BackwardsMigrationManager(BaseMigrationManager):
-    def __init__(self, app_name: str, migration_id: str):
+    def __init__(
+        self, app_name: str, migration_id: str, auto_agree: bool = False
+    ):
         self.migration_id = migration_id
         self.app_name = app_name
+        self.auto_agree = auto_agree
         super().__init__()
 
     def run(self):
@@ -57,10 +60,14 @@ def run(self):
 
         #######################################################################
 
-        _continue = input(
-            "About to undo the following migrations:\n"
-            f"{reversed_migration_ids}\n"
-            "Enter y to continue.\n"
+        _continue = (
+            input(
+                "About to undo the following migrations:\n"
+                f"{reversed_migration_ids}\n"
+                "Enter y to continue.\n"
+            )
+            if not self.auto_agree
+            else "y"
         )
         if _continue == "y":
             print("Undoing migrations")
@@ -80,7 +87,9 @@ def run(self):
             print("Not proceeding.")
 
 
-def backwards(app_name: str, migration_id: str = "1"):
+def backwards(
+    app_name: str, migration_id: str = "1", auto_agree: bool = False
+):
     """
     Undo migrations up to a specific migration.
 
@@ -91,16 +100,23 @@ def backwards(app_name: str, migration_id: str = "1"):
         Migrations will be reversed up to and including this migration_id.
         Specify a value of 'all' to undo all of the migrations. Specify a
         value of '1' to undo the most recent migration.
+    :param auto_agree:
+        Automatically agree to any input prompts.
+
     """
     if app_name == "all":
         sorted_app_names = BaseMigrationManager().get_sorted_app_names()
         sorted_app_names.reverse()
 
-        _continue = input(
-            "You're about to undo the migrations for the following apps:\n"
-            f"{sorted_app_names}\n"
-            "Are you sure you want to continue?\n"
-            "Enter y to continue.\n"
+        _continue = (
+            input(
+                "You're about to undo the migrations for the following apps:\n"
+                f"{sorted_app_names}\n"
+                "Are you sure you want to continue?\n"
+                "Enter y to continue.\n"
+            )
+            if not auto_agree
+            else "y"
         )
         if _continue == "y":
             for _app_name in sorted_app_names:
@@ -111,6 +127,6 @@ def backwards(app_name: str, migration_id: str = "1"):
                 manager.run()
     else:
         manager = BackwardsMigrationManager(
-            app_name=app_name, migration_id=migration_id
+            app_name=app_name, migration_id=migration_id, auto_agree=auto_agree
         )
         manager.run()

piccolo/query/methods/alter.py

@@ -229,12 +229,20 @@ def querystring(self) -> QueryString:
 class DropTable:
     tablename: str
     cascade: bool
+    if_exists: bool
 
     @property
     def querystring(self) -> QueryString:
-        query = f'DROP TABLE "{self.tablename}"'
+        query = "DROP TABLE"
+
+        if self.if_exists:
+            query += " IF EXISTS"
+
+        query += f" {self.tablename}"
+
         if self.cascade:
             query += " CASCADE"
+
         return QueryString(query)
 
 
@@ -294,12 +302,16 @@ def drop_default(self, column: t.Union[str, Column]) -> Alter:
         self._drop_default.append(DropDefault(column=column))
         return self
 
-    def drop_table(self, cascade: bool = False) -> Alter:
+    def drop_table(
+        self, cascade: bool = False, if_exists: bool = False
+    ) -> Alter:
         """
         Band.alter().drop_table()
         """
         self._drop_table = DropTable(
-            tablename=self.table._meta.tablename, cascade=cascade
+            tablename=self.table._meta.tablename,
+            cascade=cascade,
+            if_exists=if_exists,
         )
         return self
 

tests/apps/migrations/commands/test_forwards_backwards.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+from unittest import TestCase
+import typing as t
+
+from piccolo.apps.migrations.commands.backwards import backwards
+from piccolo.apps.migrations.commands.forwards import forwards
+
+from tests.example_app.tables import (
+    Manager,
+    Band,
+    Venue,
+    Concert,
+    Ticket,
+    Poster,
+)
+
+if t.TYPE_CHECKING:
+    from piccolo.table import Table
+
+
+TABLE_CLASSES: t.List[t.Type[Table]] = [
+    Manager,
+    Band,
+    Venue,
+    Concert,
+    Ticket,
+    Poster,
+]
+
+
+class TestForwardsBackwards(TestCase):
+    """
+    How to test this? Add migrations for the example app, and just run them
+    forwards and backwards. Or create a new app with loads of migrations in
+    it.
+    """
+
+    # TODO - use pytest parameterise ...
+    def test_forwards_backwards_all_migrations(self):
+        """
+        Test running all of the migrations forwards, then backwards.
+        """
+        forwards(app_name="example_app", migration_id="all")
+
+        # Check the tables exist
+        for table_class in TABLE_CLASSES:
+            self.assertTrue(table_class.table_exists().run_sync())
+
+        backwards(app_name="example_app", migration_id="all", auto_agree=True)
+
+        # Check the tables don't exist
+        for table_class in TABLE_CLASSES:
+            self.assertTrue(not table_class.table_exists().run_sync())
+
+    def tearDown(self):
+        for table_class in TABLE_CLASSES:
+            table_class.alter().drop_table(if_exists=True).run_sync()

tests/example_app/piccolo_app.py

@@ -1,11 +1,18 @@
+import os
+
 from piccolo.conf.apps import AppConfig
 
 from .tables import Manager, Band, Venue, Concert, Ticket, Poster
 
 
+CURRENT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))
+
+
 APP_CONFIG = AppConfig(
     app_name="example_app",
     table_classes=[Manager, Band, Venue, Concert, Ticket, Poster],
-    migrations_folder_path="",
+    migrations_folder_path=os.path.join(
+        CURRENT_DIRECTORY, "piccolo_migrations"
+    ),
     commands=[],
 )

tests/example_app/piccolo_migrations/2020-12-17T18:44:30.py

@@ -0,0 +1,86 @@
+from piccolo.apps.migrations.auto import MigrationManager
+from piccolo.columns.base import OnDelete
+from piccolo.columns.base import OnUpdate
+from piccolo.table import Table
+
+
+class Manager(Table, tablename="manager"):
+    pass
+
+
+ID = "2020-12-17T18:44:30"
+VERSION = "0.14.7"
+
+
+async def forwards():
+    manager = MigrationManager(migration_id=ID, app_name="example_app")
+
+    manager.add_table("Manager", tablename="manager")
+    manager.add_table("Band", tablename="band")
+
+    manager.add_column(
+        table_class_name="Band",
+        tablename="band",
+        column_name="name",
+        column_class_name="Varchar",
+        params={
+            "length": 50,
+            "default": "",
+            "null": False,
+            "primary": False,
+            "key": False,
+            "unique": False,
+            "index": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="Band",
+        tablename="band",
+        column_name="manager",
+        column_class_name="ForeignKey",
+        params={
+            "references": Manager,
+            "on_delete": OnDelete.cascade,
+            "on_update": OnUpdate.cascade,
+            "default": None,
+            "null": True,
+            "primary": False,
+            "key": False,
+            "unique": False,
+            "index": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="Band",
+        tablename="band",
+        column_name="popularity",
+        column_class_name="Integer",
+        params={
+            "default": 0,
+            "null": False,
+            "primary": False,
+            "key": False,
+            "unique": False,
+            "index": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="Manager",
+        tablename="manager",
+        column_name="name",
+        column_class_name="Varchar",
+        params={
+            "length": 50,
+            "default": "",
+            "null": False,
+            "primary": False,
+            "key": False,
+            "unique": False,
+            "index": False,
+        },
+    )
+
+    return manager