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

<hr>
<h2>Introdução aos Testes de Banco de Dados</h2>

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

<h3>Por Que Testar Banco de Dados?</h3>

<ul><li>Garantir integridade de dados</li><ol><li>Validar constraints e triggers</li>
<li>Verificar performance de queries</li>
<li>Proteger contra regressions</li>
<li>Validar migrations</li></ol></ul>

<hr>
<h2>Tipos de Testes de Banco de Dados</h2>

<h3>1. Testes de Schema</h3>

<pre><code># 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'
    """)

<p>tables = [row[0] for row in result.fetchall()]</p>

<p>assert 'users' in tables<br>    assert 'products' in tables<br>    assert 'orders' in tables</p>

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

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

<p>assert 'id' in columns<br>    assert columns['id']['type'] == 'integer'<br>    assert 'email' in columns<br>    assert columns['email']['is_nullable'] == 'NO'</p>

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

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

<p>assert primary_keys['users'] == 'id'<br>    assert primary_keys['products'] == 'id'<br></code></pre></p>

<h3>2. Testes de Constraints</h3>

<pre><code>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')
    """)

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

<p>assert 'unique constraint' in str(exc.value).lower()</p>

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

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

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

<h3>3. Testes de Queries</h3>

<pre><code>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')
    """)

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

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

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

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

<p>row = result.fetchone()<br>    assert row[0] == 'João'<br>    assert row[2] == 100.00</p>

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

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

<p>row = result.fetchone()<br>    assert row[0] == 300.00<br>    assert row[1] == 2<br></code></pre></p>

<h3>4. Testes de Migrations</h3>

<pre><code>import subprocess
import os

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

<p>assert result.returncode == 0<br>    assert 'Migration complete' in result.stdout or 'No migrations to apply' in result.stdout</p>

<p>def test_migration_down(postgres):<br>    """Rollback migration"""<br>    migration_name = 'add_phone_to_users'</p>

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

<p>assert result.returncode == 0</p>

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

<p>assert len(result.fetchall()) == 0</p>

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

<p># Run migration<br>    subprocess.run(['python', 'manage.py', 'migrate'], check=True)</p>

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

<p>assert result.fetchone()[0] == 1<br></code></pre></p>

<hr>
<h2>TestContainers para Testes de DB</h2>

<pre><code># conftest.py
import pytest
from testcontainers.postgres import PostgresContainer

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

<p># Run migrations<br>        subprocess.run(['python', 'manage.py', 'migrate'], check=True)</p>

<p>yield pg</p>

<p># Teardown</p>

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

<p># Disable triggers para speed<br>    cursor.execute("SET session_replication_role = 'replica'")</p>

<p># Truncate tables (exceto immutable)<br>    cursor.execute("""<br>        TRUNCATE TABLE orders, users, products CASCADE<br>    """)</p>

<p>db.commit()<br>    cursor.close()<br>    db.close()</p>

<p>yield</p>

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

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

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

<p>db.commit()<br>    cursor.close()<br>    db.close()</p>

<p>yield {<br>        'user_ids': user_ids,<br>        'product_ids': product_ids<br>    }<br></code></pre></p>

<hr>
<h2>Testes de NoSQL</h2>

<h3>MongoDB</h3>

<pre><code>from pymongo import MongoClient
from datetime import datetime

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

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

<p>assert result.inserted_id is not None</p>

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

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

<p>pipeline = [<br>        {'$match': {'status': 'completed'}},<br>        {'$group': {'_id': '$user_id', 'total': {'$sum': '$total'}}}<br>    ]</p>

<p>results = list(mongodb.orders.aggregate(pipeline))</p>

<p>assert len(results) == 1<br>    assert results[0]['total'] == 300<br></code></pre></p>

<h3>Redis</h3>

<pre><code>import redis

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

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

<p># Get value<br>    name = redis_client.get('user:1:name')</p>

<p>assert name == 'João'</p>

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

<p># Get top 3<br>    top = redis_client.zrevrange('leaderboard', 0, 2, withscores=True)</p>

<p>assert top[0][0] == 'user2'<br>    assert top[0][1] == 200<br>    assert len(top) == 3<br></code></pre></p>

<hr>
<h2>Performance Testing de Queries</h2>

<pre><code>import time

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

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

<p>elapsed = (time.time() - start) * 1000  # ms</p>

<p>assert elapsed &lt; 100, f&quot;Query took {elapsed}ms, expected &lt; 100ms&quot;</p>

<p># Log para monitoramento<br>    print(f"Query executed in {elapsed}ms")</p>

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

<p>explain_output = result.fetchall()<br>    explain_text = ' '.join([row[0] for row in explain_output])</p>

<p>assert 'Index Scan' in explain_text or 'Bitmap Index Scan' in explain_text<br>    assert 'Seq Scan' not in explain_text  # Bad!<br></code></pre></p>

<hr>
<h2>Conclusão</h2>

<p>Testar banco de dados é crucial para garantir integridade, performance e confiabilidade dos dados. As chaves são:<br><ol><li><strong>Testar schema</strong> - Valide estrutura</li><br><li><strong>Testar constraints</strong> - Garanta regras de negócio</li><br><li><strong>Testar queries</strong> - Verifique funcionalidade</li><br><li><strong>Testar migrations</strong> - Previna regressions</li><br><li><strong>Testar performance</strong> - Queries devem ser rápidas</li><br></ol><br><hr><br><h3>FAQ</h3></p>

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

<p><strong>P: Como testar migrations?</strong>  <br>R: Teste UP e DOWN, verifice integridade de dados após migration.</p>

<p><strong>P: Mock vs DB real para testes?</strong>  <br>R: DB real para queries complexas e joins. Mock para unit tests isolados.</p>