Хранимые процедуры и триггеры

Updated Jul 2026

Бизнес-логика, живущая в базе данных — хранимые процедуры, функции и триггеры — должна тестироваться как любой другой код. Они особенно распространены в финансовых системах, управлении запасами и везде, где целостность данных достаточно критична для обеспечения на уровне базы данных.

Почему бизнес-логика в базе данных?

Компромисс: логика в базе данных сложнее для контроля версий, сложнее для тестирования и сложнее для отладки, чем код приложения. Именно поэтому её тестирование так важно.

Тестирование хранимых процедур

Базовый тест функции

def test_calculate_order_total(db):
    cursor = db.cursor()

    # Set up test data
    cursor.execute("INSERT INTO orders (id, status) VALUES ('test-order', 'pending')")
    cursor.execute("""
        INSERT INTO line_items (order_id, quantity, unit_price) VALUES
        ('test-order', 2, 10.00),
        ('test-order', 1, 25.50)
    """)

    # Call the stored procedure
    cursor.execute("SELECT calculate_order_total('test-order')")
    total = cursor.fetchone()[0]

    # Verify calculation
    expected = (2 * 10.00) + (1 * 25.50)  # 45.50
    assert total == expected, f"Expected {expected}, got {total}"

Тестирование граничных случаев

def test_calculate_order_total_empty_order(db):
    """Order with no line items should return 0."""
    cursor = db.cursor()
    cursor.execute("INSERT INTO orders (id, status) VALUES ('empty-order', 'pending')")
    cursor.execute("SELECT calculate_order_total('empty-order')")
    assert cursor.fetchone()[0] == 0

def test_calculate_order_total_nonexistent(db):
    """Non-existent order should raise an error or return NULL."""
    cursor = db.cursor()
    cursor.execute("SELECT calculate_order_total('nonexistent')")
    result = cursor.fetchone()[0]
    assert result is None or result == 0

def test_calculate_order_total_large_quantities(db):
    """Test with large quantities to check for overflow."""
    cursor = db.cursor()
    cursor.execute("INSERT INTO orders (id, status) VALUES ('big-order', 'pending')")
    cursor.execute("""
        INSERT INTO line_items (order_id, quantity, unit_price) VALUES
        ('big-order', 999999, 9999.99)
    """)
    cursor.execute("SELECT calculate_order_total('big-order')")
    total = cursor.fetchone()[0]
    expected = 999999 * 9999.99
    assert abs(total - expected) < 0.01  # Allow small floating-point difference

def test_calculate_order_total_precision(db):
    """Test decimal precision (common bug with money calculations)."""
    cursor = db.cursor()
    cursor.execute("INSERT INTO orders (id, status) VALUES ('precise-order', 'pending')")
    cursor.execute("""
        INSERT INTO line_items (order_id, quantity, unit_price) VALUES
        ('precise-order', 3, 0.10)
    """)
    cursor.execute("SELECT calculate_order_total('precise-order')")
    total = cursor.fetchone()[0]
    # 3 * 0.10 should be exactly 0.30, not 0.30000000000000004
    assert total == 0.30, f"Precision error: expected 0.30, got {total}"

Тестирование триггеров

Триггеры выполняются автоматически при вставке, обновлении или удалении данных. Они невидимы для приложения — что делает их особенно важными для тестирования.

Триггер updated_at

def test_updated_at_trigger(db):
    """Verify that updated_at changes when a row is modified."""
    cursor = db.cursor()

    # Create a user
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('trigger-test', 'Original Name', 'trigger@test.com')
        RETURNING created_at, updated_at
    """)
    created_at, updated_at = cursor.fetchone()
    assert created_at == updated_at  # On creation, both should be the same

    # Wait briefly and update
    import time
    time.sleep(1)
    cursor.execute("""
        UPDATE users SET name = 'Updated Name'
        WHERE id = 'trigger-test'
    """)

    # Verify updated_at changed
    cursor.execute("SELECT updated_at FROM users WHERE id = 'trigger-test'")
    new_updated_at = cursor.fetchone()[0]
    assert new_updated_at > updated_at, "updated_at should be newer after UPDATE"

def test_updated_at_unchanged_on_same_value(db):
    """Some implementations only update timestamp if data actually changed."""
    cursor = db.cursor()
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('same-val-test', 'Same Name', 'sameval@test.com')
    """)

    import time
    time.sleep(1)

    # Update with the same value
    cursor.execute("""
        UPDATE users SET name = 'Same Name'
        WHERE id = 'same-val-test'
    """)

    cursor.execute("""
        SELECT created_at, updated_at FROM users WHERE id = 'same-val-test'
    """)
    created_at, updated_at = cursor.fetchone()
    # Behavior depends on trigger implementation:
    # Some triggers update timestamp regardless, others only on actual change

Триггер аудиторского следа

def test_audit_trail_on_update(db):
    """Verify that changes to users are logged in audit table."""
    cursor = db.cursor()

    # Create user
    cursor.execute("""
        INSERT INTO users (id, name, email, role)
        VALUES ('audit-test', 'Audit User', 'audit@test.com', 'viewer')
    """)

    # Update role
    cursor.execute("""
        UPDATE users SET role = 'admin' WHERE id = 'audit-test'
    """)

    # Check audit table
    cursor.execute("""
        SELECT old_value, new_value, changed_field, changed_at
        FROM audit_log
        WHERE table_name = 'users' AND record_id = 'audit-test'
        ORDER BY changed_at DESC LIMIT 1
    """)
    row = cursor.fetchone()
    assert row is not None, "Audit log entry should exist"
    assert row[0] == "viewer"   # old_value
    assert row[1] == "admin"    # new_value
    assert row[2] == "role"     # changed_field

def test_audit_trail_on_delete(db):
    """Verify that deletions are logged."""
    cursor = db.cursor()
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('delete-audit', 'Delete Me', 'delete@test.com')
    """)
    cursor.execute("DELETE FROM users WHERE id = 'delete-audit'")

    cursor.execute("""
        SELECT action FROM audit_log
        WHERE table_name = 'users' AND record_id = 'delete-audit'
    """)
    row = cursor.fetchone()
    assert row is not None
    assert row[0] == "DELETE"

Тестирование вычисляемых/производных значений

Некоторые системы вычисляют значения в базе данных (материализованные представления, генерируемые столбцы):

def test_user_order_count_computed(db):
    """Verify computed order_count matches actual orders."""
    cursor = db.cursor()

    # Create user with orders
    cursor.execute("INSERT INTO users (id, name, email) VALUES ('count-test', 'Count User', 'count@test.com')")
    cursor.execute("""
        INSERT INTO orders (user_id, status, total) VALUES
        ('count-test', 'completed', 50.00),
        ('count-test', 'completed', 75.00),
        ('count-test', 'cancelled', 25.00)
    """)

    # If there is a materialized view or computed column:
    cursor.execute("SELECT order_count FROM user_stats WHERE user_id = 'count-test'")
    computed_count = cursor.fetchone()[0]

    # Verify against actual count
    cursor.execute("SELECT COUNT(*) FROM orders WHERE user_id = 'count-test'")
    actual_count = cursor.fetchone()[0]

    assert computed_count == actual_count

Тестирование обработки ошибок в процедурах

def test_transfer_funds_insufficient_balance(db):
    """Transfer should fail if source account has insufficient funds."""
    cursor = db.cursor()

    # Set up accounts
    cursor.execute("""
        INSERT INTO accounts (id, balance) VALUES
        ('acct-a', 100.00),
        ('acct-b', 50.00)
    """)

    # Attempt transfer exceeding balance
    with pytest.raises(Exception):  # Specific exception depends on implementation
        cursor.execute("SELECT transfer_funds('acct-a', 'acct-b', 200.00)")

    # Verify no money was moved (transaction should be rolled back internally)
    cursor.execute("SELECT balance FROM accounts WHERE id = 'acct-a'")
    assert cursor.fetchone()[0] == 100.00  # Unchanged
    cursor.execute("SELECT balance FROM accounts WHERE id = 'acct-b'")
    assert cursor.fetchone()[0] == 50.00   # Unchanged

Практическое упражнение

1. Напишите тесты для функции calculate_order_total: обычный случай, пустой заказ, большие числа, точность
2. Напишите тесты для триггера updated_at: проверьте, что он меняется при обновлении
3. Напишите тесты для триггера аудиторского следа: проверьте, что вставки, обновления и удаления логируются
4. Напишите тесты для процедуры перевода средств: успешный случай, недостаточный баланс, один и тот же счёт
5. Протестируйте триггер, предотвращающий удаление пользователей-администраторов

Ключевые выводы

• Бизнес-логика в базе данных (процедуры, триггеры) должна тестироваться как код приложения
• Тестируйте граничные случаи: пустые входные данные, большие числа, точность, параллельный доступ
• Триггеры невидимы для приложения — тестирование единственный способ проверить их работу
• Тестируйте как успешные, так и ошибочные пути хранимых процедур
• Триггеры аудиторского следа требуют специфической проверки: старое значение, новое значение, изменённое поле, временная метка