dishka tests

We have some app and dishka container

from fastapi import FasAPI

from dishka.integrations.fastapi import Depends, inject, setup_dishka

router = APIRouter()

@router.get("/")
@inject
async def index(connection: Annotated[Connection, Depends()]) -> str:
    result = connection.query("select 1";)
    return result

app = FastAPI(lifespan=lifespan)
app.include_router(router)
container = make_async_container(ConfigProvider(), ConnectionProvider())
setup_dishka(container, app)

and there is our providers

from dishka import make_container, Provider, provide, Scope

class ConfigProvider(Provider):
    @provide(scope=Scope.APP)
    def get_config(self) -> Config:
    ...

class ConnectionProvider(Provider):
    @provide(scope=Scope.REQUEST)
    def get_connection(self, uri: Config) -> Iterable[Connection]:
        conn = connect(uri)
        yield conn
        conn.close()

Let's test it! First declare our fixtures

class MockConfigProvider(Provider):
    @provide(scope=APP)
    def get_mock_config(self) -> Config:
       ...

@pytest.fixture(scope=session)
def dishka():
    container = make_async_container(MockConfigProvider(), ConnectionProvider())
    yield container
    container.close()

def app(dishka: AsyncContiner):
    app_ = FastAPI()
    app_.include_router(app_)
    setup_dishka(dishka, app_)

@pytest.fixture
async def request_dishka(dishka: AsyncContiner):
    async with dishka() as request_container:
        yield request_continer

@pytest_asyncio.fixture(scope=session)
async def config(dishka: AsyncContiner):
    return await dishka.get(Config)

@pytest_adyncio.fixture
async def connection(request_dishka: AsyncContiner):
    return await request_dishka.get(Connection)

Now we ready to start test

async def test_controller(request_dishka: AsyncContiner, connection: Connection, app: FastAPI):
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    connection.check_requested()

bomzheg/Desc.md