77 lines
2.0 KiB
Python
77 lines
2.0 KiB
Python
import os
|
|
|
|
from argon2 import PasswordHasher
|
|
import pytest
|
|
import asyncio
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy import text
|
|
from alembic.config import Config
|
|
from alembic import command
|
|
|
|
from src.core.config import settings
|
|
|
|
hasher = PasswordHasher()
|
|
|
|
ADMIN_PASSWORD = hasher.hash(os.getenv("OPENBAO__TEST_ADMIN_PASSWORD", "admin123"))
|
|
ISP_PASSWORD = hasher.hash(os.getenv("OPENBAO__TEST_EXECUTOR_PASSWORD", "admin123"))
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption(
|
|
"--init_db",
|
|
action="store_true",
|
|
default=False,
|
|
help="Initialize database with migrations and seed data"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def init_db_flag(request):
|
|
return request.config.getoption("--init_db")
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def setup_database(init_db_flag):
|
|
if not init_db_flag:
|
|
return
|
|
|
|
asyncio.run(run_migrations_async())
|
|
asyncio.run(
|
|
run_seed_script_async(
|
|
admin_password=ADMIN_PASSWORD,
|
|
isp_password=ISP_PASSWORD,
|
|
)
|
|
)
|
|
|
|
|
|
async def run_migrations_async():
|
|
loop = asyncio.get_running_loop()
|
|
await loop.run_in_executor(None, run_migrations_sync)
|
|
|
|
|
|
def run_migrations_sync():
|
|
alembic_cfg = Config("alembic.ini")
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
|
|
async def run_seed_script_async(admin_password, isp_password):
|
|
sql_script_path = os.path.join(os.path.dirname(__file__), "sqls", "fixture.sql")
|
|
|
|
with open(sql_script_path, "r", encoding="utf-8") as f:
|
|
sql_script = f.read()
|
|
|
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
|
|
|
try:
|
|
async with engine.connect() as conn:
|
|
for statement in sql_script.split(";"):
|
|
statement = statement.strip().format(
|
|
admin_password=admin_password,
|
|
isp_password=isp_password,
|
|
)
|
|
if statement:
|
|
await conn.execute(text(statement))
|
|
await conn.commit()
|
|
finally:
|
|
await engine.dispose()
|