Тестирование пайплайна: unit tests для SQL-трансформаций

Условие задания

**Компания:** ВК

**Контекст:** ETL-пайплайн содержит сложные SQL-трансформации. Изменения в логике периодически ломают витрины. Нужно покрыть трансформации тестами.

**Задание:**
1. Напишите Python-фреймворк для тестирования SQL-трансформаций с использованием SQLite (in-memory)
2. Реализуйте тесты для 3 трансформаций:
- Дедупликация по ROW_NUMBER
- Расчёт retention
- Воронка конверсии
3. Покажите, как интегрировать тесты в CI/CD

Пример данных

Структура для ориентира — реальные значения из эталонного решения.

import sqlite3
import pytest
from pathlib import Path

class SQLTestRunner:
    """Фреймворк для тестирования SQL-трансформаций."""

    def __init__(self):
        self.conn = sqlite3.connect(':memory:')
        self.conn.row_factory = sqlite3.Row

    def setup_table(self, ddl: str, data: list[dict]):
        """Создать таблицу и загрузить тестовые данные."""
        self.conn.execute(ddl)
        if data:
            cols = data[0].keys()
            placeholders = ', '.join(['?'] * len(cols))
            col_names = ', '.join(cols)
            self.conn.executemany(
                f'INSERT INTO ({col_names}) VALUES ({placeholders})',
                [tuple(row.values()) for row in data]
            )

    def execute(self, sql: str) -> list[dict]:
        """Выполнить SQL и вернуть результат."""
        rows = self.conn.execute(sql).fetchall()
        return [dict(row) for row in rows]

    def teardown(self):
        self.conn.close()


@pytest.fixture
def sql_runner():
    runner = SQLTestRunner()
    yield runner
    runner.teardown()

Темы

тестирование unit tests pytest SQL mock

Подсказки

• SQLite in-memory (:memory:) для быстрых тестов без внешних зависимостей
• Создайте fixture-данные, выполните SQL, проверьте результат
• pytest + conftest.py для переиспользуемых fixture

Все тестовые задания →

Частые вопросы

Какой уровень знаний нужен для задачи "Тестирование пайплайна: unit tests для SQL-трансформаций"?

Это задание для уровня hard. Senior-уровень — глубокое понимание темы, опыт решения нестандартных задач, обсуждение trade-off на собеседовании.

На каких собеседованиях встречается такая задача?

Подобные задания в категории «data_engineering» регулярно дают на собеседованиях аналитика данных в Яндекс, Сбер, Ozon, Авито, Тинькофф, Wildberries, T-Bank, X5, ВТБ и других крупных IT-компаниях. Тематика: тестирование, unit tests, pytest, SQL, mock.

Сколько времени даётся на решение?

На реальном собеседовании на подобную задачу отводится 30-60 минут с обсуждением подходов, оптимизаций и trade-off. Для тренировки рекомендуем сначала решить самостоятельно, потом сверить с эталонным решением и подсказками.

Где ещё потренироваться по теме «data_engineering»?

На zasqlpython.ru есть другие задания в категории «data_engineering», продуктовые кейсы, справочник метрик, AI мок-собеседование с разбором ваших ответов.

← Все задания