# Testing, Deployment and Streaming

Companion to the FastAPI skill: testing an app built with `create_app()` and a lifespan, and running it in production.

## Testing

```python
# tests/conftest.py
import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient

from app.main import create_app
from app.auth import get_current_user

@pytest.fixture
async def client(test_user):
    app = create_app()
    app.dependency_overrides[get_current_user] = lambda: test_user
    async with LifespanManager(app):
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as c:
            yield c
    app.dependency_overrides.clear()
```

```python
# tests/test_notes.py
async def test_create_rejects_unknown_fields(client):
    res = await client.post("/notes", json={"title": "a", "body": "b", "owner_id": 9})
    assert res.status_code == 422

async def test_create_then_read(client):
    created = await client.post("/notes", json={"title": "Shopping", "body": "milk"})
    assert created.status_code == 201
    note_id = created.json()["id"]
    fetched = await client.get(f"/notes/{note_id}")
    assert fetched.json()["title"] == "Shopping"
```

- `ASGITransport` does not run the lifespan; wrap the app in
  `asgi_lifespan.LifespanManager` (or use the sync `TestClient` as a context
  manager) so resources exist.
- Run async tests with `pytest-asyncio` (`asyncio_mode = "auto"`) or AnyIO's
  pytest plugin.
- Override dependencies (`get_settings`, `get_current_user`, `get_session`)
  rather than monkeypatching modules.
- For database tests, run Postgres in Testcontainers and wrap each test in a
  transaction that is rolled back, or truncate between tests.

## Running in production

- `fastapi run` (Uvicorn under the hood) or Uvicorn directly behind a reverse
  proxy; one process per container and scale horizontally, or use
  `--workers` on a single VM.
- Set `--proxy-headers` and `--forwarded-allow-ips` to your load balancer so
  client IPs and schemes are correct.
- Disable or protect `/docs` and `/openapi.json` on internal-only APIs if
  exposure is a concern.
- Instrument with OpenTelemetry (`opentelemetry-instrumentation-fastapi`,
  `-sqlalchemy`, `-httpx`) and emit structured JSON logs.

## Streaming and WebSockets

- `StreamingResponse` with an async generator for large exports; server-sent
  events for one-way live updates.
- WebSockets: authenticate during the handshake (token in a query parameter or
  subprotocol, since browsers cannot set headers), handle `WebSocketDisconnect`,
  and use a broker (Redis pub/sub) when more than one process must broadcast.