Merge pull request piccolo-orm#26 from piccolo-orm/test_coverage

Test coverage

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/app/commands/new.py

@@ -22,8 +22,7 @@ def new_app(app_name: str, root: str = "."):
     app_root = os.path.join(root, app_name)
 
     if os.path.exists(app_root):
-        print("Folder already exists - exiting.")
-        sys.exit(1)
+        sys.exit("Folder already exists - exiting.")
     os.mkdir(app_root)
 
     with open(os.path.join(app_root, "__init__.py"), "w"):

piccolo/apps/migrations/commands/backwards.py

@@ -1,16 +1,30 @@
+from __future__ import annotations
+import os
+import sys
+
 from piccolo.apps.migrations.auto import MigrationManager
+from piccolo.apps.migrations.commands.base import BaseMigrationManager
 from piccolo.apps.migrations.tables import Migration
 from piccolo.utils.sync import run_sync
-from .base import BaseMigrationManager
 
 
 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,
+        clean: bool = False,
+    ):
         self.migration_id = migration_id
         self.app_name = app_name
+        self.auto_agree = auto_agree
+        self.clean = clean
         super().__init__()
 
     def run(self):
+        self.create_migration_table()
+
         app_modules = self.get_app_modules()
 
         migration_modules = []
@@ -27,8 +41,7 @@ def run(self):
             app_name=self.app_name
         )
         if len(ran_migration_ids) == 0:
-            print("No migrations to reverse!")
-            return
+            sys.exit("No migrations to reverse!")
 
         #######################################################################
 
@@ -40,7 +53,7 @@ def run(self):
             earliest_migration_id = self.migration_id
 
         if earliest_migration_id not in ran_migration_ids:
-            print(
+            sys.exit(
                 "Unrecognized migration name - must be one of "
                 f"{ran_migration_ids}"
             )
@@ -57,10 +70,14 @@ def run(self):
 
         #######################################################################
 
-        _continue = input(
-            "About to undo the following migrations:\n"
-            f"{reversed_migration_ids}\n"
-            "Enter y to continue.\n"
+        _continue = (
+            "y"
+            if self.auto_agree
+            else input(
+                "About to undo the following migrations:\n"
+                f"{reversed_migration_ids}\n"
+                "Enter y to continue.\n"
+            )
         )
         if _continue == "y":
             print("Undoing migrations")
@@ -76,11 +93,20 @@ def run(self):
                 Migration.delete().where(
                     Migration.name == migration_id
                 ).run_sync()
-        else:
-            print("Not proceeding.")
 
+                if self.clean:
+                    os.unlink(migration_module.__file__)
 
-def backwards(app_name: str, migration_id: str = "1"):
+        else:  # pragma: no cover
+            sys.exit("Not proceeding.")
+
+
+def backwards(
+    app_name: str,
+    migration_id: str = "1",
+    auto_agree: bool = False,
+    clean: bool = False,
+):
     """
     Undo migrations up to a specific migration.
 
@@ -91,26 +117,41 @@ 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.
+    :param clean:
+        If true, the migration files which have been run backwards are deleted
+        from the disk after completing.
+
     """
     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 = (
+            "y"
+            if auto_agree
+            else 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 _continue == "y":
             for _app_name in sorted_app_names:
                 print(f"Undoing {_app_name}")
                 manager = BackwardsMigrationManager(
-                    app_name=_app_name, migration_id="all"
+                    app_name=_app_name,
+                    migration_id="all",
+                    auto_agree=auto_agree,
                 )
                 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,
+            clean=clean,
         )
         manager.run()

piccolo/apps/migrations/commands/forwards.py

@@ -2,11 +2,10 @@
 import sys
 import typing as t
 
-from .base import BaseMigrationManager
-from piccolo.conf.apps import AppConfig
-from piccolo.apps.migrations.tables import Migration
 from piccolo.apps.migrations.auto import MigrationManager
-from piccolo.conf.apps import MigrationModule
+from piccolo.apps.migrations.commands.base import BaseMigrationManager
+from piccolo.apps.migrations.tables import Migration
+from piccolo.conf.apps import AppConfig, MigrationModule
 from piccolo.utils.sync import run_sync
 
 
@@ -40,8 +39,7 @@ def run_migrations(self, app_config: AppConfig) -> None:
         print(f"Haven't run = {havent_run}")
 
         if len(havent_run) == 0:
-            print("No migrations left to run!")
-            return
+            sys.exit("No migrations left to run!")
 
         if self.migration_id == "all":
             subset = havent_run

piccolo/apps/migrations/commands/new.py

@@ -75,8 +75,7 @@ def _create_new_migration(app_config: AppConfig, auto=False) -> None:
         )
 
         if sum([len(i.statements) for i in alter_statements]) == 0:
-            print("No changes detected - exiting.")
-            sys.exit(1)
+            sys.exit("No changes detected - exiting.")
 
         file_contents = render_template(
             migration_id=_id,
@@ -151,8 +150,7 @@ def new(app_name: str, auto: bool = False):
 
     engine = Finder().get_engine()
     if auto and isinstance(engine, SQLiteEngine):
-        print("Auto migrations aren't currently supported by SQLite.")
-        sys.exit(1)
+        sys.exit("Auto migrations aren't currently supported by SQLite.")
 
     app_config = Finder().get_app_config(app_name=app_name)
 

piccolo/apps/playground/commands/run.py

@@ -131,10 +131,9 @@ def run(
     try:
         import IPython  # type: ignore
     except ImportError:
-        print(
+        sys.exit(
             "Install iPython using `pip install ipython` to use this feature."
         )
-        sys.exit(1)
 
     if engine.upper() == "POSTGRES":
         db: Engine = PostgresEngine(

piccolo/apps/project/commands/new.py

@@ -23,8 +23,7 @@ def new_piccolo_conf(engine_name: str, force: bool = False, root: str = "."):
     file_path = os.path.join(root, "piccolo_conf.py")
 
     if os.path.exists(file_path) and not force:
-        print("The file already exists - exiting.")
-        sys.exit(1)
+        sys.exit("The file already exists - exiting.")
 
     with open(file_path, "w") as f:
         template = JINJA_ENV.get_template("piccolo_conf.py.jinja")

piccolo/apps/shell/commands/run.py

@@ -4,23 +4,28 @@
 from piccolo.conf.apps import AppRegistry, AppConfig, Finder
 from piccolo.table import Table
 
+try:
+    import IPython  # type: ignore
+    from IPython.core.interactiveshell import _asyncio_runner  # type: ignore
+
+    IPYTHON = True
+except ImportError:
+    IPYTHON = False
+
 
-def start_ipython_shell(**tables: t.Dict[str, t.Type[Table]]):
-    try:
-        import IPython  # type: ignore
-    except ImportError:
-        print(
+def start_ipython_shell(
+    **tables: t.Dict[str, t.Type[Table]]
+):  # pragma: no cover
+    if not IPYTHON:
+        sys.exit(
             "Install iPython using `pip install ipython` to use this feature."
         )
-        sys.exit(1)
 
     existing_global_names = globals().keys()
     for table_class_name, table_class in tables.items():
         if table_class_name not in existing_global_names:
             globals()[table_class_name] = table_class
 
-    from IPython.core.interactiveshell import _asyncio_runner  # type: ignore
-
     IPython.embed(using=_asyncio_runner)
 
 
@@ -41,7 +46,9 @@ def run():
             app_config: AppConfig = app_config
             print(f"Importing {app_name} tables:")
             if app_config.table_classes:
-                for table_class in app_config.table_classes:
+                for table_class in sorted(
+                    app_config.table_classes, key=lambda x: x.__name__
+                ):
                     table_class_name = table_class.__name__
                     print(f"- {table_class_name}")
                     tables[table_class_name] = table_class

piccolo/apps/user/commands/change_password.py

@@ -17,8 +17,7 @@ def change_password():
     confirmed_password = get_confirmed_password()
 
     if not password == confirmed_password:
-        print("Passwords don't match!")
-        sys.exit(1)
+        sys.exit("Passwords don't match!")
 
     BaseUser.update_password_sync(user=username, password=password)
 

piccolo/apps/user/commands/create.py

@@ -43,12 +43,10 @@ def create():
     confirmed_password = get_confirmed_password()
 
     if not password == confirmed_password:
-        print("Passwords don't match!")
-        sys.exit(1)
+        sys.exit("Passwords don't match!")
 
     if len(password) < 4:
-        print("The password is too short")
-        sys.exit(1)
+        sys.exit("The password is too short")
 
     is_admin = get_is_admin()