Blog API: juntando, testando e rodando

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import Configuracoes, obter_configuracoes
from app.database import Base, criar_engine, criar_fabrica
from app.routers import auth, posts


def criar_app(config: Configuracoes | None = None) -> FastAPI:
    config = config or obter_configuracoes()
    engine = criar_engine(config.banco_url)

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        # Para aprender e para os testes. Em produção, as tabelas vêm de migrações (Alembic).
        Base.metadata.create_all(engine)
        yield
        engine.dispose()

    app = FastAPI(title="Blog API", version="1.0.0", lifespan=lifespan)
    app.state.config = config
    app.state.engine = engine
    app.state.fabrica_sessao = criar_fabrica(engine)

    if config.origens_permitidas:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=config.origens_permitidas,
            allow_credentials=True,
            allow_methods=["GET", "POST", "PUT", "DELETE"],
            allow_headers=["authorization", "content-type"],
        )

    app.include_router(auth.router)
    app.include_router(posts.router)

    @app.get("/saude", tags=["Infra"])
    def saude() -> dict[str, str]:
        return {"status": "ok"}

    return app


app = criar_app()

import os

# A chave precisa existir antes de importar a aplicação (que lê as configurações na partida).
os.environ.setdefault("BLOG_CHAVE_SECRETA", "chave-de-teste-com-no-minimo-32-caracteres-0123")

from collections.abc import Callable, Iterator  # noqa: E402

import pytest  # noqa: E402
from fastapi import FastAPI  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

from app.config import Configuracoes  # noqa: E402
from app.database import Base  # noqa: E402
from app.main import criar_app  # noqa: E402

# Por padrão, SQLite em memória. Para testar no PostgreSQL de verdade:
#   TEST_DATABASE_URL=postgresql+psycopg://blog:blog_senha@localhost:5432/blog_teste uv run pytest
URL_TESTE = os.environ.get("TEST_DATABASE_URL", "sqlite://")


@pytest.fixture
def app() -> Iterator[FastAPI]:
    aplicacao = criar_app(Configuracoes(banco_url=URL_TESTE, _env_file=None))
    yield aplicacao
    # Cada teste começa com o banco vazio: nenhum teste depende do que outro deixou.
    engine = aplicacao.state.engine
    Base.metadata.drop_all(engine)
    engine.dispose()  # fecha o pool: sem isso, cada teste deixaria uma conexão aberta


@pytest.fixture
def cliente(app: FastAPI) -> Iterator[TestClient]:
    with TestClient(app) as c:
        yield c


Entrar = Callable[[str, str], dict[str, str]]


@pytest.fixture
def novo_usuario(cliente: TestClient) -> Entrar:
    """Cria um usuário, faz login e devolve o cabeçalho Authorization."""

    def criar(nome: str, email: str) -> dict[str, str]:
        senha = "senha-segura-123"
        cliente.post("/usuarios", json={"nome": nome, "email": email, "senha": senha})
        resposta = cliente.post("/auth/token", data={"username": email, "password": senha})
        return {"Authorization": f"Bearer {resposta.json()['access_token']}"}

    return criar


@pytest.fixture
def ana(novo_usuario: Entrar) -> dict[str, str]:
    return novo_usuario("Ana", "ana@exemplo.com")


@pytest.fixture
def bia(novo_usuario: Entrar) -> dict[str, str]:
    return novo_usuario("Bia", "bia@exemplo.com")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.security import criar_token

CADASTRO = {"nome": "Ana", "email": "Ana@Exemplo.com", "senha": "senha-segura-123"}


def test_cadastro_nao_devolve_a_senha_nem_o_hash(cliente: TestClient) -> None:
    resposta = cliente.post("/usuarios", json=CADASTRO)
    assert resposta.status_code == 201
    corpo = resposta.json()
    assert corpo["email"] == "ana@exemplo.com"
    assert set(corpo) == {"id", "email", "nome"}


def test_email_repetido_devolve_409(cliente: TestClient) -> None:
    cliente.post("/usuarios", json=CADASTRO)
    repetido = cliente.post("/usuarios", json={**CADASTRO, "email": "ana@exemplo.com"})
    assert repetido.status_code == 409


@pytest.mark.parametrize(
    "mudanca",
    [{"senha": "curta"}, {"email": "nao-e-email"}, {"nome": ""}, {"senha": "x" * 129}],
)
def test_cadastro_invalido_devolve_422(cliente: TestClient, mudanca: dict[str, str]) -> None:
    assert cliente.post("/usuarios", json={**CADASTRO, **mudanca}).status_code == 422


def test_login_com_sucesso_e_acesso_aos_proprios_dados(cliente: TestClient) -> None:
    cliente.post("/usuarios", json=CADASTRO)
    login = cliente.post(
        "/auth/token", data={"username": "ana@exemplo.com", "password": "senha-segura-123"}
    )
    assert login.status_code == 200 and login.json()["token_type"] == "bearer"
    cabecalho = {"Authorization": f"Bearer {login.json()['access_token']}"}
    assert cliente.get("/usuarios/eu", headers=cabecalho).json()["email"] == "ana@exemplo.com"


def test_senha_errada_e_email_inexistente_dao_a_mesma_resposta(cliente: TestClient) -> None:
    cliente.post("/usuarios", json=CADASTRO)
    errada = cliente.post("/auth/token", data={"username": "ana@exemplo.com", "password": "errada"})
    inexistente = cliente.post("/auth/token", data={"username": "zeca@x.com", "password": "errada"})
    assert errada.status_code == inexistente.status_code == 401
    assert errada.json() == inexistente.json()


def test_sem_token_ou_com_token_ruim_devolve_401(cliente: TestClient) -> None:
    assert cliente.get("/usuarios/eu").status_code == 401
    ruim = cliente.get("/usuarios/eu", headers={"Authorization": "Bearer lixo"})
    assert ruim.status_code == 401
    assert ruim.headers["www-authenticate"] == "Bearer"


def test_token_vencido_e_recusado(app: FastAPI, cliente: TestClient) -> None:
    cadastro = cliente.post("/usuarios", json=CADASTRO).json()
    vencido = criar_token(app.state.config, cadastro["id"], minutos=-1)
    resposta = cliente.get("/usuarios/eu", headers={"Authorization": f"Bearer {vencido}"})
    assert resposta.status_code == 401

from fastapi.testclient import TestClient

NOVO = {"titulo": "Meu primeiro post", "conteudo": "Olá, mundo do FastAPI"}


def criar_post(
    cliente: TestClient, cabecalho: dict[str, str], **mudancas: str
) -> dict[str, object]:
    resposta = cliente.post("/posts", json={**NOVO, **mudancas}, headers=cabecalho)
    assert resposta.status_code == 201, resposta.text
    corpo: dict[str, object] = resposta.json()
    return corpo


def test_criar_exige_login(cliente: TestClient) -> None:
    assert cliente.post("/posts", json=NOVO).status_code == 401


def test_criar_e_ler(cliente: TestClient, ana: dict[str, str]) -> None:
    criado = criar_post(cliente, ana)
    assert criado["titulo"] == NOVO["titulo"] and criado["atualizado_em"] is None
    lido = cliente.get(f"/posts/{criado['id']}")
    assert lido.status_code == 200 and lido.json()["conteudo"] == NOVO["conteudo"]


def test_leitura_e_publica(cliente: TestClient, ana: dict[str, str]) -> None:
    criar_post(cliente, ana)
    assert cliente.get("/posts").status_code == 200


def test_post_inexistente_devolve_404(cliente: TestClient) -> None:
    assert cliente.get("/posts/999").status_code == 404


def test_validacao_do_corpo(cliente: TestClient, ana: dict[str, str]) -> None:
    assert cliente.post("/posts", json={"titulo": ""}, headers=ana).status_code == 422
    assert (
        cliente.post("/posts", json={"titulo": "x" * 201, "conteudo": "c"}, headers=ana).status_code
        == 422
    )


def test_so_o_autor_altera_e_remove(
    cliente: TestClient, ana: dict[str, str], bia: dict[str, str]
) -> None:
    post_id = criar_post(cliente, ana)["id"]
    novo = {"titulo": "Título novo", "conteudo": "Conteúdo novo"}

    assert cliente.put(f"/posts/{post_id}", json=novo, headers=bia).status_code == 403
    assert cliente.delete(f"/posts/{post_id}", headers=bia).status_code == 403
    assert cliente.get(f"/posts/{post_id}").json()["titulo"] == NOVO["titulo"]

    alterado = cliente.put(f"/posts/{post_id}", json=novo, headers=ana)
    assert alterado.status_code == 200 and alterado.json()["titulo"] == "Título novo"
    assert alterado.json()["atualizado_em"] is not None

    assert cliente.delete(f"/posts/{post_id}", headers=ana).status_code == 204
    assert cliente.get(f"/posts/{post_id}").status_code == 404
    assert cliente.delete(f"/posts/{post_id}", headers=ana).status_code == 404


def test_paginacao(cliente: TestClient, ana: dict[str, str]) -> None:
    for i in range(25):
        criar_post(cliente, ana, titulo=f"Post {i:02d}")

    primeira = cliente.get("/posts?limite=10").json()
    assert (primeira["total"], primeira["paginas"], len(primeira["dados"])) == (25, 3, 10)
    assert primeira["dados"][0]["titulo"] == "Post 24"  # os mais novos primeiro

    ultima = cliente.get("/posts?limite=10&pagina=3").json()
    assert len(ultima["dados"]) == 5
    assert cliente.get("/posts?pagina=99").json()["dados"] == []

    ids = [
        p["id"]
        for n in (1, 2, 3)
        for p in cliente.get(f"/posts?limite=10&pagina={n}").json()["dados"]
    ]
    assert len(ids) == len(set(ids)) == 25
    assert cliente.get("/posts?limite=51").status_code == 422
    assert cliente.get("/posts?pagina=0").status_code == 422


def test_busca_no_titulo_e_no_conteudo(cliente: TestClient, ana: dict[str, str]) -> None:
    criar_post(cliente, ana, titulo="Introdução ao FastAPI", conteudo="rotas e modelos")
    criar_post(cliente, ana, titulo="Banco de dados", conteudo="usando SQLAlchemy")
    criar_post(cliente, ana, titulo="Outro assunto", conteudo="sobre fastapi também")

    por_titulo = cliente.get("/posts?busca=introdução").json()
    assert por_titulo["total"] == 1
    sem_caixa = cliente.get("/posts?busca=FASTAPI").json()
    assert sem_caixa["total"] == 2
    assert cliente.get("/posts?busca=nada-disso").json()["total"] == 0


def test_busca_trata_o_percentual_como_texto(cliente: TestClient, ana: dict[str, str]) -> None:
    criar_post(cliente, ana, titulo="Promoção de 50% hoje")
    criar_post(cliente, ana, titulo="Sem símbolo")
    assert cliente.get("/posts?busca=%25").json()["total"] == 1


def test_filtrar_por_autor(cliente: TestClient, ana: dict[str, str], bia: dict[str, str]) -> None:
    criar_post(cliente, ana)
    criar_post(cliente, bia)
    criar_post(cliente, bia)
    autor_bia = cliente.get("/usuarios/eu", headers=bia).json()["id"]
    assert cliente.get(f"/posts?autor_id={autor_bia}").json()["total"] == 2


def test_resposta_nunca_expoe_dados_do_usuario(cliente: TestClient, ana: dict[str, str]) -> None:
    post = criar_post(cliente, ana)
    assert set(post) == {"id", "titulo", "conteudo", "autor_id", "criado_em", "atualizado_em"}


def test_saude(cliente: TestClient) -> None:
    assert cliente.get("/saude").json() == {"status": "ok"}

uv sync
uv run pytest

TEST_DATABASE_URL="postgresql+psycopg://blog:blog_senha@localhost:5432/blog_teste" uv run pytest

uv run mypy
uv run ruff check .

export BLOG_CHAVE_SECRETA="$(openssl rand -hex 32)"
export BLOG_BANCO_URL="postgresql+psycopg://blog:blog_senha@localhost:5432/blog_db"
uv run fastapi run app/main.py

POST /usuarios               201 {"id": 1, "email": "ana@exemplo.com", "nome": "Ana"}
POST /usuarios (repetido)    409 {"detail": "E-mail já cadastrado"}
POST /auth/token             200 token_type=bearer partes_do_jwt=3
POST /posts (sem token)      401 {"detail": "Not authenticated"}
POST /posts                  201 {"id": 3, "titulo": "Post 3 sobre FastAPI", "autor_id": 1}
GET /posts?limite=2          200 total=3 paginas=2 titulos=['Post 3 sobre FastAPI', 'Post 2 sobre FastAPI']
GET /posts?busca=Post 2      200 total=1
PUT /posts/1 (sem token)     401 {"detail": "Not authenticated"}
DELETE /posts/3              204 ""
GET /posts/3                 404 {"detail": "Post não encontrado"}

                                        Table "public.posts"
    Column     |           Type           | Collation | Nullable |              Default
---------------+--------------------------+-----------+----------+-----------------------------------
 id            | integer                  |           | not null | nextval('posts_id_seq'::regclass)
 titulo        | character varying(200)   |           | not null |
 conteudo      | text                     |           | not null |
 autor_id      | integer                  |           | not null |
 criado_em     | timestamp with time zone |           | not null | now()
 atualizado_em | timestamp with time zone |           |          |
Indexes:
    "posts_pkey" PRIMARY KEY, btree (id)
    "ix_posts_autor_id" btree (autor_id)
Foreign-key constraints:
    "posts_autor_id_fkey" FOREIGN KEY (autor_id) REFERENCES usuarios(id) ON DELETE CASCADE