Add tests (#70)

Author: lsbardel

fluid/__init__.py

@@ -1,3 +1,3 @@
 """Reusable server side python modules"""
 
-__version__ = "1.5.0"
+__version__ = "1.5.1"

fluid/db/cli.py

@@ -155,9 +155,24 @@ def current(ctx: click.Context, verbose: bool) -> None:
 def create(ctx: click.Context, dbname: str) -> None:
     """Creates a new database"""
     if get_db(ctx).migration().db_create(dbname):
-        click.echo(f"database {dbname} created")
+        click.echo(f"database '{dbname}' created")
     else:
-        click.echo(f"database {dbname} already available")
+        click.echo(f"database '{dbname}' already available")
+
+
+@_db.command()
+@click.argument("dbname", nargs=1)
+@click.option("-y", "--yes", is_flag=True, help="confirm")
+@click.pass_context
+def drop(ctx: click.Context, dbname: str, yes: bool) -> None:
+    """Drop an existing database"""
+    if not yes:
+        click.echo(f"Are you sure you want to drop database '{dbname}'?")
+        click.confirm("Please confirm", abort=True)
+    if get_db(ctx).migration().db_drop(dbname):
+        click.echo(f"database '{dbname}' dropped")
+    else:
+        click.echo(f"database '{dbname}' not found")
 
 
 @_db.command()

fluid/db/migration.py

@@ -9,7 +9,7 @@
 from alembic import command as alembic_cmd
 from alembic.config import Config
 from sqlalchemy.engine import Engine
-from sqlalchemy_utils import create_database, database_exists
+from sqlalchemy_utils import create_database, database_exists, drop_database
 
 if TYPE_CHECKING:
     from .container import Database
@@ -94,6 +94,15 @@ def db_create(self, dbname: str = "") -> bool:
         create_database(url)
         return True
 
+    def db_drop(self, dbname: str = "") -> bool:
+        url = self.sync_engine.url
+        if dbname:
+            url = url.set(database=dbname)
+        if database_exists(url):
+            drop_database(url)
+            return True
+        return False
+
     def truncate_all(self) -> None:
         """Drop all tables from :attr:`metadata` in database"""
         with self.sync_engine.begin() as conn:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "aio-fluid"
-version = "1.5.0"
+version = "1.5.1"
 description = "Tools for backend python services"
 authors = [
     { name = "Luca Sbardella", email = "luca@quantmind.com" }

tests/db/test_cli.py

@@ -1,6 +1,24 @@
+from click.testing import CliRunner
+
 from examples.db.cli import cli
 from fluid.db import CrudDB
 
 
 def test_cli():
     assert isinstance(cli.db, CrudDB)
+
+
+def test_create_drop_db():
+    runner = CliRunner()
+    result = runner.invoke(cli, ["create", "test_db_abc"])
+    assert result.exit_code == 0
+    assert "database 'test_db_abc' created" in result.output
+    result = runner.invoke(cli, ["create", "test_db_abc"])
+    assert result.exit_code == 0
+    assert "database 'test_db_abc' already available" in result.output
+    result = runner.invoke(cli, ["drop", "test_db_abc", "-y"])
+    assert result.exit_code == 0
+    assert "database 'test_db_abc' dropped" in result.output
+    result = runner.invoke(cli, ["drop", "test_db_abc", "-y"])
+    assert result.exit_code == 0
+    assert "database 'test_db_abc' not found" in result.output