Fixing tests and adding manage.py to debug

api/asset_manager/src/database.py

@@ -39,11 +39,12 @@ TORTOISE_ORM = {
 }
 
 
-async def migrate_db():
+async def migrate_db(tortoise_config=TORTOISE_ORM):
-    aerich = Command(tortoise_config=TORTOISE_ORM)
+    aerich = Command(tortoise_config)
     await aerich.init()
     await aerich.upgrade(run_in_transaction=True)
-    await Tortoise.init(config=TORTOISE_ORM)
+    await Tortoise.init(tortoise_config)
+    await Tortoise.generate_schemas(safe=True)
 
 async def end_connections_to_db():
     await Tortoise.close_connections()

api/asset_manager/src/main.py

@@ -31,7 +31,12 @@ app = FastAPI(
 
 if settings.USE_HTTPS_ONLY:
     app.add_middleware(HTTPSRedirectMiddleware)
-app.add_middleware(TrustedHostMiddleware, allowed_hosts=[settings.PROJECT_PUBLIC_URL,])
+app.add_middleware(
+    TrustedHostMiddleware,
+    allowed_hosts=[
+        settings.PROJECT_PUBLIC_URL,
+    ],
+)
 
 # Set all CORS enabled origins
 if settings.BACKEND_CORS_ORIGINS:

api/asset_manager/src/manage.py

@@ -1,24 +1,103 @@
 #!/usr/bin/env python3
 
-from ptpython.repl import embed  # type: ignore
+from ptpython.repl import embed, ReplExit
 
-from database import *
+import asyncio, importlib, contextlib, sys, os, tomllib, asyncclick
 
-import asyncio
+from database import migrate_db
+from pathlib import Path
+from asyncclick import BadOptionUsage, ClickException
+from collections.abc import AsyncGenerator
+from tortoise import Tortoise, connections
 
+#
+# Custom implementation of Tortoise CLI
+# Original script is located under: https://github.com/tortoise/tortoise-cli
+# License: Apache-2.0 as dictated as [here](https://github.com/tortoise/tortoise-cli/blob/main/LICENSE)
+#
+
+
+def tortoise_orm_config(file="pyproject.toml") -> str:
+    """
+    get tortoise orm config from os environment variable or aerich item in pyproject.toml
+
+    :param file: toml file that aerich item loads from it
+    :return: module path and var name that store the tortoise config, e.g.: 'settings.TORTOISE_ORM'
+    """
+    if not (config := os.getenv("TORTOISE_ORM", "")) and (p := Path(file)).exists():
+        doc = tomllib.loads(p.read_text("utf-8"))
+        config = doc.get("tool", {}).get("aerich", {}).get("tortoise_orm", "")
+    return config
+
+
+def get_tortoise_config(config: str) -> dict:
+    """
+    get tortoise config from module
+    :param ctx:
+    :param config:
+    :return:
+    """
+    splits = config.split(".")
+    config_path = ".".join(splits[:-1])
+    tortoise_config = splits[-1]
+
+    try:
+        config_module = importlib.import_module(config_path)
+    except ModuleNotFoundError as e:
+        raise ClickException(
+            f"Error while importing configuration module: {e}"
+        ) from None
+    c = getattr(config_module, tortoise_config, None)
+    if not c:
+        raise BadOptionUsage(
+            option_name="--config",
+            message=f'Can\'t get "{tortoise_config}" from module "{config_module}"',
+            ctx=None,
+        )
+    return c
+
+
+@contextlib.asynccontextmanager
+async def aclose_tortoise() -> AsyncGenerator[None]:
+    try:
+        yield
+    finally:
+        if Tortoise._inited:
+            await connections.close_all()
+
+def history():
+    import readline
+    for i in range(1, readline.get_current_history_length()+1):
+        print("%3d %s" % (i, readline.get_history_item(i)))
 
 async def setup():
-    try:
+    if not (config := tortoise_orm_config()):
-        await embed(globals=globals(), return_asyncio_coroutine=True, patch_stdout=True)
+        raise asyncclick.UsageError(
-    except EOFError:
+            "You must specify TORTOISE_ORM in option or env, or config file pyproject.toml from config of aerich",
-        loop.stop()
+            ctx=None,
+        )
+    await migrate_db(get_tortoise_config(config))
+
+    async with aclose_tortoise():
+        await embed(
+            globals=globals(),
+            title="shell",
+            vi_mode=True,
+            return_asyncio_coroutine=True,
+            patch_stdout=True,
+        )
 
 
 if __name__ == "__main__":
+    if sys.path[0] != ".":
+        sys.path.insert(0, ".")
+
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
     try:
-        asyncio.ensure_future(setup())
+        history()
-        loop.run_forever()
+        loop.run_until_complete(asyncio.ensure_future(setup()))
-    except KeyboardInterrupt:
+    except (KeyboardInterrupt, ReplExit) as e:
-        pass
+        print(e)
+        loop.stop()
+

api/asset_manager/src/requirements/requirements.txt

@@ -18,4 +18,5 @@ pytest>=8.4.0
 asyncio>=3.4.3
 pytest-mock>=3.14.1
 pytest-asyncio>=1.0.0
-asgi-lifespan>=2.1.0
+asgi-lifespan>=2.1.0
+Faker>=37.4.0

api/asset_manager/src/tests/base_test.py

@@ -0,0 +1,5 @@
+import pytest
+
+@pytest.mark.usefixtures("use_database_during_testing")
+class Test():
+    pass

api/asset_manager/src/tests/conftest.py

@@ -2,9 +2,10 @@ import asyncio
 from contextlib import asynccontextmanager
 from typing import AsyncGenerator
 import httpx, pytest
+from tortoise import Tortoise
+from asgi_lifespan import LifespanManager
 
-from asgi_lifespan import LifespanManager  # type: ignore
+from database import migrate_db
-
 from tests.fixtures.account import *
 
 try:
@@ -31,12 +32,20 @@ def event_loop():
     loop.close()
 
 
+@pytest.fixture
+async def use_database_during_testing():
+    await migrate_db()
+    yield
+    await Tortoise._drop_databases()
+
+
 @asynccontextmanager
 async def client_manager(app, base_url="https://localhost", **kw) -> ClientManagerType:
     app.state.testing = True
     async with LifespanManager(app):
-        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
-        async with httpx.AsyncClient(transport=transport, base_url=base_url, **kw) as c:
+            transport=httpx.ASGITransport(app=app), base_url=base_url, **kw
+        ) as c:
             yield c
 
 

api/asset_manager/src/tests/test_authentication/test_authentication.py

@@ -1,3 +1,4 @@
+from tests.base_test import Test
 from modules.users.models import User
 import pytest  # type: ignore
 from httpx import AsyncClient
@@ -8,7 +9,7 @@ from tortoise.expressions import Q
 crypt = settings.CRYPT
 
 
-class TestAuthentication(object):
+class TestAuthentication(Test):
     @pytest.mark.asyncio
     async def test_authentication_with_non_existing_user_and_password(
         self, client: AsyncClient