Testes de Banco de Dados: SQL, NoSQL e Data Integrity

---

Introdução aos Testes de Banco de Dados

Testar banco de dados é essencial para garantir que dados são armazenados, recuperados e manipulados corretamente. Inclui testes de schema, queries, migrations e performance.

Por Que Testar Banco de Dados?

• Garantir integridade de dados
1. Validar constraints e triggers
2. Verificar performance de queries
3. Proteger contra regressions
4. Validar migrations

---

Tipos de Testes de Banco de Dados

1. Testes de Schema

# Testes de Schema
def test_schema_has_required_tables(postgres):
    """Verifica que tabelas existem"""
    result = postgres.run("""
        SELECT table_name 
        FROM information_schema.tables 
        WHERE table_schema = 'public'
    """)

tables = [row[0] for row in result.fetchall()]

assert 'users' in tables
    assert 'products' in tables
    assert 'orders' in tables

def test_users_table_columns(postgres):
    """Verifica colunas da tabela users"""
    result = postgres.run("""
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = 'users'
        ORDER BY ordinal_position
    """)

columns = {row[0]: {'type': row[1], 'nullable': row[2]} 
               for row in result.fetchall()}

assert 'id' in columns
    assert columns['id']['type'] == 'integer'
    assert 'email' in columns
    assert columns['email']['is_nullable'] == 'NO'

def test_primary_keys(postgres):
    """Verifica primary keys"""
    result = postgres.run("""
        SELECT tc.table_name, kcu.column_name
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu
            ON tc.constraint_name = kcu.constraint_name
        WHERE tc.constraint_type = 'PRIMARY KEY'
    """)

primary_keys = {row[0]: row[1] for row in result.fetchall()}

assert primary_keys['users'] == 'id'
    assert primary_keys['products'] == 'id'

2. Testes de Constraints

def test_unique_constraint_email(postgres):
    """Email deve ser único"""
    # Insert primeiro usuário
    postgres.run("""
        INSERT INTO users (name, email, password)
        VALUES ('User 1', 'test@example.com', 'hash')
    """)

# Tentar inserir email duplicado
    with pytest.raises(Exception) as exc:
        postgres.run("""
            INSERT INTO users (name, email, password)
            VALUES ('User 2', 'test@example.com', 'hash2')
        """)

assert 'unique constraint' in str(exc.value).lower()

def test_foreign_key_constraint(postgres):
    """Order deve referenciar user válido"""
    with pytest.raises(Exception):
        postgres.run("""
            INSERT INTO orders (user_id, total)
            VALUES (99999, 100.00)  # user_id não existe
        """)

def test_check_constraint(postgres):
    """Preço deve ser positivo"""
    with pytest.raises(Exception):
        postgres.run("""
            INSERT INTO products (name, price)
            VALUES ('Invalid Product', -10.00)
        """)

def test_not_null_constraint(postgres):
    """Nome do usuário não pode ser nulo"""
    with pytest.raises(Exception):
        postgres.run("""
            INSERT INTO users (email, password)
            VALUES ('test@example.com', 'hash')
        """)

3. Testes de Queries

def test_select_users_by_email(postgres):
    """Busca usuário por email"""
    # Setup
    postgres.run("""
        INSERT INTO users (name, email, password)
        VALUES ('João Silva', 'joao@example.com', 'hash')
    """)

# Execute
    result = postgres.run("""
        SELECT name, email 
        FROM users 
        WHERE email = 'joao@example.com'
    """)

# Assert
    assert len(result.fetchall()) == 1
    row = result.fetchone()
    assert row[0] == 'João Silva'
    assert row[1] == 'joao@example.com'

def test_join_orders_with_users(postgres):
    """Join orders com users"""
    # Setup
    user_id = insert_user(postgres, 'João', 'joao@example.com')
    insert_order(postgres, user_id, 100.00)

# Execute
    result = postgres.run("""
        SELECT u.name, u.email, o.total
        FROM users u
        JOIN orders o ON u.id = o.user_id
        WHERE u.id = :user_id
    """, {'user_id': user_id})

row = result.fetchone()
    assert row[0] == 'João'
    assert row[2] == 100.00

def test_aggregate_query(postgres):
    """Query com agregação"""
    # Setup
    user_id = insert_user(postgres, 'João', 'joao@example.com')
    insert_order(postgres, user_id, 100.00)
    insert_order(postgres, user_id, 200.00)

# Execute
    result = postgres.run("""
        SELECT SUM(total) as total_sum, COUNT(*) as order_count
        FROM orders
        WHERE user_id = :user_id
    """, {'user_id': user_id})

row = result.fetchone()
    assert row[0] == 300.00
    assert row[1] == 2

4. Testes de Migrations

import subprocess
import os

def test_migration_up(postgres):
    """Executa migration up"""
    result = subprocess.run(
        ['python', 'manage.py', 'migrate'],
        capture_output=True,
        text=True
    )

assert result.returncode == 0
    assert 'Migration complete' in result.stdout or 'No migrations to apply' in result.stdout

def test_migration_down(postgres):
    """Rollback migration"""
    migration_name = 'add_phone_to_users'

# Rollback
    result = subprocess.run(
        ['python', 'manage.py', 'migrate', migration_name, 'down'],
        capture_output=True,
        text=True
    )

assert result.returncode == 0

# Verificar que coluna foi removida
    result = postgres.run("""
        SELECT column_name 
        FROM information_schema.columns 
        WHERE table_name = 'users' AND column_name = 'phone'
    """)

assert len(result.fetchall()) == 0

def test_data_integrity_after_migration(postgres):
    """Verifica integridade após migration"""
    # Setup data
    postgres.run("""
        INSERT INTO users (name, email, password)
        VALUES ('Test', 'test@example.com', 'hash')
    """)

# Run migration
    subprocess.run(['python', 'manage.py', 'migrate'], check=True)

# Verify data intact
    result = postgres.run("""
        SELECT COUNT(*) FROM users WHERE email = 'test@example.com'
    """)

assert result.fetchone()[0] == 1

---

TestContainers para Testes de DB

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope='session')
def postgres():
    with PostgresContainer("postgres:15") as pg:
        # Setup schema
        db_url = pg.get_connection_url()

# Run migrations
        subprocess.run(['python', 'manage.py', 'migrate'], check=True)

yield pg

# Teardown

@pytest.fixture
def clean_db(postgres):
    """Limpa dados antes de cada teste"""
    db = psycopg2.connect(postgres.get_connection_url())
    cursor = db.cursor()

# Disable triggers para speed
    cursor.execute("SET session_replication_role = 'replica'")

# Truncate tables (exceto immutable)
    cursor.execute("""
        TRUNCATE TABLE orders, users, products CASCADE
    """)

db.commit()
    cursor.close()
    db.close()

yield

@pytest.fixture
def sample_data(postgres, clean_db):
    """Popula dados de exemplo"""
    db = psycopg2.connect(postgres.get_connection_url())
    cursor = db.cursor()

# Insert users
    cursor.execute("""
        INSERT INTO users (name, email, password)
        VALUES 
            ('User 1', 'user1@example.com', 'hash1'),
            ('User 2', 'user2@example.com', 'hash2')
        RETURNING id
    """)
    user_ids = [row[0] for row in cursor.fetchall()]

# Insert products
    cursor.execute("""
        INSERT INTO products (name, price, stock)
        VALUES 
            ('Product A', 10.00, 100),
            ('Product B', 20.00, 50)
        RETURNING id
    """)
    product_ids = [row[0] for row in cursor.fetchall()]

db.commit()
    cursor.close()
    db.close()

yield {
        'user_ids': user_ids,
        'product_ids': product_ids
    }

---

Testes de NoSQL

MongoDB

from pymongo import MongoClient
from datetime import datetime

@pytest.fixture
def mongodb():
    with MongoDBContainer("mongo:7") as mongo:
        client = MongoClient(mongo.get_connection_url())
        yield client['testdb']
        client.close()

def test_mongodb_insert(mongodb):
    """Insere e busca documento"""
    result = mongodb.users.insert_one({
        'name': 'João',
        'email': 'joao@example.com',
        'created_at': datetime.now()
    })

assert result.inserted_id is not None

found = mongodb.users.find_one({'email': 'joao@example.com'})
    assert found['name'] == 'João'

def test_mongodb_aggregation(mongodb):
    """Testa aggregation pipeline"""
    # Insert orders
    mongodb.orders.insert_many([
        {'user_id': 1, 'total': 100, 'status': 'completed'},
        {'user_id': 1, 'total': 200, 'status': 'completed'},
        {'user_id': 2, 'total': 50, 'status': 'pending'},
    ])

pipeline = [
        {'$match': {'status': 'completed'}},
        {'$group': {'_id': '$user_id', 'total': {'$sum': '$total'}}}
    ]

results = list(mongodb.orders.aggregate(pipeline))

assert len(results) == 1
    assert results[0]['total'] == 300

Redis

import redis

@pytest.fixture
def redis_client():
    with RedisContainer("redis:7") as redis:
        client = redis.get_connection_url(client_class=redis.Redis)
        yield client

def test_redis_cache(redis_client):
    """Testa caching com Redis"""
    # Set value
    redis_client.setex('user:1:name', 3600, 'João')

# Get value
    name = redis_client.get('user:1:name')

assert name == 'João'

def test_redis_sorted_set(redis_client):
    """Testa sorted set para ranking"""
    # Add scores
    redis_client.zadd('leaderboard', {'user1': 100, 'user2': 200, 'user3': 150})

# Get top 3
    top = redis_client.zrevrange('leaderboard', 0, 2, withscores=True)

assert top[0][0] == 'user2'
    assert top[0][1] == 200
    assert len(top) == 3

---

Performance Testing de Queries

import time

def test_query_performance(postgres, sample_data):
    """Verifica que query executa em menos de 100ms"""
    start = time.time()

result = postgres.run("""
        SELECT u.name, COUNT(o.id) as order_count, SUM(o.total) as total
        FROM users u
        LEFT JOIN orders o ON u.id = o.user_id
        GROUP BY u.id
        ORDER BY total DESC
        LIMIT 100
    """)

elapsed = (time.time() - start) * 1000  # ms

assert elapsed < 100, f"Query took {elapsed}ms, expected < 100ms"

# Log para monitoramento
    print(f"Query executed in {elapsed}ms")

def test_index_usage(postgres):
    """Verifica que índice está sendo usado"""
    result = postgres.run("""
        EXPLAIN ANALYZE
        SELECT * FROM users WHERE email = 'test@example.com'
    """)

explain_output = result.fetchall()
    explain_text = ' '.join([row[0] for row in explain_output])

assert 'Index Scan' in explain_text or 'Bitmap Index Scan' in explain_text
    assert 'Seq Scan' not in explain_text  # Bad!

---

Conclusão

Testar banco de dados é crucial para garantir integridade, performance e confiabilidade dos dados. As chaves são:

1. Testar schema – Valide estrutura
2. Testar constraints – Garanta regras de negócio
3. Testar queries – Verifique funcionalidade
4. Testar migrations – Previna regressions
5. Testar performance – Queries devem ser rápidas

---

FAQ

P: Usar TestContainers em vez de DB real?
R: Sim! TestContainers oferece DB isolado, descartável e rápido para testes.

P: Como testar migrations?
R: Teste UP e DOWN, verifice integridade de dados após migration.

P: Mock vs DB real para testes?
R: DB real para queries complexas e joins. Mock para unit tests isolados.