how to use db session in router level unit test

[zzmao]

Hi,

I have added fastapi-sqlalchemy to my app and it runs well if I start from main.py, where

app.add_middleware(DBSessionMiddleware, db_url=settings.DATABASE_RUL)

is added.

Now I am trying to build router-level unit test, which uses client = TestClient(router) to create router level client. So it won't run the code in main. the test failed and reports:

    @property
    def session(self) -> Session:
        """Return an instance of Session local to the current async context."""
        if _Session is None:
>           raise SessionNotInitialisedError
E           fastapi_sqlalchemy.exceptions.SessionNotInitialisedError: 
E                   Session not initialised! Ensure that DBSessionMiddleware has been initialised before
E                   attempting database access.

Any suggestion to use it in router level unit test?

[zzmao]

@mfreeborn any suggestion?

[rakib-09]

@zzmao did you get any solution of this?

[lfvilella]

One QUICK WAY to fix this is using unittest.mock.patch, example:

conftest.py

import unittest.mock

import pytest
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy_utils import create_database, database_exists

from fastapi_sqlalchemy import db

from app.config import config
from app import models


@pytest.fixture
def use_db():
    if not database_exists(config.DB_URL):
        create_database(config.DB_URL)

    engine = create_engine(config.DB_URL, pool_pre_ping=True)
    models.Base.metadata.create_all(engine)
    _Session = sessionmaker(bind=engine)

    with unittest.mock.patch(    # !! HERE !! 
            'fastapi_sqlalchemy.middleware._Session',
            _Session,
            ):
        with db():
            yield None

    models.Base.metadata.drop_all(engine)
    engine.dispose()

test_dummy_model.py

import uuid

import pytest
from fastapi_sqlalchemy import db

from app import models


@pytest.mark.usefixtures('use_db')
class TestDummyModel:
    def test_create(self):
        assert len(db.session.query(models.DummyModel).all()) == 0
        instance = models.DummyModel(foo='foo', bar='bar')
        assert not instance.id

        db.session.add(instance)
        db.session.commit()
        assert len(db.session.query(models.DummyModel).all()) == 1
        
        instance_id = db.session.query(models.DummyModel).first().id
        assert isinstance(instance_id, uuid.UUID)

[goodwind]

I am doing it this way. In my case it's service-level testing:

import aiounittest
from api.v1.stock.service import StockService
from core.config import config
from fastapi_sqlalchemy import DBSessionMiddleware, db
from main import app
from core.exceptions import NotFoundException

DBSessionMiddleware(app=app, db_url=config.DB_URL)

class test_stock(aiounittest.AsyncTestCase):
    async def test_get_stock_by_id(self):
        with db():
            result = await StockService().get_stock_by_id(stock_id=235)
            self.assertTrue(result.employee_account_login == "test")

            with self.assertRaises(NotFoundException):
                result = await StockService().get_stock_by_id(stock_id=99999)