Async Tests — pytest-asyncio & AsyncClient

# pip install pytest-asyncio
# pyproject.toml:
# [tool.pytest.ini_options]
# asyncio_mode = "auto"                      # async def tests just work
# asyncio_default_fixture_loop_scope = "session"   # ← THE gotcha fix:
#   fixtures + tests share one loop; without it, session-scoped async
#   fixtures die with "Future attached to a different loop"

import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app

@pytest.fixture
async def aclient():
    transport = ASGITransport(app=app)       # in-process, no server — TestClient's
    async with AsyncClient(transport=transport,
                           base_url="http://test") as c:
        yield c

async def test_create_student(aclient, async_db):        # async db fixture
    # arrange — AWAIT the seed (impossible around sync TestClient):
    await seed_students(async_db, count=3)

    # act
    resp = await aclient.get("/students?branch=CS")

    # assert on response AND database state:
    assert resp.status_code == 200
    row = (await async_db.execute(
        select(func.count()).select_from(Student))).scalar()
    assert row == 3

# ── concurrency bugs only async tests can catch ──
import asyncio

async def test_no_double_registration_race(aclient):
    """Two simultaneous registrations for the last seat: exactly one wins."""
    results = await asyncio.gather(
        aclient.post("/drives/7/register/1"),
        aclient.post("/drives/7/register/1"),
    )
    codes = sorted(r.status_code for r in results)
    assert codes == [201, 409]        # one created, one conflict — not [201, 201]!
# TestClient cannot fire truly concurrent requests; gather can.
# This test catches the missing unique-constraint/locking bug.

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_chat_echo_and_broadcast():
    with client.websocket_connect("/ws/drives/7/chat") as ws:
        joined = ws.receive_json()                   # the "someone joined" sys msg
        assert joined["sys"] == "someone joined"

        ws.send_json({"name": "Asha", "text": "TCS drive when?"})
        msg = ws.receive_json()
        assert msg == {"from": "Asha", "text": "TCS drive when?"}
    # context-manager exit = clean disconnect — manager cleanup runs

def test_ws_rejects_bad_token():
    import pytest
    from starlette.websockets import WebSocketDisconnect
    with pytest.raises(WebSocketDisconnect) as exc:
        with client.websocket_connect("/ws/notifications?token=garbage"):
            pass
    assert exc.value.code == 1008                    # policy violation

def test_two_clients_see_each_other():
    with client.websocket_connect("/ws/drives/7/chat") as a, \
         client.websocket_connect("/ws/drives/7/chat") as b:
        a.receive_json(); b.receive_json(); a.receive_json()   # join notices
        a.send_json({"name": "Asha", "text": "hi"})
        assert b.receive_json()["text"] == "hi"      # broadcast reached b

# ── Which client, when ──
#   plain endpoint tests, validation, auth   → TestClient (sync, simplest)
#   need to await around the call            → AsyncClient (async DB, seeding)
#   concurrent-request race tests            → AsyncClient + gather
#   WebSocket conversations                  → TestClient.websocket_connect
#   SSE/streaming                            → client.stream(...) — iterate
#                                              chunks, assert frames arrive
# Mixed suites are normal: 80% sync tests, async where awaiting IS the test.