LocalStack и Testcontainers: локальная S3-эмуляция - ETL-тесты без реального кластера

# Кажется умным решением - замокировать boto3 и SparkContext
from unittest.mock import patch, MagicMock

@patch("boto3.client")
@patch("pyspark.sql.SparkSession.read")
def test_pipeline(mock_read, mock_boto):
    mock_read.return_value.parquet.return_value = mock_df
    mock_boto.return_value.list_objects_v2.return_value = {...}

    result = run_pipeline("s3a://bucket/input", "s3a://bucket/output")
    assert result.count() == 100

from testcontainers.localstack import LocalStackContainer

# Testcontainers сам:
# 1. Скачивает образ localstack/localstack:latest (если нет)
# 2. Запускает контейнер
# 3. Маппирует случайный свободный порт хоста → 4566 контейнера
# 4. Ждёт, пока LocalStack полностью инициализируется (healthcheck)
# 5. Возвращает управление тесту
with LocalStackContainer(image="localstack/localstack:3.4") as localstack:
    endpoint = localstack.get_url()
    # endpoint = "http://localhost:54321" (случайный порт!)

    # ... тест ...

# После выхода из контекста контейнер автоматически останавливается и удаляется

# Установка через pip
pip install \
    pyspark==3.5.1 \
    boto3==1.34.0 \
    testcontainers[localstack]==4.7.0 \
    pytest==8.1.0 \
    pytest-timeout==2.3.0

# Или через requirements-test.txt (рекомендуется)

# requirements-test.txt
pyspark==3.5.1
boto3>=1.34.0
testcontainers[localstack]>=4.7.0
pytest>=8.1.0
pytest-timeout>=2.3.0

spark = SparkSession.builder \
    .config("spark.jars.packages",
            "org.apache.hadoop:hadoop-aws:3.3.4,"
            "com.amazonaws:aws-java-sdk-bundle:1.12.367") \
    .getOrCreate()

# Dockerfile или CI конфигурация
ENV SPARK_HOME=/opt/spark
ENV HADOOP_CONF_DIR=/etc/hadoop/conf

# Копируем JARы в директорию Spark
COPY hadoop-aws-3.3.4.jar $SPARK_HOME/jars/
COPY aws-java-sdk-bundle-1.12.367.jar $SPARK_HOME/jars/

import os
os.environ["PYSPARK_SUBMIT_ARGS"] = (
    "--packages org.apache.hadoop:hadoop-aws:3.3.4,"
    "com.amazonaws:aws-java-sdk-bundle:1.12.367 "
    "pyspark-shell"
)

https://{bucket-name}.s3.amazonaws.com/{object-key}

https://s3.amazonaws.com/{bucket-name}/{object-key}

http://my-bucket.localhost:4566/data/file.parquet

http://localhost:4566/my-bucket/data/file.parquet

from pyspark.sql import SparkSession
from testcontainers.localstack import LocalStackContainer


def create_spark_for_localstack(endpoint_url: str) -> SparkSession:
    """
    Создаёт SparkSession с конфигурацией S3A Connector для LocalStack.

    endpoint_url: URL LocalStack контейнера, например "http://localhost:54321"
    Порт динамически выбирается Testcontainers - не хардкодьте его.
    """
    return (
        SparkSession.builder
        .master("local[2]")
        .appName("integration-tests")

        # ── Hadoop AWS JARы ──────────────────────────────────────────
        # Указываем пакеты, которые Spark скачает из Maven.
        # hadoop-aws: реализует FileSystem API для протокола s3a://
        # aws-java-sdk-bundle: клиент AWS API (используется hadoop-aws внутри)
        .config(
            "spark.jars.packages",
            "org.apache.hadoop:hadoop-aws:3.3.4,"
            "com.amazonaws:aws-java-sdk-bundle:1.12.367"
        )

        # ── S3A Endpoint ─────────────────────────────────────────────
        # Перенаправляем все s3a:// запросы на LocalStack вместо реального AWS.
        # Без этого параметра Spark будет пытаться соединиться с
        # https://s3.amazonaws.com и упадёт с NoSuchBucketException или
        # AWSException: Unable to execute HTTP request.
        .config("spark.hadoop.fs.s3a.endpoint", endpoint_url)

        # ── Path Style Access ────────────────────────────────────────
        # true: использовать http://host/bucket/key (работает с LocalStack)
        # false: использовать http://bucket.host/key (Virtual-Hosted, только для AWS)
        # Подробно разобрано в предыдущем разделе.
        .config("spark.hadoop.fs.s3a.path.style.access", "true")

        # ── Credentials Provider ─────────────────────────────────────
        # LocalStack принимает любые credentials, но Spark должен их предоставить.
        # SimpleAWSCredentialsProvider: берёт ключи из конфигурации напрямую
        # (в отличие от InstanceProfileCredentialsProvider, который ищет IMDSv2).
        # Без этого Spark может попытаться обратиться к EC2 Instance Metadata
        # Service и зависнуть на 30 секунд ожидания ответа от несуществующего URL.
        .config(
            "spark.hadoop.fs.s3a.aws.credentials.provider",
            "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
        )

        # ── Фиктивные credentials ────────────────────────────────────
        # LocalStack игнорирует конкретные значения, но они должны быть
        # непустыми - S3A Connector валидирует их формат перед отправкой.
        # Конвенция: используйте "test"/"test" - это стандарт для LocalStack.
        .config("spark.hadoop.fs.s3a.access.key", "test")
        .config("spark.hadoop.fs.s3a.secret.key", "test")

        # ── SSL ──────────────────────────────────────────────────────
        # LocalStack работает по HTTP (не HTTPS). Отключаем SSL проверку,
        # чтобы S3A Connector не пытался валидировать TLS сертификат.
        .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")

        # ── S3A Committer ────────────────────────────────────────────
        # Для тестов используем простой FileOutputCommitter v1.
        # В продакшне с MinIO или S3 нужен magic committer или staging committer
        # (разобрано в уроке 3 этого модуля), но для локальных тестов
        # v1 достаточен и не требует дополнительной настройки.
        .config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")

        # ── Delta Lake (опционально) ─────────────────────────────────
        # Если тестируете Delta Lake пайплайны, добавьте эти параметры.
        # Они позволяют Delta использовать S3A для хранения логов транзакций.
        # .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        # .config("spark.sql.catalog.spark_catalog",
        #         "org.apache.spark.sql.delta.catalog.DeltaCatalog")

        # ── Производительность ───────────────────────────────────────
        # Для тестов уменьшаем overhead от параллелизма.
        # В продакшне эти параметры не трогайте.
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.default.parallelism", "4")

        .getOrCreate()
    )

# tests/conftest.py

import pytest
import boto3
from botocore.config import Config
from testcontainers.localstack import LocalStackContainer
from pyspark.sql import SparkSession


# ─── Инфраструктурные фикстуры ────────────────────────────────────────────────

@pytest.fixture(scope="session")
def localstack_container():
    """
    Запускает LocalStack контейнер один раз на всю тестовую сессию.

    scope="session" означает: один контейнер для всех тестов.
    Это экономит ~10 секунд на каждый тест - время старта LocalStack.

    with LocalStackContainer(...) as container:
        Testcontainers автоматически:
        1. Запускает docker pull localstack/localstack:3.4 (если нет образа)
        2. Создаёт контейнер с случайным портом вместо 4566
        3. Ждёт, пока healthcheck вернёт 200 OK
        4. Yield - передаёт контейнер в тесты
        5. После всех тестов: docker stop + docker rm

    image="localstack/localstack:3.4" - pinned версия. Никогда не используйте
    "latest" в тестах - в следующем релизе LocalStack может сломать совместимость.
    """
    with LocalStackContainer(image="localstack/localstack:3.4") as container:
        yield container


@pytest.fixture(scope="session")
def localstack_endpoint(localstack_container):
    """
    Возвращает URL endpoint LocalStack.

    LocalStackContainer.get_url() возвращает "http://localhost:{random_port}".
    Порт выбирается Testcontainers автоматически из свободных портов хоста.
    Это позволяет запускать несколько тестовых сессий одновременно —
    у каждой будет свой порт.
    """
    url = localstack_container.get_url()
    print(f"\nLocalStack endpoint: {url}")
    return url


@pytest.fixture(scope="session")
def s3_client(localstack_endpoint):
    """
    Создаёт boto3 S3 client, подключённый к LocalStack.

    Этот клиент используется для:
    1. Создания тестовых бакетов
    2. Загрузки входных тестовых данных
    3. Проверки результатов ETL (чтение метаданных, листинг файлов)

    Для boto3 также нужны специальные параметры для работы с LocalStack.
    endpoint_url: направляем запросы в LocalStack, а не в AWS.
    aws_access_key_id / secret_access_key: любые непустые значения.
    region_name: LocalStack принимает любой регион. "us-east-1" - конвенция.

    Config(signature_version="s3v4"): принудительно использовать Signature V4
    (SigV4) для подписи запросов. LocalStack поддерживает SigV4.
    Без этого параметра boto3 иногда использует Signature V2 (устаревший),
    что может вызвать ошибки аутентификации.
    """
    client = boto3.client(
        "s3",
        endpoint_url=localstack_endpoint,
        aws_access_key_id="test",
        aws_secret_access_key="test",
        region_name="us-east-1",
        config=Config(signature_version="s3v4")
    )
    return client


@pytest.fixture(scope="session")
def spark(localstack_endpoint):
    """
    Создаёт SparkSession с конфигурацией S3A для LocalStack.

    scope="session": SparkSession создаётся один раз на всю тестовую сессию.
    Создание SparkSession - дорогая операция (запуск JVM, загрузка JARов).
    При scope="function" каждый тест тратит 15-20 секунд только на старт Spark.

    Важный момент: SparkSession - синглтон в рамках JVM.
    Если сессия уже создана с одним набором конфигов, второй вызов
    SparkSession.builder.getOrCreate() вернёт ту же сессию с теми же конфигами.
    Поэтому session-scope фикстура - правильное решение.
    """
    spark_session = (
        SparkSession.builder
        .master("local[2]")
        .appName("pyspark-integration-tests")
        .config(
            "spark.jars.packages",
            "org.apache.hadoop:hadoop-aws:3.3.4,"
            "com.amazonaws:aws-java-sdk-bundle:1.12.367"
        )
        .config("spark.hadoop.fs.s3a.endpoint", localstack_endpoint)
        .config("spark.hadoop.fs.s3a.path.style.access", "true")
        .config(
            "spark.hadoop.fs.s3a.aws.credentials.provider",
            "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
        )
        .config("spark.hadoop.fs.s3a.access.key", "test")
        .config("spark.hadoop.fs.s3a.secret.key", "test")
        .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.default.parallelism", "4")
        # Отключаем попытки обновить spark в тестах
        .config("spark.ui.enabled", "false")
        .getOrCreate()
    )

    # Выставляем уровень логирования: WARN скрывает INFO-сообщения Spark,
    # которые загромождают вывод тестов. ERROR будет ещё тише.
    spark_session.sparkContext.setLogLevel("WARN")

    yield spark_session

    # После завершения всех тестов - останавливаем SparkSession.
    # Без этого JVM процесс может остаться активным после pytest.
    spark_session.stop()


# ─── Фикстуры данных ─────────────────────────────────────────────────────────

@pytest.fixture(scope="module")
def test_buckets(s3_client):
    """
    Создаёт тестовые бакеты перед началом тестов в модуле.

    scope="module": бакеты создаются один раз для всех тестов в одном .py файле.
    Бакеты не удаляются между тестами - только тестовые данные внутри них.

    Почему LocalStack не требует LocationConstraint для us-east-1?
    В реальном AWS при создании бакета в us-east-1 нельзя передавать
    CreateBucketConfiguration - это вызовет ошибку. Но для us-east-1 это
    правило особое, и LocalStack его воспроизводит корректно.
    """
    bucket_names = ["bronze", "silver", "gold"]

    for bucket in bucket_names:
        try:
            # LocalStack не жалуется на повторное создание - он идемпотентен.
            # В реальном S3 повторное создание своего бакета в том же регионе
            # вернёт BucketAlreadyOwnedByYou (не ошибка) или успех.
            s3_client.create_bucket(Bucket=bucket)
            print(f"Создан бакет: {bucket}")
        except s3_client.exceptions.BucketAlreadyExists:
            print(f"Бакет уже существует: {bucket}")

    yield bucket_names

    # Teardown: очищаем содержимое бакетов (не сами бакеты)
    # Это позволяет следующим тестовым модулям создавать бакеты заново.
    for bucket in bucket_names:
        try:
            paginator = s3_client.get_paginator("list_objects_v2")
            for page in paginator.paginate(Bucket=bucket):
                objects = page.get("Contents", [])
                if objects:
                    s3_client.delete_objects(
                        Bucket=bucket,
                        Delete={"Objects": [{"Key": o["Key"]} for o in objects]}
                    )
        except Exception as e:
            print(f"Ошибка при очистке {bucket}: {e}")

# src/pipeline/bronze_ingestion.py

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType,
    LongType, TimestampType, IntegerType, DoubleType
)
import json


# Схема входящих событий от источника (JSON файлы в bronze)
EVENT_SCHEMA = StructType([
    StructField("event_id",   StringType(),    nullable=False),
    StructField("user_id",    LongType(),      nullable=False),
    StructField("event_type", StringType(),    nullable=True),
    StructField("timestamp",  StringType(),    nullable=True),
    StructField("session_id", StringType(),    nullable=True),
    StructField("amount",     DoubleType(),    nullable=True),
    StructField("currency",   StringType(),    nullable=True),
    StructField("country",    StringType(),    nullable=True),
])


def ingest_json_to_bronze(
    spark: SparkSession,
    source_path: str,
    bronze_path: str,
    date_partition: str
) -> int:
    """
    Bronze ingestion: читает сырые JSON файлы и записывает в Parquet.

    Это первая стадия Medallion Architecture. Здесь мы:
    1. Читаем JSON с явной схемой (не inferSchema - он медленнее и ненадёжен
       для больших файлов, так как требует второго прохода по данным)
    2. Добавляем технические поля (ingestion timestamp, partition)
    3. Записываем в Parquet с партиционированием по дате
    4. Возвращаем количество записанных строк для метрик

    source_path: s3a://bronze/incoming/2024/01/15/
    bronze_path: s3a://bronze/events/
    date_partition: "2024-01-15" (используется для партиционирования)

    Возвращает: количество записанных строк.
    """
    # Читаем JSON с явной схемой. multiLine=False: каждая строка - один JSON-объект
    # (JSON Lines формат, также известный как NDJSON или jsonl)
    raw_df = spark.read \
        .option("multiLine", "false") \
        .schema(EVENT_SCHEMA) \
        .json(source_path)

    # Добавляем технические поля.
    # ingested_at: timestamp момента ingestion - полезен для отладки и SLA-мониторинга.
    # date_partition: дата партиционирования - используется в partitionBy ниже.
    # _source_path: откуда пришли данные - помогает в troubleshooting.
    enriched_df = raw_df.withColumns({
        "ingested_at":     F.current_timestamp(),
        "date_partition":  F.lit(date_partition),
        "_source_path":    F.lit(source_path),
    })

    # Считаем строки ДО записи - после write() df становится неактивным планом.
    row_count = enriched_df.count()

    # Записываем в Parquet, партиционированный по date_partition.
    # mode("append"): добавляем к существующим данным.
    # partitionBy: создаёт директорию date_partition=2024-01-15/ внутри bronze_path.
    # Это позволяет Spark читать только нужную дату через predicate pushdown.
    enriched_df.write \
        .mode("append") \
        .partitionBy("date_partition") \
        .parquet(bronze_path)

    return row_count


def deduplicate_to_silver(
    spark: SparkSession,
    bronze_path: str,
    silver_path: str,
    date_partition: str
) -> int:
    """
    Silver transformation: дедупликация и очистка данных.

    Это вторая стадия Medallion Architecture. Здесь мы:
    1. Читаем только нужную дату из Bronze (partition pruning)
    2. Удаляем дубликаты по event_id (идемпотентность пайплайна)
    3. Фильтруем записи с null в обязательных полях
    4. Нормализуем event_type (lowercase, trim)
    5. Записываем в Silver

    Дедупликация по event_id критически важна: источники данных часто
    отправляют повторные события (retry при сбоях сети, at-least-once delivery).
    В Silver каждый event_id должен встречаться ровно один раз.
    """
    # Читаем только нужную партицию - Spark использует partition pruning.
    # Физически читаются только файлы в date_partition=2024-01-15/.
    # Это в разы быстрее чем читать весь bronze_path с filter().
    bronze_df = spark.read \
        .parquet(bronze_path) \
        .filter(F.col("date_partition") == date_partition)

    # Дедупликация: оставляем одну запись для каждого event_id.
    # dropDuplicates(["event_id"]): недетерминированный выбор среди дублей.
    # Если нужен детерминированный (например, последний по timestamp),
    # используйте Window с row_number() - медленнее, но предсказуемо.
    deduped_df = bronze_df.dropDuplicates(["event_id"])

    # Фильтрация: отбрасываем записи без обязательных полей.
    # event_id и user_id - первичные ключи, без них запись бессмысленна.
    # event_type - нужен для бизнес-логики downstream агрегаций.
    clean_df = deduped_df.filter(
        F.col("event_id").isNotNull()
        & F.col("user_id").isNotNull()
        & F.col("event_type").isNotNull()
    )

    # Нормализация: стандартизируем event_type.
    # lower() + trim(): "  Purchase  " → "purchase"
    # Без нормализации одно событие может иметь разные написания:
    # "Purchase", "PURCHASE", " purchase" - это разные значения для GROUP BY.
    normalized_df = clean_df.withColumn(
        "event_type",
        F.lower(F.trim(F.col("event_type")))
    )

    row_count = normalized_df.count()

    normalized_df.write \
        .mode("overwrite") \
        .partitionBy("date_partition") \
        .parquet(silver_path)

    return row_count


def aggregate_to_gold(
    spark: SparkSession,
    silver_path: str,
    gold_path: str,
    date_partition: str
) -> DataFrame:
    """
    Gold aggregation: бизнес-метрики по дням и странам.

    Это третья стадия Medallion Architecture. Здесь мы считаем
    бизнес-агрегаты, которые напрямую используются BI-инструментами.

    Агрегаты:
    - Количество уникальных пользователей по стране и типу события
    - Суммарная сумма транзакций (для event_type="purchase")
    - Среднее время сессии (если доступно)

    В отличие от Bronze и Silver, Gold обычно не партиционируется —
    таблицы небольшие и читаются целиком аналитическими инструментами.
    """
    silver_df = spark.read \
        .parquet(silver_path) \
        .filter(F.col("date_partition") == date_partition)

    # Агрегация по стране и типу события.
    # countDistinct("user_id"): уникальные пользователи (не просто count строк).
    # sum("amount"): суммарный объём транзакций. coalesce(0): null → 0.
    # Важно: sum(null) = null в Spark SQL, поэтому нужен coalesce.
    gold_df = silver_df.groupBy("country", "event_type", "date_partition") \
        .agg(
            F.count("*").alias("event_count"),
            F.countDistinct("user_id").alias("unique_users"),
            F.coalesce(F.sum("amount"), F.lit(0.0)).alias("total_amount"),
            F.countDistinct("session_id").alias("unique_sessions"),
        ) \
        .orderBy("country", "event_type")

    gold_df.write \
        .mode("overwrite") \
        .parquet(gold_path)

    return gold_df

# tests/test_pipeline_integration.py

import json
import io
import pytest
import boto3
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, LongType, StringType, DoubleType

# Импортируем тестируемые функции
from src.pipeline.bronze_ingestion import (
    ingest_json_to_bronze,
    deduplicate_to_silver,
    aggregate_to_gold,
)


# ─── Тестовые данные ─────────────────────────────────────────────────────────

# Тестовый набор событий: реалистичные данные с известными характеристиками.
# Мы точно знаем: 3 уникальных event_id, 1 дублирующийся (event_4 = event_2),
# 1 с null event_type, 2 purchase события общей суммой 350.0.
# Это позволяет писать точные assert'ы.
TEST_EVENTS = [
    {
        "event_id":   "evt-001",
        "user_id":    1001,
        "event_type": "page_view",
        "timestamp":  "2024-01-15T10:00:00Z",
        "session_id": "sess-a",
        "amount":     None,
        "currency":   None,
        "country":    "RU"
    },
    {
        "event_id":   "evt-002",
        "user_id":    1002,
        "event_type": "Purchase",           # намеренно с заглавной - тест нормализации
        "timestamp":  "2024-01-15T10:05:00Z",
        "session_id": "sess-b",
        "amount":     150.0,
        "currency":   "RUB",
        "country":    "RU"
    },
    {
        "event_id":   "evt-003",
        "user_id":    1003,
        "event_type": "purchase",
        "timestamp":  "2024-01-15T10:10:00Z",
        "session_id": "sess-c",
        "amount":     200.0,
        "currency":   "RUB",
        "country":    "DE"
    },
    {
        # Дубликат evt-002: тот же event_id, другие данные
        # После дедупликации должна остаться только одна из двух записей
        "event_id":   "evt-002",
        "user_id":    1002,
        "event_type": "purchase",
        "timestamp":  "2024-01-15T10:05:01Z",   # на секунду позже
        "session_id": "sess-b",
        "amount":     150.0,
        "currency":   "RUB",
        "country":    "RU"
    },
    {
        # Запись без event_type: должна быть отфильтрована в Silver
        "event_id":   "evt-004",
        "user_id":    1004,
        "event_type": None,
        "timestamp":  "2024-01-15T10:15:00Z",
        "session_id": "sess-d",
        "amount":     None,
        "currency":   None,
        "country":    "US"
    },
]


@pytest.fixture(scope="function")
def upload_test_data(s3_client, test_buckets):
    """
    Загружает тестовые события в bronze бакет перед каждым тестом.

    scope="function": каждый тест получает свежие данные.
    Это важно: если предыдущий тест оставил мусор в бакете,
    следующий тест не должен его видеть.

    Данные загружаются в формате JSON Lines (NDJSON):
    каждая строка - отдельный JSON-объект.
    Spark читает такие файлы через spark.read.json() с multiLine=False.

    fixture возвращает путь к загруженным данным,
    чтобы тест мог передать его в ETL функцию.
    """
    date = "2024-01-15"
    input_key = f"incoming/{date}/events.jsonl"

    # Формируем JSON Lines: каждая запись на отдельной строке
    jsonl_content = "\n".join(json.dumps(event, ensure_ascii=False) for event in TEST_EVENTS)
    jsonl_bytes = jsonl_content.encode("utf-8")

    # Загружаем файл в LocalStack S3
    s3_client.put_object(
        Bucket="bronze",
        Key=input_key,
        Body=jsonl_bytes,
        ContentType="application/x-ndjson"
    )
    print(f"\nТестовые данные загружены: s3://bronze/{input_key} ({len(TEST_EVENTS)} событий)")

    yield {
        "source_path": f"s3a://bronze/incoming/{date}/",
        "date": date,
        "event_count": len(TEST_EVENTS)
    }

    # Teardown: удаляем загруженный файл после теста
    s3_client.delete_object(Bucket="bronze", Key=input_key)
    print(f"Тестовые данные удалены: {input_key}")


# ─── Тесты ───────────────────────────────────────────────────────────────────

class TestBronzeIngestion:
    """
    Тесты первой стадии: чтение JSON → запись Parquet в Bronze.
    """

    def test_ingest_writes_correct_row_count(self, spark, s3_client, upload_test_data):
        """
        Проверяем, что Bronze ingestion записывает все строки из источника,
        включая дубликаты и записи с null - на Bronze-стадии мы не фильтруем ничего.

        Ожидание: 5 строк (все исходные события, без фильтрации)
        """
        source_path = upload_test_data["source_path"]
        date = upload_test_data["date"]
        bronze_path = "s3a://bronze/events/"

        row_count = ingest_json_to_bronze(spark, source_path, bronze_path, date)

        # Все 5 исходных событий должны быть записаны
        assert row_count == 5, (
            f"Bronze должен содержать все исходные строки без фильтрации. "
            f"Ожидалось: 5, получено: {row_count}"
        )

    def test_ingest_creates_partition_directory(self, spark, s3_client, upload_test_data):
        """
        Проверяем, что partitionBy("date_partition") создал корректную директорию.

        Spark при записи с partitionBy создаёт структуру:
        bronze/events/date_partition=2024-01-15/*.parquet

        Мы проверяем через boto3 (не через Spark), что нужные файлы появились.
        Это важно: проверка через boto3 убеждает нас, что файлы реально записаны
        в S3, а не просто существуют в памяти Spark.
        """
        source_path = upload_test_data["source_path"]
        date = upload_test_data["date"]
        bronze_path = "s3a://bronze/events/"

        ingest_json_to_bronze(spark, source_path, bronze_path, date)

        # Проверяем через boto3: листинг файлов в партиционированной директории
        expected_prefix = f"events/date_partition={date}/"
        response = s3_client.list_objects_v2(
            Bucket="bronze",
            Prefix=expected_prefix
        )

        parquet_files = [
            obj for obj in response.get("Contents", [])
            if obj["Key"].endswith(".parquet")
        ]

        assert len(parquet_files) > 0, (
            f"В директории s3://bronze/{expected_prefix} должны быть Parquet файлы. "
            f"Найдено: {len(parquet_files)} файлов."
        )

    def test_ingest_adds_technical_fields(self, spark, upload_test_data):
        """
        Проверяем, что Bronze добавляет технические поля:
        ingested_at, date_partition, _source_path.

        Эти поля важны для операционного мониторинга: по ingested_at можно
        отследить SLA (когда данные появились в Bronze), по _source_path —
        откуда пришли данные при troubleshooting.
        """
        source_path = upload_test_data["source_path"]
        date = upload_test_data["date"]
        bronze_path = "s3a://bronze/events/"

        ingest_json_to_bronze(spark, source_path, bronze_path, date)

        # Читаем записанные данные и проверяем схему
        result_df = spark.read.parquet(bronze_path).filter(
            F.col("date_partition") == date
        )

        column_names = set(result_df.columns)
        required_technical_fields = {"ingested_at", "date_partition", "_source_path"}

        missing_fields = required_technical_fields - column_names
        assert not missing_fields, (
            f"В Bronze отсутствуют технические поля: {missing_fields}. "
            f"Все колонки: {column_names}"
        )

        # Проверяем, что date_partition заполнен правильно
        partition_values = {
            row["date_partition"]
            for row in result_df.select("date_partition").distinct().collect()
        }
        assert partition_values == {date}, (
            f"date_partition должен содержать только '{date}'. "
            f"Найдено: {partition_values}"
        )


class TestSilverDeduplication:
    """
    Тесты второй стадии: дедупликация и очистка данных в Silver.
    """

    @pytest.fixture(autouse=True)
    def setup_bronze_data(self, spark, s3_client, upload_test_data):
        """
        autouse=True: эта фикстура запускается перед каждым тестом в классе.
        Создаём Bronze данные как предусловие для Silver тестов.
        """
        source_path = upload_test_data["source_path"]
        date = upload_test_data["date"]
        ingest_json_to_bronze(spark, source_path, "s3a://bronze/events/", date)
        self.date = date

    def test_deduplication_removes_duplicate_event_ids(self, spark):
        """
        Ключевой тест дедупликации.

        В тестовых данных: 5 событий, 1 дубликат (evt-002 встречается дважды).
        После дедупликации: 4 уникальных event_id.

        При этом запись с null event_type (evt-004) ещё не отфильтрована —
        Silver фильтрует по event_type, но не по event_id.
        Итого после дедупликации: 4 строки.
        """
        deduplicate_to_silver(
            spark,
            "s3a://bronze/events/",
            "s3a://silver/events/",
            self.date
        )

        silver_df = spark.read.parquet("s3a://silver/events/").filter(
            F.col("date_partition") == self.date
        )

        # Считаем уникальные event_id - должны все быть уникальными
        total_rows = silver_df.count()
        unique_event_ids = silver_df.select("event_id").distinct().count()

        assert total_rows == unique_event_ids, (
            f"После дедупликации не должно быть дубликатов. "
            f"Всего строк: {total_rows}, уникальных event_id: {unique_event_ids}"
        )

    def test_silver_filters_null_event_type(self, spark):
        """
        Проверяем, что записи с null event_type не попадают в Silver.

        В тестовых данных: evt-004 имеет event_type=null.
        Silver должен её отфильтровать.

        Это важно для downstream процессов: Gold агрегирует по event_type,
        и null-строки создали бы артефакт "неизвестный тип события".
        """
        deduplicate_to_silver(
            spark,
            "s3a://bronze/events/",
            "s3a://silver/events/",
            self.date
        )

        silver_df = spark.read.parquet("s3a://silver/events/").filter(
            F.col("date_partition") == self.date
        )

        null_event_type_count = silver_df.filter(
            F.col("event_type").isNull()
        ).count()

        assert null_event_type_count == 0, (
            f"Silver не должен содержать строки с null event_type. "
            f"Найдено: {null_event_type_count} строк с null event_type."
        )

    def test_silver_normalizes_event_type_to_lowercase(self, spark):
        """
        Проверяем нормализацию event_type.

        В тестовых данных: "Purchase" (с заглавной P) и "purchase" (строчная).
        После нормализации оба должны стать "purchase".

        Без нормализации GROUP BY посчитал бы их как разные события,
        что привело бы к дублированию в Gold агрегатах.
        """
        deduplicate_to_silver(
            spark,
            "s3a://bronze/events/",
            "s3a://silver/events/",
            self.date
        )

        silver_df = spark.read.parquet("s3a://silver/events/").filter(
            F.col("date_partition") == self.date
        )

        # Проверяем, что event_type содержит только строчные буквы
        # upper_case_events: события, где event_type содержит хотя бы одну заглавную букву
        upper_case_events = silver_df.filter(
            F.col("event_type") != F.lower(F.col("event_type"))
        ).count()

        assert upper_case_events == 0, (
            f"Silver должен нормализовать event_type к lowercase. "
            f"Найдено событий с заглавными буквами: {upper_case_events}"
        )

    def test_silver_row_count_after_all_transformations(self, spark):
        """
        Сквозная проверка итогового количества строк в Silver.

        Тестовые данные: 5 событий
        - 1 дубликат (evt-002 второй раз) → отброшен дедупликацией
        - 1 запись с null event_type (evt-004) → отброшена фильтрацией
        Итого в Silver: 5 - 1 - 1 = 3 записи.

        Проверка точного количества строк - это smoke test, который
        сразу покажет если что-то пошло не так в трансформациях.
        """
        row_count = deduplicate_to_silver(
            spark,
            "s3a://bronze/events/",
            "s3a://silver/events/",
            self.date
        )

        assert row_count == 3, (
            f"После дедупликации и фильтрации должно остаться 3 строки "
            f"(5 исходных - 1 дубликат - 1 null event_type). "
            f"Получено: {row_count}"
        )


class TestGoldAggregation:
    """
    Тесты третьей стадии: бизнес-агрегаты в Gold.
    """

    @pytest.fixture(autouse=True)
    def setup_silver_data(self, spark, s3_client, upload_test_data):
        """
        Создаём Silver данные как предусловие для Gold тестов.
        """
        date = upload_test_data["date"]
        source_path = upload_test_data["source_path"]
        ingest_json_to_bronze(spark, source_path, "s3a://bronze/events/", date)
        deduplicate_to_silver(spark, "s3a://bronze/events/", "s3a://silver/events/", date)
        self.date = date

    def test_gold_aggregates_purchase_amounts_correctly(self, spark):
        """
        Проверяем корректность агрегации сумм покупок по стране.

        В тестовых данных (после Silver обработки):
        - RU + purchase: evt-002 с amount=150.0 (дубликат отброшен)
        - DE + purchase: evt-003 с amount=200.0

        Gold должен вернуть:
        - RU, purchase: total_amount=150.0
        - DE, purchase: total_amount=200.0
        """
        aggregate_to_gold(
            spark,
            "s3a://silver/events/",
            "s3a://gold/daily_stats/",
            self.date
        )

        gold_df = spark.read.parquet("s3a://gold/daily_stats/")

        # Собираем результаты в словарь для удобной проверки
        # Ключ: (country, event_type), Значение: total_amount
        results = {
            (row["country"], row["event_type"]): row["total_amount"]
            for row in gold_df.collect()
        }

        # Проверяем суммы по стране
        ru_purchase_amount = results.get(("RU", "purchase"), 0.0)
        de_purchase_amount = results.get(("DE", "purchase"), 0.0)

        assert abs(ru_purchase_amount - 150.0) < 0.01, (
            f"Total amount для RU/purchase должен быть 150.0. "
            f"Получено: {ru_purchase_amount}"
        )

        assert abs(de_purchase_amount - 200.0) < 0.01, (
            f"Total amount для DE/purchase должен быть 200.0. "
            f"Получено: {de_purchase_amount}"
        )

    def test_gold_unique_users_count(self, spark):
        """
        Проверяем подсчёт уникальных пользователей.

        В тестовых данных (Silver):
        - RU: user_id 1001 (page_view) и 1002 (purchase) = 2 уникальных пользователя
        - DE: user_id 1003 (purchase) = 1 уникальный пользователь
        """
        aggregate_to_gold(
            spark,
            "s3a://silver/events/",
            "s3a://gold/daily_stats/",
            self.date
        )

        gold_df = spark.read.parquet("s3a://gold/daily_stats/")

        # Суммарное количество уникальных пользователей по всем строкам
        # (сумма countDistinct по каждому country+event_type)
        total_unique_users_per_segment = gold_df.agg(
            F.sum("unique_users").alias("total")
        ).collect()[0]["total"]

        # В наших данных: RU/page_view=1, RU/purchase=1, DE/purchase=1
        # Итого 3 пользователя в трёх сегментах
        assert total_unique_users_per_segment == 3, (
            f"Суммарное количество unique_users по всем сегментам должно быть 3. "
            f"Получено: {total_unique_users_per_segment}"
        )

    def test_gold_schema(self, spark):
        """
        Проверяем схему Gold таблицы.

        Схема Gold - это контракт с BI-инструментами и аналитиками.
        Если колонка переименована или удалена, дашборды сломаются.
        Этот тест гарантирует, что схема не изменится незаметно.
        """
        aggregate_to_gold(
            spark,
            "s3a://silver/events/",
            "s3a://gold/daily_stats/",
            self.date
        )

        gold_df = spark.read.parquet("s3a://gold/daily_stats/")

        expected_columns = {
            "country", "event_type", "date_partition",
            "event_count", "unique_users", "total_amount", "unique_sessions"
        }

        actual_columns = set(gold_df.columns)
        missing = expected_columns - actual_columns
        unexpected = actual_columns - expected_columns

        assert not missing, f"В Gold отсутствуют обязательные колонки: {missing}"
        assert not unexpected, f"В Gold появились неожиданные колонки: {unexpected}"

# Запуск всех интеграционных тестов
pytest tests/test_pipeline_integration.py -v

# Запуск только тестов Bronze стадии
pytest tests/test_pipeline_integration.py::TestBronzeIngestion -v

# Запуск конкретного теста
pytest tests/test_pipeline_integration.py::TestSilverDeduplication::test_deduplication_removes_duplicate_event_ids -v

# С выводом print() и логов LocalStack
pytest tests/test_pipeline_integration.py -v -s

# С таймаутом 120 секунд на тест (через pytest-timeout)
pytest tests/test_pipeline_integration.py --timeout=120

================================ test session starts ================================
platform linux -- Python 3.11.4, pytest-8.1.0

collected 8 items

LocalStack endpoint: http://localhost:54321           ← Testcontainers выбрал порт
Тестовые данные загружены: s3://bronze/incoming/...

tests/test_pipeline_integration.py::TestBronzeIngestion::test_ingest_writes_correct_row_count PASSED [12%]
tests/test_pipeline_integration.py::TestBronzeIngestion::test_ingest_creates_partition_directory PASSED [25%]
tests/test_pipeline_integration.py::TestBronzeIngestion::test_ingest_adds_technical_fields PASSED [37%]
tests/test_pipeline_integration.py::TestSilverDeduplication::test_deduplication_removes_duplicate_event_ids PASSED [50%]
tests/test_pipeline_integration.py::TestSilverDeduplication::test_silver_filters_null_event_type PASSED [62%]
tests/test_pipeline_integration.py::TestSilverDeduplication::test_silver_normalizes_event_type_to_lowercase PASSED [75%]
tests/test_pipeline_integration.py::TestGoldAggregation::test_gold_aggregates_purchase_amounts_correctly PASSED [87%]
tests/test_pipeline_integration.py::TestGoldAggregation::test_gold_schema PASSED [100%]

========================= 8 passed in 47.23s ==========================

botocore.exceptions.ClientError: An error occurred (SignatureDoesNotMatch)
when calling the PutObject operation: The request signature we calculated
does not match the signature you provided. Check your key and signing method.

from botocore.config import Config

client = boto3.client(
    "s3",
    endpoint_url=endpoint_url,
    config=Config(signature_version="s3v4")  # ← принудительно V4
)

py4j.protocol.Py4JJavaError: An error occurred while calling o25.parquet.
: java.io.IOException: No FileSystem for scheme "s3a"

.config(
    "spark.jars.packages",
    "org.apache.hadoop:hadoop-aws:3.3.4,"
    "com.amazonaws:aws-java-sdk-bundle:1.12.367"
)

Caused by: java.net.ConnectException: Connection refused (Connection refused)
или
Caused by: java.net.UnknownHostException: bronze.localhost

# Убедитесь что используете порт из Testcontainers, не хардкодите
endpoint = localstack_container.get_url()  # правильно
endpoint = "http://localhost:4566"           # неправильно - порт может быть другим

.config("spark.hadoop.fs.s3a.path.style.access", "true")  # обязательно!

# Тест зависает на несколько минут без вывода
# Потом падает с:
com.amazonaws.SdkClientException: Unable to load AWS credentials from any provider

.config(
    "spark.hadoop.fs.s3a.aws.credentials.provider",
    "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
)
.config("spark.hadoop.fs.s3a.access.key", "test")
.config("spark.hadoop.fs.s3a.secret.key", "test")

javax.net.ssl.SSLException: Unrecognized SSL message, plaintext connection?

.config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")

# .github/workflows/integration-tests.yml
name: Integration Tests

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  integration-tests:
    runs-on: ubuntu-22.04
    timeout-minutes: 20    # максимальное время всего job

    steps:
      # ── Checkout ────────────────────────────────────────────────────────────
      - name: Checkout code
        uses: actions/checkout@v4

      # ── Кеш для Maven/Ivy зависимостей Spark ──────────────────────────────
      # Spark при первом запуске скачивает hadoop-aws и aws-java-sdk через Maven.
      # На GitHub Actions это занимает 2-3 минуты.
      # Кеш кешируем по хешу файла зависимостей - обновляется только при изменении.
      - name: Cache Spark/Maven dependencies
        uses: actions/cache@v4
        with:
          path: |
            ~/.ivy2/cache
            ~/.m2/repository
          key: spark-deps-${{ hashFiles('requirements-test.txt') }}-hadoop3.3.4
          restore-keys: |
            spark-deps-${{ hashFiles('requirements-test.txt') }}-
            spark-deps-

      # ── Кеш для Docker образов LocalStack ───────────────────────────────────
      # Docker pull localstack/localstack:3.4 занимает ~30 секунд.
      # GitHub Actions кеш помогает ускорить это до 5 секунд.
      - name: Cache Docker images
        uses: ScribeMD/docker-cache@0.5.0
        with:
          key: docker-localstack-3.4-${{ runner.os }}

      # ── Python окружение ─────────────────────────────────────────────────────
      - name: Set up Python 3.11
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: "pip"

      - name: Install dependencies
        run: pip install -r requirements-test.txt

      # ── Java для Spark ───────────────────────────────────────────────────────
      - name: Set up Java 17
        uses: actions/setup-java@v4
        with:
          distribution: temurin
          java-version: "17"
          cache: maven

      # ── Запуск тестов ────────────────────────────────────────────────────────
      - name: Run integration tests
        run: |
          pytest tests/test_pipeline_integration.py \
            -v \
            --timeout=120 \
            --tb=short \
            -p no:warnings \
            --junit-xml=test-results.xml
        env:
          # Testcontainers нужен доступ к Docker socket
          DOCKER_HOST: unix:///var/run/docker.sock
          # Отключаем Ryuk (cleanup контейнер) - он требует привилегированного режима
          # на некоторых CI системах. Testcontainers всё равно очищает контейнеры.
          TESTCONTAINERS_RYUK_DISABLED: "true"

      # ── Публикация результатов ────────────────────────────────────────────────
      - name: Publish test results
        uses: mikepenz/action-junit-report@v4
        if: always()     # запускаем даже если тесты упали
        with:
          report_paths: test-results.xml

# .gitlab-ci.yml
integration-tests:
  image: python:3.11-bullseye
  services:
    # GitLab CI поднимает Docker-in-Docker сервис, который нужен Testcontainers
    - docker:24-dind

  variables:
    DOCKER_HOST: tcp://docker:2376
    DOCKER_TLS_CERTDIR: "/certs"
    DOCKER_TLS_VERIFY: "1"
    DOCKER_CERT_PATH: "/certs/client"
    TESTCONTAINERS_HOST_OVERRIDE: "docker"   # адрес DinD хоста для Testcontainers

  cache:
    key: "${CI_COMMIT_REF_SLUG}-spark-deps"
    paths:
      - .ivy2/cache/
      - .m2/repository/

  before_script:
    - apt-get update && apt-get install -y default-jdk-headless
    - pip install -r requirements-test.txt

  script:
    - pytest tests/test_pipeline_integration.py
        --timeout=120
        -v
        --junit-xml=test-results.xml

  artifacts:
    when: always
    reports:
      junit: test-results.xml

  timeout: 20 minutes

# Переиспользование контейнера через переменную окружения
import os

@pytest.fixture(scope="session")
def localstack_container():
    """
    В режиме TESTCONTAINERS_REUSE_ENABLE=true контейнер не удаляется после теста.
    При следующем запуске pytest Testcontainers находит уже запущенный контейнер
    и переиспользует его вместо создания нового.

    Включить: export TESTCONTAINERS_REUSE_ENABLE=true
    Отключить: export TESTCONTAINERS_REUSE_ENABLE=false (по умолчанию)

    ВАЖНО: никогда не включайте reuse в CI. Там контейнеры должны создаваться
    заново при каждом запуске для чистоты окружения.
    """
    reuse = os.getenv("TESTCONTAINERS_REUSE_ENABLE", "false").lower() == "true"

    with LocalStackContainer(
        image="localstack/localstack:3.4",
        reuse=reuse
    ) as container:
        yield container

# tests/test_delta_pipeline.py

import pytest
from pyspark.sql import SparkSession, functions as F
from delta import DeltaTable, configure_spark_with_delta_pip


@pytest.fixture(scope="session")
def spark_delta(localstack_endpoint):
    """
    SparkSession с поддержкой Delta Lake и LocalStack S3.

    Delta Lake требует дополнительных JAR и конфигурации.
    configure_spark_with_delta_pip - удобная функция из пакета delta-spark,
    которая автоматически добавляет нужные зависимости.
    """
    builder = (
        SparkSession.builder
        .master("local[2]")
        .appName("delta-integration-tests")
        # Delta Lake расширения
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config(
            "spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog"
        )
        # S3A для LocalStack
        .config("spark.hadoop.fs.s3a.endpoint", localstack_endpoint)
        .config("spark.hadoop.fs.s3a.path.style.access", "true")
        .config(
            "spark.hadoop.fs.s3a.aws.credentials.provider",
            "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
        )
        .config("spark.hadoop.fs.s3a.access.key", "test")
        .config("spark.hadoop.fs.s3a.secret.key", "test")
        .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
        .config("spark.sql.shuffle.partitions", "4")
    )

    spark = configure_spark_with_delta_pip(builder).getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    yield spark
    spark.stop()


class TestDeltaLakeWithLocalStack:
    """
    Тесты Delta Lake операций против LocalStack S3.

    Эти тесты проверяют, что Delta Lake корректно:
    1. Создаёт таблицы и _delta_log/ в S3
    2. Поддерживает ACID транзакции (MERGE, UPDATE)
    3. Сохраняет историю версий
    4. Работает с Schema Enforcement
    """

    def test_delta_write_and_read(self, spark_delta, test_buckets):
        """
        Базовый тест: запись и чтение Delta Lake таблицы через LocalStack S3.

        Delta при записи создаёт:
        - s3a://silver/delta-events/part-*.parquet - данные
        - s3a://silver/delta-events/_delta_log/00000000000000000000.json - лог транзакций

        Если LocalStack S3 работает корректно, оба типа файлов должны
        создаться и быть доступны для чтения.
        """
        data = [
            (1, "evt-001", "purchase", 100.0),
            (2, "evt-002", "page_view", None),
            (3, "evt-003", "purchase", 200.0),
        ]
        schema = ["user_id", "event_id", "event_type", "amount"]
        df = spark_delta.createDataFrame(data, schema)

        delta_path = "s3a://silver/delta-events/"

        # Записываем как Delta таблицу
        df.write.format("delta").mode("overwrite").save(delta_path)

        # Читаем обратно и проверяем
        result_df = spark_delta.read.format("delta").load(delta_path)
        assert result_df.count() == 3, "Delta таблица должна содержать 3 строки"

    def test_delta_merge_upsert(self, spark_delta, test_buckets):
        """
        Тест MERGE (UPSERT) операции.

        MERGE - одна из ключевых возможностей Delta Lake для CDC-пайплайнов.
        Синхронизирует целевую таблицу с источником: обновляет существующие
        строки, вставляет новые. При этом операция атомарна: либо применились
        все изменения, либо ни одно.

        Это комплексный тест, потому что MERGE использует несколько S3-операций:
        1. Чтение существующих файлов
        2. Запись новых Parquet файлов
        3. Обновление _delta_log с информацией о транзакции
        """
        # Создаём начальные данные
        initial_data = [
            (1, "evt-001", "purchase", 100.0, "RU"),
            (2, "evt-002", "page_view", None, "RU"),
        ]
        schema = ["user_id", "event_id", "event_type", "amount", "country"]
        initial_df = spark_delta.createDataFrame(initial_data, schema)

        delta_path = "s3a://silver/delta-merge-test/"
        initial_df.write.format("delta").mode("overwrite").save(delta_path)

        # Данные обновления: изменяем evt-001 и добавляем evt-003
        updates_data = [
            (1, "evt-001", "purchase", 150.0, "RU"),   # обновлённый amount
            (3, "evt-003", "login",    None,  "DE"),    # новая запись
        ]
        updates_df = spark_delta.createDataFrame(updates_data, schema)

        # Выполняем MERGE
        target_table = DeltaTable.forPath(spark_delta, delta_path)
        target_table.alias("target").merge(
            updates_df.alias("source"),
            "target.event_id = source.event_id"
        ).whenMatchedUpdateAll() \
         .whenNotMatchedInsertAll() \
         .execute()

        # Проверяем результат
        result_df = spark_delta.read.format("delta").load(delta_path)
        results = {
            row["event_id"]: row["amount"]
            for row in result_df.collect()
        }

        # evt-001 должен иметь обновлённый amount
        assert results["evt-001"] == 150.0, (
            f"MERGE должен обновить amount для evt-001 до 150.0. "
            f"Текущий: {results['evt-001']}"
        )

        # evt-003 должен быть добавлен
        assert "evt-003" in results, "MERGE должен вставить новую запись evt-003"

        # Всего должно быть 3 записи (2 исходных + 1 новая)
        assert len(results) == 3, f"После MERGE должно быть 3 записи. Найдено: {len(results)}"

    def test_delta_time_travel(self, spark_delta, test_buckets):
        """
        Тест Time Travel: чтение исторических версий таблицы.

        Delta Lake сохраняет историю транзакций в _delta_log/.
        Time Travel позволяет читать данные на момент конкретной версии или timestamp.
        Это мощная возможность для аудита данных и восстановления от ошибок.

        Для работы Time Travel все версии Parquet файлов должны оставаться в S3.
        LocalStack корректно их хранит, поэтому Time Travel работает так же,
        как в продакшне с настоящим S3.
        """
        delta_path = "s3a://gold/delta-time-travel-test/"

        # Версия 0: первоначальная запись
        v0_data = spark_delta.createDataFrame(
            [(1, "alice", 100), (2, "bob", 200)],
            ["id", "name", "score"]
        )
        v0_data.write.format("delta").mode("overwrite").save(delta_path)

        # Версия 1: добавляем записи
        v1_data = spark_delta.createDataFrame(
            [(3, "charlie", 300)],
            ["id", "name", "score"]
        )
        v1_data.write.format("delta").mode("append").save(delta_path)

        # Читаем текущую версию (v1)
        current = spark_delta.read.format("delta").load(delta_path)
        assert current.count() == 3, "Текущая версия должна содержать 3 записи"

        # Читаем версию 0 через Time Travel
        v0 = spark_delta.read.format("delta") \
            .option("versionAsOf", 0) \
            .load(delta_path)
        assert v0.count() == 2, "Версия 0 должна содержать 2 записи (до append)"