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

**Meta Description:** Aprenda a testar bancos de dados: consultas SQL, integridade de dados, migrations, performance e testes de integração com banco de dados.

---

## 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
- Validar constraints e triggers
- Verificar performance de queries
- Proteger contra regressions
- Validar migrations

---

## Tipos de Testes de Banco de Dados

### 1. Testes de Schema

```python
# 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

```python
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

```python
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

```python
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

```python
# 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

```python
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

```python
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

```python
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.