How to write integration test with mongoDB

[BobbyLeeSH]

Hi,

I am trying to write some tests to check its connection to mongo and to test simple crud, but I just cannot find many examples about it - unlike sqlalchemy :(

Would you be able to provide some guide on testing with pytest?

Thanks.

[roman-right]

Hi! Good point. Thank you.

I will update this example to the current syntax and will add tests there. For now, you can follow next example:

import http
from typing import Iterator

import pytest
from asgi_lifespan import LifespanManager
from beanie import init_beanie, Document
from fastapi import FastAPI
from httpx import AsyncClient
from motor.motor_asyncio import AsyncIOMotorClient

MONGO_URI = "YOUR_MONGODB_URI"


class User(Document):
    name: str


app = FastAPI()


@app.get("/users/{name}")
async def get_users_by_name(name: str):
    return await User.find(User.name == name).to_list()


@app.on_event("startup")
async def app_init():
    app.db = AsyncIOMotorClient(MONGO_URI).account
    await init_beanie(app.db, document_models=[User])


@pytest.fixture()
async def client() -> Iterator[AsyncClient]:
    async with LifespanManager(app):
        async with AsyncClient(app=app, base_url="http://test") as ac:
            yield ac


@pytest.mark.asyncio
async def test_user_get_all(client: AsyncClient) -> None:
    user = User(name="test")
    await user.create()
    resp = await client.get("/users/test")
    assert resp.status_code == http.HTTPStatus.OK

[BobbyLeeSH]

Hey Roman, thank you for your response. Do you have any way to create collections if they do not exist when the application start up?

[roman-right]

Hey @BobbyLeeSH , If there is no collection, the engine creates it automatically by default. Did you face a case, where it was not created?

[BobbyLeeSH]

No, I was just wondering about it. I just tried and it works fine. Thank you for your prompt response.

Can I set ttl for a document within beanie?

[roman-right]

Yes, sure. You can do it with the indexes expireAfterSeconds parameter.

example:

import asyncio
from datetime import datetime

from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import Field

from beanie import init_beanie, Document, Indexed

MONGO_URI = "YOUR_URI"


class User(Document):
    name: str
    ts: Indexed(datetime, expireAfterSeconds=60) = Field(
        default_factory=datetime.utcnow)


async def main():
    db = AsyncIOMotorClient(MONGO_URI).test_db
    await init_beanie(db, document_models=[User])

    user = User(name="Test")
    await user.insert()


asyncio.run(main())

Beanie indexes doc: https://roman-right.github.io/beanie/tutorial/indexes-%26-collection-names/#indexes MongoDB related doc: https://docs.mongodb.com/manual/core/index-ttl/

[BobbyLeeSH]

Great! Thank you Roman! Hope you have a great weekend!

Should I close the issue? Or will you close it after you add testing?

[roman-right]

Welcome :)

I'll add tests and close it, yes. Thank you for pointing this issue. I forgot about this example project and have to update it much now :-)

Have a nice weekend!

[BobbyLeeSH]

Hey Roman,

I was going through Integration tests and wondering if there are any ways to rollback or drop all collections?

Thanks!

[roman-right]

Hey @BobbyLeeSH ,

You can use the next way for this:

await YourDocument.get_motor_collection().drop()

pytest feature example:

@pytest.fixture(autouse=True)
async def init(db):
    models = [
        Sample,
        TestModel,
    ]
    await init_beanie(
        database=db,
        document_models=models,
    )
    yield None

    for model in models:
        await model.get_motor_collection().drop()
        await model.get_motor_collection().drop_indexes()

[acefei]

Hi! Good point. Thank you.

I will update this example to the current syntax and will add tests there. For now, you can follow next example:

import http
from typing import Iterator

import pytest
from asgi_lifespan import LifespanManager
from beanie import init_beanie, Document
from fastapi import FastAPI
from httpx import AsyncClient
from motor.motor_asyncio import AsyncIOMotorClient

MONGO_URI = "YOUR_MONGODB_URI"


class User(Document):
    name: str


app = FastAPI()


@app.get("/users/{name}")
async def get_users_by_name(name: str):
    return await User.find(User.name == name).to_list()


@app.on_event("startup")
async def app_init():
    app.db = AsyncIOMotorClient(MONGO_URI).account
    await init_beanie(app.db, document_models=[User])


@pytest.fixture()
async def client() -> Iterator[AsyncClient]:
    async with LifespanManager(app):
        async with AsyncClient(app=app, base_url="http://test") as ac:
            yield ac


@pytest.mark.asyncio
async def test_user_get_all(client: AsyncClient) -> None:
    user = User(name="test")
    await user.create()
    resp = await client.get("/users/test")
    assert resp.status_code == http.HTTPStatus.OK

Replacing the fixture as @pytest_asyncio.fixture, ref: https://github.com/pytest-dev/pytest-asyncio/issues/390