Кто подготавливал тест-кейсы?

Кто подготавливает тест-кейсы в Data Engineering

Подготовка тест-кейсов — важный процесс, в котором участвуют разные роли. Расскажу о том, кто и как это делает.

QA / Data Quality Engineer

Основные люди, занимающиеся тестированием данных:

Обязанности:

• Проверка корректности данных на входе и выходе ETL
• Валидация бизнес-правил
• Поиск багов в данных (дубликаты, null значения, выбросы)
• Регрессионное тестирование после изменений

Типы тест-кейсов:

# Функциональные тесты
def test_etl_daily_extract():
    """Проверяем, что каждый день извлекаются данные из источника"""
    result = etl.extract_daily_data(date='2026-03-26')
    assert result['row_count'] > 0, "No data extracted"
    assert result['errors'] == 0, "Errors occurred during extraction"

def test_data_quality():
    """Проверяем качество данных в витрине"""
    data = query_warehouse("SELECT * FROM fact_sales")
    assert data['user_id'].isna().sum() == 0, "NULL values in user_id"
    assert len(data) == len(data.drop_duplicates()), "Duplicates found"

# Граничные случаи
def test_null_handling():
    """Проверяем обработку NULL значений"""
    data = process_data({"amount": None, "quantity": 0})
    assert data['amount'] == 0, "NULL amount should be 0"
    assert data['quantity'] == 0, "Zero quantity handling"

# Производительность
def test_etl_performance():
    """Проверяем, что ETL выполняется за приемлемое время"""
    start = time.time()
    etl.run_full_load()
    elapsed = time.time() - start
    assert elapsed < 900, f"ETL took {elapsed}s, max 15 minutes"

Data Engineer (сам)

Это не только QA-шник, но и сам инженер отвечает за:

Unit тесты — тестирование отдельных функций обработки данных:

def clean_amount(value: float) -> float:
    """Очищает и нормализует сумму"""
    if value < 0:
        return 0
    return round(value, 2)

def test_clean_amount():
    assert clean_amount(10.556) == 10.56
    assert clean_amount(-5) == 0
    assert clean_amount(0) == 0

Интеграционные тесты — проверка взаимодействия компонентов:

def test_end_to_end_etl():
    # Загрузим тестовые данные
    test_data = [
        {'user_id': 1, 'amount': 100},
        {'user_id': 2, 'amount': 200}
    ]
    
    # Запустим ETL
    etl.extract(test_data)
    etl.transform()
    etl.load()
    
    # Проверим результат
    result = query_warehouse("SELECT SUM(amount) as total FROM fact_sales")
    assert result['total'] == 300

Бизнес-аналитик / Product Owner

Они готовят бизнес-требования для тестирования:

• "Витрина должна показывать выручку по странам"
• "Метрика DAU должна совпадать с Google Analytics"
• "Историческая выручка не должна меняться ретроспективно"

Аналитик (внутренний клиент)

Это "end user" данных. Они:

• Проверяют, что числа корректны ("В этом месяце выручка выросла на 20%, это правда?")
• Сравнивают с другими источниками (Google Sheets, Tableau)
• Заполняют баг-репорты: "Этот метрик вчера показывал 500K, сегодня 400K, что случилось?"

Процесс подготовки тест-кейсов

1. Requirements gathering

Data Engineer собирает требования:

• Какие данные нужны?
• Какие преобразования?
• Какие бизнес-правила?
• SLA (когда должны быть готовы)?

2. Написание тестов (TDD подход)

Сначала пишут падающие тесты, потом код:

# Шаг 1: RED — тест падает
def test_customer_lifetime_value():
    result = calculate_clv(customer_id=123)
    assert result == 5000  # Ожидаем 5000

# Шаг 2: GREEN — пишем код
def calculate_clv(customer_id):
    return db.query(f"SELECT SUM(amount) FROM sales WHERE user_id={customer_id}")

# Шаг 3: REFACTOR — улучшаем
def calculate_clv(customer_id: int) -> Decimal:
    query = """
        SELECT SUM(amount) FROM sales 
        WHERE user_id = @user_id
    """
    return db.execute(query, user_id=customer_id)

3. Data validation кейсы

-- Проверяем, что дата актуальна
SELECT MAX(date) as latest_date FROM fact_sales
WHERE DATE(date) = CURRENT_DATE();

-- Проверяем целостность ссылок
SELECT COUNT(*) FROM fact_sales fs
LEFT JOIN dim_user u ON fs.user_id = u.user_id
WHERE u.user_id IS NULL;  -- Должно быть 0

-- Сравниваем данные с источником
SELECT 
    source_count,
    warehouse_count,
    source_count - warehouse_count as difference
FROM (
    SELECT COUNT(*) as source_count FROM source_db.sales
) s,
(
    SELECT COUNT(*) as warehouse_count FROM fact_sales
) w
WHERE source_count != warehouse_count;

4. Automation

Тесты запускаются автоматически:

# CI/CD pipeline (например, GitHub Actions)
- name: Run data quality tests
  run: pytest tests/data_quality/ -v --cov
  
- name: Check schema changes
  run: pytest tests/schema/ -v
  
- name: Performance tests
  run: pytest tests/performance/ -v --timeout=300

Инструменты для тестирования

• Great Expectations — open-source framework для data validation
• dbt tests — если используется dbt, тесты встроены
• pytest — стандартный фреймворк для Python
• SQL Server Data Tools (SSDT) — для SQL сервера
• Custom скрипты — специфичные для бизнеса проверки

Итог

Тест-кейсы готовит команда:

1. QA/Data Quality — автоматизированные проверки
2. Data Engineer — unit и интеграционные тесты
3. Product Owner — бизнес-требования
4. Аналитик — валидация в production

Каждый уровень важен для обеспечения качества данных.