prisma-korea / prisma-fastapi Goto Github PK

TEST_DB_DSN = "postgresql://postgres:postgres@localhost:5433/postgres"
prisma = Prisma(datasource={"url": TEST_DB_DSN})


async def teardown_db(client: Prisma) -> None:
    print("Teardown")
    print("Dropping all tables")
    await client.execute_raw("DROP schema public CASCADE")
    print("Creating all tables")
    stream = os.popen(f"DB_DSN={TEST_DB_DSN} prisma db push --skip-generate")
    output = stream.read()


@pytest.fixture(scope="module", autouse=True)
async def test_init(request):
    await prisma.connect()
    yield
    await teardown_db(client=prisma)
    await prisma.disconnect()


@pytest.fixture(scope="module")
def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


# Test client
@pytest.fixture(scope="module")
def client(event_loop: asyncio.BaseEventLoop) -> Iterator[TestClient]:
    initializer(TORTOISE_ORM["apps"]["models"]["models"], loop=event_loop)

    with TestClient(app) as c:
        yield c
    finalizer()