Closed
Description
Hi, I having issue passing my tests because of attached to a diffrent loop error.
I tried #38. but it doesn't fix it.
I'm using SQLAlchemy 1.4 and FastAPI.
example:
async def create(self, db: AsyncSession, obj_in: CreateSchemaType) -> ModelType:
obj_in_data = obj_in.dict()
db_obj = self.model(**obj_in_data)
db.add(db_obj)
await db.commit()
await db.refresh(db_obj)
return db_obj
async def create_random_country(db: AsyncSession) -> models.Country:
data = {"name": random_lower_string(5), "calling_code": random_number(4)}
country_in = CountryCreationSchema(**data)
return await crud.country.create(db, country_in)
@pytest.mark.asyncio
async def test_get_countries(
db: AsyncSession
):
random_country = await create_random_country(db)
async with AsyncClient(app=app, base_url=settings.BASE_URL) as ac:
r = await ac.get(f"{settings.API_V1_STR}/countries")
countries = r.json()
assert r.status_code == 200
assert random_country in countries
assert "name" in countries[0]
error:
app/tests/utils/core.py:19: in create_random_country
return await crud.country.create(db, country_in)
app/crud/base.py:46: in create
await db.commit()
/usr/local/src/sqlalchemy/lib/sqlalchemy/ext/asyncio/session.py:224: in commit
return await greenlet_spawn(self.sync_session.commit)
/usr/local/src/sqlalchemy/lib/sqlalchemy/util/_concurrency_py3k.py:91: in greenlet_spawn
result = context.throw(*sys.exc_info())
/usr/local/src/sqlalchemy/lib/sqlalchemy/orm/session.py:1320: in commit
self._transaction.commit(_to_root=self.future)
......
--------------
E RuntimeError: Task <Task pending name='Task-3' coro=<test_get_countries() running at /process_maker_ema/app/tests/api_v1/test_core.py:21> cb=[_run_until_complete_cb() at /usr/local/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop
what's the problem?
Metadata
Metadata
Assignees
Labels
No labels