A primeira coisa que temos que montar é uma fixture da sessão do banco em tests/conftest.py
import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from fast_zero.models import table_registry
@pytest.fixture
def session():
engine = create_engine('sqlite:///:memory:')
table_registry.metadata.create_all(engine)
with Session(engine) as session:
yield session
table_registry.metadata.drop_all(engine)
create_engine('sqlite:///:memory:')
: cria um mecanismo de banco de dados SQLite em memória usando SQLAlchemy. Este mecanismo será usado para criar uma sessão de banco de dados para nossos testes.
table_registry.metadata.create_all(engine)
: cria todas as tabelas no banco de dados de teste antes de cada teste que usa a fixture session
.
with Session(engine) as session
: cria uma sessão Session
para que os testes possam se comunicar com o banco de dadosvia engine
.
yield session
: fornece uma instância de Session que será injetada em cada teste que solicita a fixture session
. Essa sessão será usada para interagir com o banco de dados de teste.
table_registry.metadata.drop_all(engine)
: após cada teste que usa a fixture session
, todas as tabelas do banco de dados de teste são eliminadas, garantindo que cada teste seja executado contra um banco de dados limpo.
Agora nosso teste
from sqlalchemy import select
from fast_zero.models import User
def test_create_user(session):
new_user = User(username='alice', password='secret', email='teste@test')
session.add(new_user)
session.commit()
user = session.scalar(select(User).where(User.username == 'alice'))
assert user.username == 'alice'
task test
Temos um problema nesse teste que pode tornar ele complicado de validar. Validamos somente o nome alice
, mas não validamos o objeto todo. Isso é um tanto quanto complicado. Pois para validar o objeto inteiro, precisamos saber a que horas ele foi criado, por conta do campo init=False
.
Ele inviabiliza o envido de um dado determinístico ao objeto.
Para atuar em cenários assim, podemos "roubar nos testes" usando eventos do SQLAlchemy.
A ideia por trás dos eventos é fazer alguma operação antes ou depois de alguma operação.
Chamamos o objeto event
do SQLalchemy para "ouvir" uma operação:
from sqlalchemy import event
def hook(mapper, connection, target):
...
event.listen(User, 'before_insert', hook)
Nesse caso, estamos ouvindo before_insert
. O que significa que ele executará a função hook
antes de inserir no banco de fato.
from contextlib import contextmanager
from datetime import datetime
from sqlalchemy import create_engine, event
# ...
@contextmanager
def _mock_db_time(*, model, time=datetime(2024, 1, 1)):
def fake_time_hook(mapper, connection, target):
if hasattr(target, 'created_at'):
target.created_at = time
event.listen(model, 'before_insert', fake_time_hook)
yield time
event.remove(model, 'before_insert', fake_time_hook)
@contextmanager
def _mock_db_time(*, model, time=datetime(2024, 1, 1)):
def fake_time_hook(mapper, connection, target):
if hasattr(target, 'created_at'):
target.created_at = time
event.listen(model, 'before_insert', fake_time_hook)
yield time
event.remove(model, 'before_insert', fake_time_hook)
Antes de executar o insert a função fake_time_hook
vai alterar o created_at
para o valor default do parâmetro time
. Fazendo que o ele não use o valor padrão do datetime do db.
O contextmanager
faz com que a função possa ser usada com o bloco with
.
Agora que temos a função gerenciadora de contexto, para evitar o sistema de importação durante os testes, podemos criar uma fixture para ele.
De forma bem simples, somente retornando a função _mock_db_time
:
@pytest.fixture
def mock_db_time():
return _mock_db_time
Dessa forma podemos fazer a chamada direta no teste.
from dataclasses import asdict
from sqlalchemy import select
from fast_zero.models import User
def test_create_user(session, mock_db_time):
with mock_db_time(model=User) as time:
new_user = User(
username='alice', password='secret', email='teste@test'
)
session.add(new_user)
session.commit()
user = session.scalar(select(User).where(User.username == 'alice'))
assert asdict(user) == {
'id': 1,
'username': 'alice',
'password': 'secret',
'email': 'teste@test',
'created_at': time,
}
Dessa forma todos os campos, até os que são manipulados diretamente pelo ORM podem ser testados.
assert asdict(user) ==
'id': 1,
'username': 'alice',
'password': 'secret',
'email': 'teste@test',
'created_at': time,
}
Uma boa prática no desenvolvimento de aplicações é separar as configurações do código.
Configurações, como credenciais de banco de dados, são propensas a mudanças entre ambientes diferentes (como desenvolvimento, teste e produção).
Misturá-las com o código pode tornar o processo de mudança entre esses ambientes complicado e propenso a erros.
poetry add pydantic-settings
#fast_zero/settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file='.env', env_file_encoding='utf-8'
)
DATABASE_URL: str
.env
Esse configuração permite que usemos arquivos .env
para não inserir dados do banco no código fonte
DATABASE_URL="sqlite:///database.db"
Não podemos esquecer de adicionar essa base de dados no .gitignore
echo 'database.db' >> .gitignore
Antes de avançarmos, é importante entender o que são migrações de banco de dados e por que são úteis.
poetry add alembic
alembic init migrations
.
├── .env
├── alembic.ini <-
├── fast_zero
│ ├── __init__.py
│ ├── app.py
│ ├── models.py
│ └── schemas.py
├── migrations <-
│ ├── env.py
│ ├── README
│ ├── script.py.mako
│ └── versions
├── poetry.lock
├── pyproject.toml
├── README.md
└── tests
├── __init__.py
├── conftest.py
├── test_app.py
└── test_db.py
Vamos fazer algumas alterações no arquivo migrations/env.py
para que nossa configurações de banco de dados sejam passadas ao alembic:
Settings
do nosso arquivo settings.py
e a table_registry
dos nossos modelos.Settings
.table_registry.metadata
, que é o que o Alembic utilizará para gerar automaticamente as migrações.from alembic import context
from fast_zero.settings import Settings
from fast_zero.models import table_registry
config = context.config
config.set_main_option('sqlalchemy.url', Settings().DATABASE_URL)
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = table_registry.metadata
alembic revision --autogenerate -m "create users table"
alembic upgrade head
Pra isso poderíamos usar uma ferramenta gráfica ou usando a CLI do sqlite:
python -m sqlite3 database.db
# Abrirá o shell do sqlite3
sqlite>
select * from alembic_version;
select * from users;
User
) e adicionar um campo chamado updated_at
:
datetime
init=False
now
mapped_column(onupdate=func.now())
mock_db_time
) para ser contemplado no mock o campo updated_at
na validação do teste.Obviamente, não esqueça de responder ao quiz da aula
git add .
git commit -m "Adicionada a primeira migração com Alembic. Criada tabela de usuários."
git push
mermaid.js