Pytest-Asyncio: Collectionwasnotinialized Error с Fastapi и Beanie ODM в тестовой настройкеPython

Программы на Python
Ответить Пред. темаСлед. тема
Anonymous
 Pytest-Asyncio: Collectionwasnotinialized Error с Fastapi и Beanie ODM в тестовой настройке

Сообщение Anonymous »

Я строю приложение FASTAPI, используя шаблон репозитория с Beanie ODM для MongoDB. Я пытаюсь настроить тесты с использованием pytest-asyncio, но я сталкиваюсь с ошибкой Beanie.exceptions.collectionwasnotinitialization при запуске моих тестов.
Вот моя структура проекта : < /p>

Код: Выделить всё

├── product_catalouge
│   ├── __init__.py
│   ├── core
│   │   ├── __init__.py
│   │   └── config.py
│   ├── exceptions
│   │   ├── database_exceptions.py
│   │   └── http_exceptions.py
│   ├── lib
│   │   ├── __init__.py
│   │   ├── db_initializer.py
│   │   ├── models_loader.py
│   │   └── utils.py
│   ├── models
│   │   ├── __init__.py
│   │   ├── attribute.py
│   │   ├── .
│   │   ├── .
│   │   └── .
│   ├── repository
│   │   ├── __init__.py
│   │   ├── attribute_repository.py
│   │   ├── base_repository.py
│   │   ├── .
│   │   ├── .
│   │   ├── .
│   ├── schemas
│   │   ├── __init__.py
│   │   ├── attribute.py
│   │   ├── .
│   │   ├── .
│   │   └── .
│   ├── service
│   │   ├── __init__.py
│   │   ├── attributes_service.py
│   │   ├── .
│   │   ├── .
│   │   ├── .
│   │   ├── base_service.py
│   │   ├── domain.py
│   │   └── unit_of_work.py
│   └── web
│       ├── __init__.py
│       ├── api
│       │   ├── __init__.py
│       │   ├── api.py
│       │   ├── routers
│       │   │   ├── __init__.py
│       │   │   ├── attribute.py
│       │   │   ├── .
│       │   │   ├── .
│       │   │   └── .
│       └── app.py
├── pyproject.toml
└── tests
├── __init__.py
├── conftest.py
└── test_attributes
├── conftest.py
└── test_attributes.py
< /code>
инициализация моего приложения Fastapi (web/api/api.py
):

Код: Выделить всё

# imports are omitted

@asynccontextmanager
async def lifespan(app: FastAPI):
await init_db(settings.db_uri, settings.db_name)
yield

app = FastAPI(title="Product Catalogue Service",
docs_url="/api/docs",
openapi_url="/api/openapi",
lifespan=lifespan)

app.include_router(attribute.router)
< /code>
инициализация базы данных (lib/db_initializer.py
):

Код: Выделить всё

# imports are omitted

async def init_db(mongo_uri:str, db_name:str):
client = AsyncIOMotorClient(mongo_uri)
db = client[db_name]
await init_beanie(database=db,
document_models=get_beanie_models())
< /code>
Настройка теста в (tests/conftest.py
):

Код: Выделить всё

# some imports are omitted

from product_catalouge.core.config import settings
from product_catalouge.lib.models_loader import get_beanie_models
from product_catalouge.lib.db_initializer import init_db
from product_catalouge.web.api.routers import attribute

routers = [attribute]

@pytest_asyncio.fixture(autouse=True)
async def testdb():
db_client = AsyncIOMotorClient(settings.db_uri)
try:
await db_client.server_info()
print("[TEST DB CONNECTION] MongoDB connection successful.")
except Exception as e:
print(f"[TEST DB CONNECTION] MongoDB connection failed: {e}")
raise
await init_beanie(database=db_client.get_database(name=settings.test_db_name),
document_models=get_beanie_models())
print("[TEST DB INITIALIZATION] Beanie initialization complete.")
yield db_client
await db_client.drop_database(settings.test_db_name)
print("[TEST DB TEARDOWN] Test database dropped.")

@pytest_asyncio.fixture()
async def test_lifespan():
@asynccontextmanager
async def lifespan(app: FastAPI):
await init_db(settings.db_uri, settings.test_db_name)
yield
return lifespan

@pytest.fixture(autouse=True)
def test_app(test_lifespan):
app_ = FastAPI(
title="Product Catalogue Service (Test)",
docs_url="/api/docs",
openapi_url="/api/openapi",
lifespan=test_lifespan,
)
for router in routers:
app_.include_router(router.router)
return app_
< /code>
тестовые приспособления в (tests/test_attributes/conftest.py
):

Код: Выделить всё

# some imports are omitted
from product_catalouge.models.attribute import AttributeModel

@pytest.fixture()
def str_attr():
payload = {
"label": "String",
"internal_code": "String",
"type": "string",
"is_required": True,
"measurement_type":  "distance",
"unit": "cm",
"options": ["option1"]
}
return payload

@pytest.fixture()
def num_attr():
payload = {
"label": "Number",
"internal_code": "Number",
"type": "number",
"is_required": True,
"measurement_type": "distance",
"unit": "m",
"options": ["option1"]
}
return payload

@pytest.fixture()
def select_attr():
payload = {
"label": "Select",
"internal_code": "Select",
"type": "select",
"is_required": True,
"measurement_type": "distance",
"unit": "m",
"options": ["option1"]
}
return payload

@pytest.fixture(params=["str_attr", "select_attr"])
def str_select_payload(request):
return request.getfixturevalue(request.param)

@pytest.fixture()
def attrs(str_attr, num_attr, select_attr):
return [str_attr, num_attr, select_attr]

@pytest_asyncio.fixture()
async def three_attrs(testdb, attrs):
inserted_attrs = []
for attr in attrs:
inserted_attr = await AttributeModel(**attr).insert()
inserted_attrs.append(inserted_attr)
yield inserted_attrs
< /code>
Когда я запускаю этот тест: < /p>
async def test_get_all_attrs(testdb, three_attrs, test_app):
async with AsyncClient(transport=ASGITransport(test_app),
base_url="http://test",
follow_redirects=True) as ac:
res = await ac.get("/api/attributes")
assert res.status_code == 200
assert len(res.json()) == 3
< /code>
Я получаю ошибку: beanie.exceptions.collectionwasnotinitialization < /code>, приложение работает нормально, а также другие тесты.
Я подозреваю Я настраиваю тестовую базу данных или управляю асинхронным контекстом, но я не уверен, что делаю не так. Beanie Odm? Когда я получил ошибку beanie.exceptions.collectionwasnotinialialized 
, я, хотя это произошло из -за отсутствия функциональности в mongomock_motor.asyncmongomockclient , но ошибка сохраняется даже с реальным DB Comen. п>

Подробнее здесь: https://stackoverflow.com/questions/794 ... nie-odm-in
Реклама
Ответить Пред. темаСлед. тема

Быстрый ответ

Изменение регистра текста: 
Смайлики
:) :( :oops: :roll: :wink: :muza: :clever: :sorry: :angel: :read: *x)
Ещё смайлики…
   
К этому ответу прикреплено по крайней мере одно вложение.

Если вы не хотите добавлять вложения, оставьте поля пустыми.

Максимально разрешённый размер вложения: 15 МБ.

  • Похожие темы
    Ответы
    Просмотры
    Последнее сообщение
  • Не могу использовать API ODM Beanie Inside Celery/ARQ
    Anonymous » » в форуме Python
    0 Ответы
    4 Просмотры
    Последнее сообщение Anonymous
  • Asyncio Async Funcitons вешает с Asyncio.gather. (Код работает без Asyncio.gather)
    Anonymous » » в форуме Python
    0 Ответы
    23 Просмотры
    Последнее сообщение Anonymous
  • ## React Native Document Ficker Error Error Error: `GuardEdresultAsynctask` не найдена
    Anonymous » » в форуме Android
    0 Ответы
    18 Просмотры
    Последнее сообщение Anonymous
  • ## React Native Document Ficker Error Error Error: `GuardEdresultAsynctask` не найдена
    Anonymous » » в форуме Android
    0 Ответы
    18 Просмотры
    Последнее сообщение Anonymous
  • ## React Native Document Ficker Error Error Error: `GuardEdresultAsynctask` не найдена
    Anonymous » » в форуме Android
    0 Ответы
    19 Просмотры
    Последнее сообщение Anonymous

Вернуться в «Python»