My endpoints work when i run uvicorn but I cant get the testing part right. I tried looking online for a solution and nothing works.
I know the attribute error is caused since the method is missing, but there has to be a way in which we should be able to create pytest fixtures for async webhook client right?
Fastapi documentation is missing info on testing with the asyncclient and webhooks. Would really appreciate some help!
Here is the conftest.py file
```#conftest.py
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from httpx._transports.asgi import ASGITransport
from app.db.session import DATABASE_URL, get_session
from app.main import app
from httpx import AsyncClient
@pytest_asyncio.fixture
async def async_engine() -> AsyncEngine:
# Create an asynchronous engine for tests
engine = create_async_engine(DATABASE_URL, echo=True)
return engine
@pytest_asyncio.fixture
async def test_session(async_engine: AsyncEngine):
"""Create a new database session for a test."""
connection = await async_engine.connect()
transaction = await connection.begin()
test_session = sessionmaker(
bind=connection, class_=AsyncSession, expire_on_commit=False
)()
yield test_session
await transaction.rollback()
await connection.close()
@pytest_asyncio.fixture
def override_get_session(test_session):
async def _override_get_session():
yield test_session
return _override_get_session
@pytest_asyncio.fixture
async def async_client(override_get_session):
app.dependency_overrides[get_session] = override_get_session
async with AsyncClient(
transport=ASGITransport(app=app), base_url="https://topok.com"
) as client:
yield client
app.dependency_overrides.clear()
@pytest_asyncio.fixture
async def websocket_client(async_client):
async with async_client.websocket_connect("/ws/123") as websocket:
yield websocket```
Just for clarity I added the routers file so you have full context of what is going on
```#app/routers/websocket.py
from fastapi import WebSocket, APIRouter, WebSocketDisconnect
from app.services.ws_connection_manager import ws_conn_manager
router = APIRouter(
prefix="/ws",
tags=["ws"],
responses={404: {"description": "Not found"}},
)
@router.websocket("/{transaction_id}")
async def websocket_transaction_endpoint(websocket: WebSocket, transaction_id: str):
await ws_conn_manager.connect(transaction_id, websocket)
verification_message = f"Connection established for transaction {transaction_id}."
await ws_conn_manager.broadcast_to_transaction(verification_message, transaction_id)
try:
while True:
data = await websocket.receive_text()
# Optionally handle received data here. For now, just echo it back.
await ws_conn_manager.broadcast_to_transaction(
f"Echo to {transaction_id}: {data}", transaction_id
)
print(f"Received data: {data}")
except WebSocketDisconnect:
ws_conn_manager.disconnect(transaction_id, websocket)```
Here is the test file. I know this does not overlap with the code above with transaction id, but for now I just need it to work. Adding the id later will be fine
```#test_websocket.py
@pytest.mark.asyncio
async def test_websocket_interaction(websocket_client):
await websocket_client.send_text("Message for transaction Hello World!")
response = await websocket_client.receive_text()
assert response == "Message for transaction Hello World"
```