How to test GraphQL endpoint using separate database for testing in FastAPI Strawberry?

Question

I have set up 2 databases in my project. One is primary which I would use for day to day use and the other is just for testing. The problem is that my Strawberry fields all use primary database and I don't know how can I properly replace that primary db with testing db.

I know that in FastAPI there's something like dependency injection with the help of which I can replace primary database connection dependency with testing database connection dependency but I don't seem to find similar solution in Strawberry.

I am using sqlalchemy with 2 postgres databases(primary and for testing), alembic for migrations and pytest with requests to perform tests.

What I'm trying to achieve:

1. I want to apply migrations to my database for testing with alembic.
2. Replace primary database session in all the Strawberry fields with test database session.
3. Perform tests on GraphQL endpoints with requests module.

Answer 1

So requests library is more suitable for an end to end testing and it's much harder to do that with a separate database, so I replaced that part with

strawberry.Schema(query=Query, mutation=Mutation).execute_sync(query="...", context_value=CustomContext())

And here's more code of what I did to accomplish what I wanted:

1. I replaced hard coded SessionLocal() in my endpoints with get_db() function which is being passed with a context to the endpoint:

   # app/db/database.py
   
   ... # Some boiler plate code to work with database from FastAPI documentation
   
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   

   # app/graphql/context.py
   
   from strawberry.fastapi import BaseContext
   from app.db.database import get_db
   
   
   class Context(BaseContext):
   
       ... # Some of mine additional context code in here
   
       def get_db(self):
           return get_db()
   
   
   def get_context() -> Context:
       return Context()
   

   # app/graphql/inputs.py
   
   from app.graphql.context import Context
   from strawberry.types.info import RootValueType
   
   Info = _Info[Context, RootValueType]
   

   # app/graphql/controllers.py
   
   from app.graphql.inputs import Info
   
   class Meeting:
       def get_profiles_within(distance: int, info: Info):
           sess = next(info.context.get_db()) # this is how I now access the db session
   
           ... # The rest of my controller code
   

   # app/main.py
   
   import strawberry
   from strawberry.fastapi import GraphQLRouter
   
   from app.graphql.core import Query, Mutation
   from app.graphql.context import get_context
   
   schema = strawberry.Schema(query=Query, mutation=Mutation)
   graphql_app = GraphQLRouter(schema, context_getter=get_context) # This is where previously defined get_context function is used 
   

2. Then I did pretty much the exact same thing for my second database. I created get_test_db() function for it and CustomContext class for it(It would've been better if I'd named it TestContext, but because this class of mine is located in the exact same file as my testing code I can't do this, because pytest automatically detects it as test):

   # app/tests/db.py
   
   ... # Boiler plate code from FastAPI documentation but this time SQLALCHEMY_DATABASE_URL stores link to my test database
   
   def get_test_db():
       db = TestSessionLocal()
       try:
           yield db
       finally:
           db.close()
   

   # app/tests/test_meeting.py
   
   from app.tests import db
   from strawberry.fastapi import BaseContext
   
   class CustomContext(BaseContext):
   
       ... # The rest of my context, which is exactly the some code as in my regular Context class
   
       def get_db(self):
           return db.get_test_db()
   

3. And for the last part to perform mocking of the database connection I just needed to replace context when I query graphql:

   # app/tests/test_meeting.py
   
   from app.main import schema
   
   class TestMeeting():
       def test_seeking_people_around(self):
   
           ... # other testing code
   
           response = schema.execute_sync(query=test_query, context_value=CustomContext())
   

   by default schema doesn't have any context_value to it so I need to provide it this way for my code to work and that's exactly what I need.

Hope this helps someone who struggles setting up separate database for testing!