break oracle thin mode asyncio in pytest

test_oracle_break.py

import asyncio
from contextlib import asynccontextmanager

import oracledb
import pytest
import pytest_asyncio
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import AsyncClient
from oracledb import AsyncConnectionPool

from mirouteapi.connections.oracle import ENV, MRTE_ORACLE_PWD


class OracleWrapper:
    pool: AsyncConnectionPool

    async def init(self):
        self.pool = oracledb.create_pool_async(**ENV, password=MRTE_ORACLE_PWD)

    async def close(self):
        await self.pool.close()


oracle = OracleWrapper()


@asynccontextmanager
async def lifespan(app: FastAPI):
    await oracle.init()
    yield
    await oracle.close()


app = FastAPI(lifespan=lifespan)


@app.get("/test-oracle")
async def get_something_from_oracle():
    async with oracle.pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute("SELECT 1 FROM DUAL")
            row = await cursor.fetchone()
            return row[0]


@pytest.fixture(scope="session")
def event_loop():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest_asyncio.fixture(scope="session")
async def app_fixture():
    async with LifespanManager(app) as manager:
        yield manager.app


@pytest_asyncio.fixture(scope="session")
async def client(app_fixture):
    async with AsyncClient(app=app_fixture, base_url="http://app.io") as client:
        yield client


@pytest.mark.asyncio
async def test_oracle(client: AsyncClient):
    response = await client.get("/test-oracle")
    assert response.status_code == 200