Testing: изоляция от внешних источников без mock-антипаттернов

# АНТИПАТТЕРН: тестируем Python-вызовы, не Spark-логику
from unittest.mock import MagicMock, patch

def test_pipeline_with_mocks():
    """
    Этот тест ПРОХОДИТ всегда, даже если пайплайн сломан.
    Мы тестируем, что Python вызвал нужные методы,
    но не то, что Spark трансформации корректны.
    """
    mock_spark = MagicMock()
    mock_df = MagicMock()

    # mock возвращает другой mock при любом вызове
    mock_spark.read.parquet.return_value = mock_df
    mock_df.filter.return_value = mock_df
    mock_df.withColumn.return_value = mock_df
    mock_df.write.format.return_value = mock_df.write

    with patch("src.pipeline.SparkSession", return_value=mock_spark):
        run_pipeline("s3://bronze/", "s3://silver/")

    # Что мы проверяем? Только то, что .filter() был вызван.
    # Не то, что фильтр работает правильно с реальными данными.
    # Не то, что схема DataFrame совместима с filter-предикатом.
    # Не то, что write произошёл с корректными данными.
    mock_df.filter.assert_called_once()  # ← Это не тест логики. Это тест вызова.
    # Если переименуем .filter() в .where() → тест упадёт, логика не изменилась

"""
pipeline/io/readers.py
Конфигурируемые Reader-функции.
В unit-тестах не используются вообще.
В integration-тестах получают tmp_path вместо prod-пути.
"""

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.types import StructType


def read_bronze_orders(spark: SparkSession, source_path: str) -> DataFrame:
    """
    Читает сырые заказы из Parquet.

    Args:
        source_path: путь к данным - prod: s3://bronze/orders/
                                      test: /tmp/pytest-xxx/orders/
    """
    return spark.read.parquet(source_path)


def read_bronze_orders_with_schema(
    spark: SparkSession,
    source_path: str,
    schema: StructType,
) -> DataFrame:
    """
    Читает с явной схемой вместо schema inference.

    Явная схема критична для тестов: предотвращает schema drift
    и делает поведение предсказуемым независимо от данных.
    """
    return spark.read.schema(schema).parquet(source_path)


def read_customers_from_jdbc(
    spark: SparkSession,
    jdbc_url: str,          # test: jdbc:postgresql://localhost:PORT/testdb
    table: str,             # prod/test: одинаково
    credentials: dict,      # test: {"user": "test", "password": "test"}
) -> DataFrame:
    """
    Читает из PostgreSQL через JDBC.
    В integration-тестах jdbc_url указывает на testcontainers-контейнер.
    """
    return (
        spark.read
        .format("jdbc")
        .option("url", jdbc_url)
        .option("dbtable", table)
        .option("user", credentials["user"])
        .option("password", credentials["password"])
        .option("driver", "org.postgresql.Driver")
        .load()
    )

"""
pipeline/io/writers.py
Конфигурируемые Writer-функции с явными параметрами пути.
"""

from pyspark.sql import DataFrame


def write_silver_orders_parquet(
    df: DataFrame,
    output_path: str,    # test: tmp_path / prod: s3://silver/orders/
    partition_col: str = "order_date",
    mode: str = "overwrite",
) -> None:
    """
    Записывает Silver заказы в Parquet.
    Идемпотентна: overwrite гарантирует отсутствие дублей при повторном запуске.
    """
    (
        df.write
        .partitionBy(partition_col)
        .mode(mode)
        .parquet(output_path)
    )


def write_gold_kpi_delta(
    df: DataFrame,
    output_path: str,
    date: str,
) -> None:
    """
    Записывает Gold KPI в Delta Lake.
    replaceWhere - атомарная замена партиции (идемпотентна).
    """
    (
        df.write
        .format("delta")
        .mode("overwrite")
        .option("replaceWhere", f"kpi_date = '{date}'")
        .save(output_path)
    )

"""
pipeline/transformations.py
Чистые трансформационные функции - сердце пайплайна.
Принимают DataFrame → возвращают DataFrame.
Никакого IO. Никаких side effects.
"""

from pyspark.sql import DataFrame, functions as F
from pyspark.sql.window import Window


def clean_raw_orders(df: DataFrame) -> DataFrame:
    """Очищает сырые заказы: убирает невалидные строки, нормализует типы."""
    return (
        df
        .filter(F.col("order_id").isNotNull())
        .filter(F.col("quantity") > 0)
        .withColumn("order_id",   F.trim(F.col("order_id")))
        .withColumn("discount",   F.coalesce(F.col("discount"), F.lit(0.0)))
    )


def deduplicate_orders(df: DataFrame, key_col: str = "order_id") -> DataFrame:
    """Дедупликация: при дублях оставляем строку с максимальным updated_at."""
    window = Window.partitionBy(key_col).orderBy(F.col("updated_at").desc_nulls_last())
    return (
        df
        .withColumn("_rn", F.row_number().over(window))
        .filter(F.col("_rn") == 1)
        .drop("_rn")
    )


def enrich_with_customer(orders: DataFrame, customers: DataFrame) -> DataFrame:
    """Обогащает заказы данными клиента через LEFT JOIN."""
    customers_unique = customers.dropDuplicates(["customer_id"])
    return orders.join(
        customers_unique.select("customer_id", "customer_name", "segment"),
        on="customer_id",
        how="left",
    )


def calculate_order_total(df: DataFrame) -> DataFrame:
    """Вычисляет итоговую сумму с учётом скидки."""
    return df.withColumn(
        "order_total",
        F.round(F.col("quantity") * F.col("unit_price") * (1 - F.col("discount")), 2)
    )


def aggregate_daily_kpi(df: DataFrame) -> DataFrame:
    """Агрегирует суточные KPI по сегментам клиентов."""
    return (
        df
        .groupBy("order_date", "segment")
        .agg(
            F.count("order_id").alias("order_count"),
            F.sum("order_total").alias("revenue"),
            F.avg("order_total").alias("avg_order_value"),
            F.countDistinct("customer_id").alias("unique_customers"),
        )
    )


# ── Главный пайплайн: оркестрирует IO + трансформации ──────────────────────

def run_orders_pipeline(
    spark,
    source_path: str,         # Передаётся снаружи → легко подменить в тесте
    customers_jdbc_url: str,
    customers_credentials: dict,
    silver_output_path: str,  # Передаётся снаружи → легко подменить в тесте
    gold_output_path: str,
    date: str,
) -> None:
    """
    Основной пайплайн.

    Все пути и connection strings - параметры.
    Это и есть Dependency Injection:
    тест передаёт tmp_path, production - реальные пути.
    """
    from pipeline.io.readers import read_bronze_orders, read_customers_from_jdbc
    from pipeline.io.writers import write_silver_orders_parquet, write_gold_kpi_delta

    # IO: Чтение
    orders_raw = read_bronze_orders(spark, f"{source_path}/date={date}/")
    customers  = read_customers_from_jdbc(spark, customers_jdbc_url,
                                          "public.customers", customers_credentials)

    # Трансформации (чистые функции)
    orders_clean    = clean_raw_orders(orders_raw)
    orders_deduped  = deduplicate_orders(orders_clean)
    orders_enriched = enrich_with_customer(orders_deduped, customers)
    orders_total    = calculate_order_total(orders_enriched)

    # IO: Запись Silver
    write_silver_orders_parquet(orders_total, silver_output_path, date)

    # Gold KPI
    kpi = aggregate_daily_kpi(orders_total)
    write_gold_kpi_delta(kpi, gold_output_path, date)

"""
tests/integration/test_io_isolation.py
Integration-тесты с Local Directory Substitution.
"""

import pytest
from pathlib import Path
from pyspark.testing import assertDataFrameEqual
from pyspark.sql import functions as F
from pipeline.io.writers import write_silver_orders_parquet
from pipeline.transformations import clean_raw_orders, calculate_order_total


def test_write_parquet_and_read_back(spark, tmp_path):
    """
    Integration-тест: writer записывает корректный Parquet.

    Сценарий:
    1. Создаём DataFrame с тестовыми данными (synthetic)
    2. Пишем через writer-функцию в tmp_path
    3. Читаем обратно через spark.read
    4. Сравниваем через assertDataFrameEqual

    Это тестирует реальную запись/чтение Parquet с реальным Spark,
    но без S3 и без сетевых зависимостей.
    """
    # Arrange: создаём тестовые данные
    orders = spark.createDataFrame(
        [("ORD-001", "CUST-01", "2024-01-15", 18.0),
         ("ORD-002", "CUST-02", "2024-01-15", 50.0)],
        schema="order_id STRING, customer_id STRING, order_date STRING, order_total DOUBLE"
    )

    output_path = str(tmp_path / "silver_orders")

    # Act: записываем через writer (без partition для простоты теста)
    orders.write.mode("overwrite").parquet(output_path)

    # Assert: читаем обратно и проверяем
    result = spark.read.parquet(output_path)
    assertDataFrameEqual(orders, result, checkRowOrder=False)


def test_partition_overwrite_idempotency(spark, tmp_path):
    """
    Критичный integration-тест: идемпотентность OVERWRITE.

    Проверяем, что повторная запись той же партиции
    не создаёт дубли (это распространённый production-баг).

    Реализация в Spark: write.partitionBy + mode("overwrite")
    с spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
    перезаписывает только конкретные партиции, не всю таблицу.
    """
    spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

    output_path = str(tmp_path / "partitioned_orders")

    orders_day1 = spark.createDataFrame(
        [("ORD-001", "2024-01-15", 18.0),
         ("ORD-002", "2024-01-15", 50.0)],
        schema="order_id STRING, order_date STRING, order_total DOUBLE"
    )
    orders_day2 = spark.createDataFrame(
        [("ORD-003", "2024-01-16", 30.0)],
        schema="order_id STRING, order_date STRING, order_total DOUBLE"
    )

    # Первый запуск: записываем 2 партиции
    orders_day1.write.partitionBy("order_date").mode("overwrite").parquet(output_path)
    orders_day2.write.partitionBy("order_date").mode("overwrite").parquet(output_path)

    # Повторный запуск day1: должен перезаписать только 2024-01-15
    orders_day1.write.partitionBy("order_date").mode("overwrite").parquet(output_path)

    # После повторного запуска: должно быть 3 строки (2 + 1), не 5 (2+2+1)
    result = spark.read.parquet(output_path)
    assert result.count() == 3, \
        f"Idempotent overwrite failed: expected 3 rows, got {result.count()}"


def test_full_pipeline_with_local_paths(spark, tmp_path):
    """
    End-to-end integration-тест с Local Directory Substitution.

    Тот же пайплайн clean → calculate → write, но теперь мы проверяем
    не только трансформации, но и корректность записи/чтения.
    """
    # Arrange: Bronze данные на локальном диске
    bronze_path = str(tmp_path / "bronze")
    spark.createDataFrame(
        [("ORD-001", "CUST-01", 2, 10.0, 0.1, "2024-01-15 10:00:00"),
         ("ORD-002", "CUST-02", 1, 50.0, 0.0, "2024-01-15 11:00:00"),
         (None,      "CUST-03", 3, 20.0, 0.0, "2024-01-15 12:00:00")],  # NULL → отфильтровать
        schema="order_id STRING, customer_id STRING, quantity INT, "
               "unit_price DOUBLE, discount DOUBLE, updated_at STRING"
    ).write.parquet(bronze_path)

    silver_path = str(tmp_path / "silver")

    # Act: читаем Bronze, трансформируем, пишем Silver
    orders_raw = spark.read.parquet(bronze_path)
    orders_clean = clean_raw_orders(orders_raw)
    orders_total = calculate_order_total(orders_clean)
    orders_total.write.mode("overwrite").parquet(silver_path)

    # Assert: читаем Silver и проверяем результат
    silver = spark.read.parquet(silver_path)

    # Строка с NULL order_id не должна попасть в Silver
    assert silver.filter("order_id IS NULL").count() == 0

    # Проверяем корректность расчёта total для конкретного заказа
    ord1 = silver.filter("order_id = 'ORD-001'").select("order_total").collect()[0]["order_total"]
    assert abs(ord1 - 18.0) < 0.01, f"Expected 18.0, got {ord1}"

# tests/conftest.py - добавляем Delta Lake конфигурацию
import pytest
from pyspark.sql import SparkSession


@pytest.fixture(scope="session")
def spark_delta():
    """
    SparkSession с Delta Lake расширениями для интеграционных тестов.

    Важные настройки:
    - spark.sql.extensions: регистрирует Delta Lake команды (MERGE, OPTIMIZE, etc.)
    - spark.sql.catalog.spark_catalog: использует Delta как default catalog
    - catalogImplementation=in-memory: НЕ нужен Hive Metastore! Работает в CI без доп. сервисов
    """
    return (
        SparkSession.builder
        .master("local[2]")
        .appName("delta-integration-tests")
        .config("spark.sql.extensions",
                "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog",
                "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .config("spark.sql.catalogImplementation", "in-memory")  # Нет Hive!
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.driver.memory", "2g")
        .config("spark.ui.enabled", "false")
        .getOrCreate()
    )


def test_delta_merge_upsert(spark_delta, tmp_path):
    """
    Integration-тест: MERGE INTO в Delta Lake.

    Проверяем, что UPSERT логика работает корректно:
    - Существующая строка → UPDATE (если изменились данные)
    - Новая строка → INSERT
    - Удалённая строка → остаётся (нет DELETE в этом сценарии)
    """
    delta_path = str(tmp_path / "delta_orders")

    # Создаём начальное состояние таблицы
    initial = spark_delta.createDataFrame(
        [("ORD-001", 18.0, "2024-01-15"),
         ("ORD-002", 50.0, "2024-01-15")],
        schema="order_id STRING, order_total DOUBLE, order_date STRING"
    )
    initial.write.format("delta").save(delta_path)

    # Инкремент: ORD-001 обновился (новая сумма), ORD-003 - новый
    increment = spark_delta.createDataFrame(
        [("ORD-001", 20.0, "2024-01-15"),  # UPDATE
         ("ORD-003", 35.0, "2024-01-15")], # INSERT
        schema="order_id STRING, order_total DOUBLE, order_date STRING"
    )
    increment.createOrReplaceTempView("increment_data")

    # Выполняем MERGE
    spark_delta.sql(f"""
        MERGE INTO delta.`{delta_path}` AS target
        USING increment_data AS source
        ON target.order_id = source.order_id
        WHEN MATCHED THEN
            UPDATE SET target.order_total = source.order_total
        WHEN NOT MATCHED THEN
            INSERT (order_id, order_total, order_date)
            VALUES (source.order_id, source.order_total, source.order_date)
    """)

    # Проверяем результат
    result = spark_delta.read.format("delta").load(delta_path)
    assert result.count() == 3  # ORD-001, ORD-002, ORD-003

    updated_total = (
        result
        .filter("order_id = 'ORD-001'")
        .select("order_total")
        .collect()[0]["order_total"]
    )
    assert abs(updated_total - 20.0) < 0.01, \
        f"ORD-001 should be updated to 20.0, got {updated_total}"

# Установка зависимостей
pip install testcontainers[postgres]
# Или для более полной версии:
pip install testcontainers

# Требования:
# - Docker установлен и запущен
# - PostgreSQL JDBC драйвер для Spark:
#   spark.jars = /opt/postgresql-42.7.0.jar

# tests/integration/test_jdbc_reader.py
# Integration-тест: чтение из PostgreSQL через Spark JDBC

import pytest
from testcontainers.postgres import PostgresContainer
from pyspark.testing import assertDataFrameEqual
from pipeline.io.readers import read_customers_from_jdbc


@pytest.fixture(scope="module")
def postgres_container():
    """
    Поднимает контейнер PostgreSQL для тестов в этом модуле.

    scope="module": один контейнер на весь модуль тестов.
    Это быстрее, чем поднимать контейнер на каждый тест,
    но требует что каждый тест убирает за собой (или использует транзакции).

    Fixture автоматически останавливает контейнер после завершения всех тестов.
    """
    with PostgresContainer(
        image="postgres:15-alpine",  # Лёгкий alpine-образ
        username="testuser",
        password="testpass",
        dbname="testdb",
    ) as postgres:
        yield postgres


@pytest.fixture(scope="module")
def postgres_jdbc_url(postgres_container):
    """Возвращает JDBC URL для подключения к контейнеру."""
    # testcontainers маппирует порт контейнера на случайный порт хоста
    # get_connection_url() возвращает строку вида: postgresql://localhost:RANDOM_PORT/testdb
    return (
        f"jdbc:postgresql://{postgres_container.get_container_host_ip()}:"
        f"{postgres_container.get_exposed_port(5432)}/testdb"
    )


@pytest.fixture(scope="module", autouse=True)
def setup_test_data(postgres_container):
    """
    Создаёт тестовые таблицы и данные в PostgreSQL контейнере.

    autouse=True: выполняется автоматически перед всеми тестами в модуле.
    """
    import psycopg2

    conn = psycopg2.connect(
        host=postgres_container.get_container_host_ip(),
        port=postgres_container.get_exposed_port(5432),
        user="testuser",
        password="testpass",
        dbname="testdb",
    )
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS customers (
            customer_id   VARCHAR(50) PRIMARY KEY,
            customer_name VARCHAR(200),
            segment       VARCHAR(50),
            created_at    TIMESTAMP DEFAULT NOW()
        )
    """)

    cursor.execute("""
        INSERT INTO customers (customer_id, customer_name, segment) VALUES
        ('CUST-01', 'Alice Smith',  'Premium'),
        ('CUST-02', 'Bob Johnson',  'Standard'),
        ('CUST-03', 'Carol White',  'Premium')
        ON CONFLICT DO NOTHING
    """)

    conn.commit()
    cursor.close()
    conn.close()


def test_read_customers_via_jdbc(spark, postgres_jdbc_url):
    """
    Integration-тест: Spark читает данные из реального PostgreSQL.

    Тестирует всю цепочку: Spark JDBC → PostgreSQL → DataFrame.
    Это невозможно протестировать через unit-тест или mock.
    """
    credentials = {"user": "testuser", "password": "testpass"}

    result = read_customers_from_jdbc(
        spark=spark,
        jdbc_url=postgres_jdbc_url,
        table="customers",
        credentials=credentials,
    )

    # Проверяем, что все три строки прочитаны корректно
    assert result.count() == 3

    # Проверяем, что схема прочитана корректно
    assert "customer_id" in result.columns
    assert "customer_name" in result.columns
    assert "segment" in result.columns

    # Проверяем конкретную строку
    alice = result.filter("customer_id = 'CUST-01'").collect()
    assert len(alice) == 1
    assert alice[0]["customer_name"] == "Alice Smith"
    assert alice[0]["segment"] == "Premium"


def test_jdbc_reader_with_filter_pushdown(spark, postgres_jdbc_url):
    """
    Проверяем, что Spark правильно применяет предикаты к JDBC источнику.

    Filter pushdown: Spark передаёт WHERE условие в PostgreSQL
    вместо того, чтобы читать всю таблицу и фильтровать на стороне Spark.
    Для теста нас интересует корректность результата.
    """
    credentials = {"user": "testuser", "password": "testpass"}

    # Читаем только Premium сегмент через predicates
    result = (
        spark.read
        .format("jdbc")
        .option("url", postgres_jdbc_url)
        .option("dbtable", "(SELECT * FROM customers WHERE segment = 'Premium') as premium_customers")
        .option("user", credentials["user"])
        .option("password", credentials["password"])
        .option("driver", "org.postgresql.Driver")
        .load()
    )

    assert result.count() == 2  # Alice + Carol
    segments = [row["segment"] for row in result.collect()]
    assert all(s == "Premium" for s in segments), \
        f"All rows should be Premium, got: {segments}"

# tests/integration/test_kafka_streaming.py
# Integration-тест: Spark Structured Streaming читает из реального Kafka

import pytest
import threading
import time
from testcontainers.kafka import KafkaContainer
from pyspark.sql import functions as F


@pytest.fixture(scope="module")
def kafka_container():
    """Поднимает Kafka контейнер для тестов стриминга."""
    with KafkaContainer(image="confluentinc/cp-kafka:7.5.0") as kafka:
        yield kafka


def test_spark_reads_from_kafka(spark, kafka_container, tmp_path):
    """
    Integration-тест: Spark Structured Streaming читает события из Kafka.

    Схема теста:
    1. Запускаем streaming query в фоне
    2. Продьюсируем несколько событий в Kafka
    3. Ждём обработки (небольшая пауза)
    4. Останавливаем query
    5. Читаем output и проверяем результат

    Это реалистичный тест: реальный Kafka, реальный Spark Streaming,
    но всё на localhost, без staging-кластера.
    """
    bootstrap_servers = kafka_container.get_bootstrap_server()
    topic_name = "test_events"
    output_path = str(tmp_path / "streaming_output")
    checkpoint_path = str(tmp_path / "checkpoint")

    # Streaming query: читаем из Kafka, пишем в tmp_path
    streaming_query = (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", bootstrap_servers)
        .option("subscribe", topic_name)
        .option("startingOffsets", "earliest")
        .load()
        .select(
            F.col("value").cast("string").alias("raw_event"),
            F.from_json(
                F.col("value").cast("string"),
                schema="event_id STRING, event_type STRING, user_id STRING"
            ).alias("data")
        )
        .select("raw_event", "data.*")
        .writeStream
        .format("parquet")
        .option("path", output_path)
        .option("checkpointLocation", checkpoint_path)
        .trigger(processingTime="1 second")
        .start()
    )

    try:
        # Продьюсируем события в Kafka из отдельного потока
        def produce_events():
            from kafka import KafkaProducer
            import json
            producer = KafkaProducer(
                bootstrap_servers=bootstrap_servers,
                value_serializer=lambda v: json.dumps(v).encode()
            )
            for i in range(5):
                producer.send(topic_name, {
                    "event_id": f"EVT-{i:03d}",
                    "event_type": "click",
                    "user_id": f"USER-{i % 3}",
                })
            producer.flush()
            producer.close()

        producer_thread = threading.Thread(target=produce_events)
        producer_thread.start()
        producer_thread.join()

        # Ждём пока Spark обработает события
        time.sleep(5)

    finally:
        streaming_query.stop()

    # Проверяем результат
    result = spark.read.parquet(output_path)
    assert result.count() == 5, f"Expected 5 events, got {result.count()}"

    event_types = result.select("event_type").distinct().collect()
    assert len(event_types) == 1
    assert event_types[0]["event_type"] == "click"

"""
tests/unit/test_streaming_logic.py
Unit-тесты стриминговых трансформаций через MemoryStream.
Не требует Kafka, работает полностью in-memory.
"""

import pytest
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, LongType


def test_dedup_with_watermark_using_memory_stream(spark, tmp_path):
    """
    Тест дедупликации Structured Streaming с Watermark.

    Watermark - механизм, ограничивающий состояние (state) при стриминге.
    dropDuplicates() со Watermark удаляет строки только в пределах временного окна.

    MemoryStream позволяет:
    1. Добавлять данные партиями (addData)
    2. Запускать один micro-batch за раз (processAllAvailable)
    3. Проверять промежуточный и финальный результат

    Это даёт детерминированность - нет случайных задержек, нет гонок.
    """
    from pyspark.sql.streaming import DataStreamWriter

    # Определяем схему входных событий
    schema = StructType([
        StructField("event_id",   StringType(),    nullable=False),
        StructField("user_id",    StringType(),    nullable=True),
        StructField("event_type", StringType(),    nullable=True),
        StructField("event_ts",   TimestampType(), nullable=False),
    ])

    # Создаём MemoryStream - фиктивный стриминговый источник
    from pyspark.sql.streaming.data_source import DataStreamReader

    # Используем более простой подход через createDataFrame + writeStream
    # для имитации микробатчей

    # Первый батч данных (включая дубль event_id = "EVT-001")
    batch1 = spark.createDataFrame(
        [("EVT-001", "USER-01", "click", "2024-01-15 10:00:00"),
         ("EVT-002", "USER-02", "view",  "2024-01-15 10:01:00"),
         ("EVT-001", "USER-01", "click", "2024-01-15 10:00:00")],  # Дубль
        schema="event_id STRING, user_id STRING, event_type STRING, event_ts STRING"
    ).withColumn("event_ts", F.to_timestamp("event_ts"))

    # Дедупликация на static DataFrame (unit-тест трансформации)
    from pipeline.transformations import deduplicate_orders

    # Применяем нашу функцию дедупликации (адаптированную для событий)
    def dedup_events(df):
        from pyspark.sql.window import Window
        window = Window.partitionBy("event_id").orderBy(F.col("event_ts").desc())
        return (
            df
            .withColumn("_rn", F.row_number().over(window))
            .filter(F.col("_rn") == 1)
            .drop("_rn")
        )

    result = dedup_events(batch1)

    # EVT-001 должен встречаться только один раз
    evt001_count = result.filter("event_id = 'EVT-001'").count()
    assert evt001_count == 1, f"EVT-001 should appear once, got {evt001_count}"
    assert result.count() == 2, f"Expected 2 unique events, got {result.count()}"


def test_streaming_aggregation_in_window(spark, tmp_path):
    """
    Тест агрегации в скользящем временном окне.

    Проверяем логику подсчёта событий в 5-минутных окнах.
    Используем статические данные с явными timestamp - детерминированный тест.
    """
    events = spark.createDataFrame(
        [("EVT-001", "USER-01", "2024-01-15 10:00:00"),
         ("EVT-002", "USER-01", "2024-01-15 10:02:00"),  # В том же окне
         ("EVT-003", "USER-02", "2024-01-15 10:04:00"),  # В том же окне
         ("EVT-004", "USER-01", "2024-01-15 10:06:00")], # В следующем окне
        schema="event_id STRING, user_id STRING, event_ts STRING"
    ).withColumn("event_ts", F.to_timestamp("event_ts"))

    # Агрегация в 5-минутных окнах
    result = (
        events
        .groupBy(
            F.window("event_ts", "5 minutes"),
            "user_id"
        )
        .agg(F.count("event_id").alias("event_count"))
    )

    # USER-01 в первом окне (10:00-10:05): 2 события
    user01_window1 = (
        result
        .filter("user_id = 'USER-01'")
        .filter("window.start = '2024-01-15 10:00:00'")
        .select("event_count")
        .collect()
    )
    assert len(user01_window1) == 1
    assert user01_window1[0]["event_count"] == 2

    # USER-01 в следующем окне (10:05-10:10): 1 событие
    user01_window2 = (
        result
        .filter("user_id = 'USER-01'")
        .filter("window.start = '2024-01-15 10:05:00'")
        .select("event_count")
        .collect()
    )
    assert len(user01_window2) == 1
    assert user01_window2[0]["event_count"] == 1

def test_checkpoint_isolation_between_tests(spark, tmp_path):
    """
    Критичный тест: каждый стриминговый тест должен иметь
    собственный checkpoint путь.

    Если два теста используют один checkpoint, второй тест
    продолжит с того места, где остановился первый.
    Это нарушает изоляцию тестов и приводит к ложным результатам.

    tmp_path уникален для каждого теста - используем его для checkpoint.
    """
    checkpoint_path_1 = str(tmp_path / "checkpoint_test_1")
    checkpoint_path_2 = str(tmp_path / "checkpoint_test_2")

    # Проверяем, что пути уникальны
    assert checkpoint_path_1 != checkpoint_path_2

    # В реальном тесте стриминга:
    # query_1 = df.writeStream.option("checkpointLocation", checkpoint_path_1).start()
    # query_2 = df.writeStream.option("checkpointLocation", checkpoint_path_2).start()
    # Каждая query начинает с чистого состояния
    print("Checkpoint paths are isolated:", checkpoint_path_1, checkpoint_path_2)

# tests/conftest.py - конфигурация LocalStack

@pytest.fixture(scope="session")
def localstack_s3():
    """
    Поднимает LocalStack контейнер с S3-эмуляцией.

    Для использования нужно:
    pip install testcontainers localstack-client
    """
    from testcontainers.localstack import LocalStackContainer
    with LocalStackContainer(image="localstack/localstack:3.0") as localstack:
        # Создаём тестовый bucket
        import boto3
        s3 = boto3.client(
            "s3",
            endpoint_url=localstack.get_url(),
            aws_access_key_id="test",
            aws_secret_access_key="test",
            region_name="us-east-1",
        )
        s3.create_bucket(Bucket="test-bronze")
        s3.create_bucket(Bucket="test-silver")
        yield {
            "endpoint": localstack.get_url(),
            "bronze_bucket": "test-bronze",
            "silver_bucket": "test-silver",
        }


@pytest.fixture(scope="session")
def spark_with_s3(localstack_s3):
    """SparkSession, настроенная для работы с LocalStack S3."""
    spark = (
        SparkSession.builder
        .master("local[2]")
        .config("spark.hadoop.fs.s3a.endpoint",
                localstack_s3["endpoint"])
        .config("spark.hadoop.fs.s3a.access.key",      "test")
        .config("spark.hadoop.fs.s3a.secret.key",      "test")
        .config("spark.hadoop.fs.s3a.path.style.access", "true")
        .config("spark.hadoop.fs.s3a.impl",
                "org.apache.hadoop.fs.s3a.S3AFileSystem")
        .getOrCreate()
    )
    yield spark
    spark.stop()


def test_spark_reads_from_localstack_s3(spark_with_s3, localstack_s3):
    """
    Integration-тест: Spark читает из S3-эмулятора.

    Spark думает, что работает с реальным S3.
    На самом деле - с LocalStack. Полная изоляция от AWS.
    """
    import boto3
    import json

    # Загружаем тестовые данные в LocalStack S3
    s3 = boto3.client(
        "s3",
        endpoint_url=localstack_s3["endpoint"],
        aws_access_key_id="test",
        aws_secret_access_key="test",
        region_name="us-east-1",
    )

    # Создаём Parquet файл в памяти и загружаем в S3
    test_data = spark_with_s3.createDataFrame(
        [("ORD-001", "CUST-01", 18.0)],
        schema="order_id STRING, customer_id STRING, order_total DOUBLE"
    )
    # В реальном тесте: записываем напрямую через Spark + S3A

    print(f"LocalStack S3 endpoint: {localstack_s3['endpoint']}")
    # Дальнейший тест: spark.read.parquet("s3a://test-bronze/orders/")

"""
pipeline/config.py
Configuration-driven pipeline design.
"""

from dataclasses import dataclass, field
from typing import Optional


@dataclass
class PipelineConfig:
    """
    Конфигурация пайплайна.

    Все параметры - внешние зависимости, которые могут быть подменены в тестах.
    Production: берёт значения из переменных окружения / Airflow Variables.
    Tests: передаются явно через pytest fixture.
    """
    # Пути к данным
    bronze_path: str
    silver_path: str
    gold_path: str

    # JDBC параметры
    jdbc_url: str = "jdbc:postgresql://prod-db:5432/warehouse"
    jdbc_user: str = "etl_user"
    jdbc_password: str = ""  # В prod: из secrets manager

    # Spark параметры
    num_executors: int = 20
    executor_memory: str = "8g"

    # Бизнес-параметры
    date: str = ""
    lookback_days: int = 7

    @classmethod
    def for_testing(cls, tmp_path: str, date: str = "2024-01-15") -> "PipelineConfig":
        """
        Фабричный метод: создаёт конфиг для тестирования.
        Все пути - локальные tmp директории, не реальные S3/PostgreSQL.
        """
        return cls(
            bronze_path=f"{tmp_path}/bronze",
            silver_path=f"{tmp_path}/silver",
            gold_path=f"{tmp_path}/gold",
            jdbc_url="",    # Не нужен для unit-тестов
            date=date,
        )

    @classmethod
    def from_environment(cls) -> "PipelineConfig":
        """Создаёт production-конфиг из переменных окружения."""
        import os
        return cls(
            bronze_path=os.environ["BRONZE_PATH"],
            silver_path=os.environ["SILVER_PATH"],
            gold_path=os.environ["GOLD_PATH"],
            jdbc_url=os.environ.get("JDBC_URL", ""),
            jdbc_user=os.environ.get("JDBC_USER", ""),
            jdbc_password=os.environ.get("JDBC_PASSWORD", ""),
            date=os.environ.get("PIPELINE_DATE", ""),
        )

# tests/integration/test_configurable_pipeline.py

def test_pipeline_with_test_config(spark, tmp_path):
    """
    Integration-тест через конфигурируемый пайплайн.

    PipelineConfig.for_testing() создаёт изолированную конфигурацию
    без изменения кода пайплайна - только параметры разные.
    """
    from pipeline.config import PipelineConfig
    from pipeline.transformations import clean_raw_orders, calculate_order_total

    config = PipelineConfig.for_testing(
        tmp_path=str(tmp_path),
        date="2024-01-15"
    )

    # Пишем Bronze данные по конфигурируемому пути
    bronze_df = spark.createDataFrame(
        [("ORD-001", "CUST-01", 2, 10.0, 0.1, "2024-01-15 10:00:00"),
         ("ORD-002", "CUST-02", 1, 50.0, 0.0, "2024-01-15 11:00:00")],
        schema="order_id STRING, customer_id STRING, quantity INT, "
               "unit_price DOUBLE, discount DOUBLE, updated_at STRING"
    )
    bronze_df.write.parquet(config.bronze_path)

    # Читаем Bronze по конфигурируемому пути (не хардкоженному)
    orders = spark.read.parquet(config.bronze_path)
    orders_clean = clean_raw_orders(orders)
    orders_total = calculate_order_total(orders_clean)

    # Пишем Silver по конфигурируемому пути
    orders_total.write.mode("overwrite").parquet(config.silver_path)

    # Проверяем Silver
    silver = spark.read.parquet(config.silver_path)
    assert silver.count() == 2

    # В prod: config.bronze_path = s3://prod-bronze/
    # В test:  config.bronze_path = /tmp/pytest-xxx/bronze/
    # Код пайплайна - одинаковый. Только config разный.
    print(f"Pipeline ran with config: bronze={config.bronze_path}")

# АНТИПАТТЕРН 1: assert только на row count
def test_useless_count_assertion(spark):
    """
    Этот тест проходит даже если данные полностью неверные.
    count() = 3 ни о чём не говорит без проверки значений.
    """
    result = some_pipeline(spark)
    assert result.count() == 3  # ← Что в этих 3 строках? Мы не знаем.


# АНТИПАТТЕРН 2: тестирование implementation details
def test_mock_chain(spark):
    """
    Тестируем цепочку вызовов, а не результат.
    Если рефакторим .filter() на .where() → тест падает (ложный negative).
    Если логика в .filter() неверная → тест проходит (false positive).
    """
    mock_df = MagicMock()
    mock_df.filter.return_value = mock_df
    mock_df.withColumn.return_value = mock_df

    clean_raw_orders(mock_df)

    mock_df.filter.assert_called()  # ← Это не тест бизнес-логики


# АНТИПАТТЕРН 3: тест зависит от состояния другого теста
_shared_df = None  # Глобальная переменная - катастрофа для изоляции

def test_step1(spark):
    global _shared_df
    _shared_df = spark.createDataFrame([("A", 1)], schema="key STRING, val INT")

def test_step2(spark):
    global _shared_df
    # Если test_step1 не запустился → _shared_df = None → NullPointerError
    result = transform(_shared_df)  # Зависит от порядка запуска!


# ПРАВИЛЬНО: каждый тест независим, использует свои данные
def test_step1_correct(spark):
    df = spark.createDataFrame([("A", 1)], schema="key STRING, val INT")
    result = transform(df)
    assertDataFrameEqual(result, expected_df, checkRowOrder=False)

# АНТИПАТТЕРН: интеграционный тест против shared staging
def test_pipeline_against_staging():
    """
    Проблемы этого подхода:
    1. Staging данные меняются → тест нестабилен (flaky)
    2. Параллельный запуск тестов → race conditions
    3. Тест оставляет мусор в staging БД
    4. Нужен VPN / network access → не работает в CI без специальной настройки
    5. Дорого: занимает ресурсы staging-кластера
    """
    spark = SparkSession.builder.master("yarn").getOrCreate()  # ← реальный кластер!
    df = spark.table("staging_catalog.silver.orders")           # ← реальные данные!
    result = aggregate_daily_kpi(df)
    assert result.count() > 0  # ← нестабильно, зависит от staging данных


# ПРАВИЛЬНО: testcontainers или local file substitution
def test_pipeline_isolated(spark, tmp_path):
    """Полная изоляция от любой внешней инфраструктуры."""
    # Синтетические данные → детерминированный результат
    df = spark.createDataFrame(
        [("ORD-001", "2024-01-15", "Premium", 18.0),
         ("ORD-002", "2024-01-15", "Standard", 50.0)],
        schema="order_id STRING, order_date STRING, segment STRING, order_total DOUBLE"
    )
    result = aggregate_daily_kpi(df)
    # Точное ожидаемое значение, не "count > 0"
    premium_revenue = (
        result
        .filter("segment = 'Premium'")
        .select("revenue")
        .collect()[0]["revenue"]
    )
    assert abs(premium_revenue - 18.0) < 0.01

orders_pipeline/
├── pipeline/
│   ├── __init__.py
│   ├── config.py                 ← PipelineConfig с for_testing()
│   ├── transformations.py        ← Чистые функции (unit-тесты)
│   └── io/
│       ├── readers.py            ← Параметризованные readers
│       └── writers.py            ← Параметризованные writers
│
├── tests/
│   ├── conftest.py               ← spark, spark_delta fixtures
│   │
│   ├── unit/                     ← Быстро (секунды), нет IO
│   │   ├── test_clean_orders.py
│   │   ├── test_calculate_total.py
│   │   ├── test_enrich.py
│   │   └── test_dedup.py
│   │
│   ├── integration/              ← Умеренно быстро (10-30 сек), локальный IO
│   │   ├── test_local_parquet.py  ← tmp_path substitution
│   │   ├── test_delta_lake.py     ← Delta MERGE/OPTIMIZE локально
│   │   └── test_jdbc_reader.py    ← testcontainers PostgreSQL
│   │
│   ├── streaming/                ← Стриминговые тесты
│   │   └── test_window_aggregation.py
│   │
│   └── schema/                   ← Schema contracts
│       └── test_data_contracts.py
│
├── pyproject.toml
└── pytest.ini

# tests/integration/test_orders_pipeline_e2e.py
"""
End-to-end integration-тест без mocks.
Все внешние зависимости заменены локальными суррогатами.
"""

import pytest
from pathlib import Path
from pyspark.testing import assertDataFrameEqual
from pyspark.sql import functions as F
from pipeline.config import PipelineConfig
from pipeline.transformations import (
    clean_raw_orders, deduplicate_orders,
    enrich_with_customer, calculate_order_total, aggregate_daily_kpi
)


@pytest.fixture
def pipeline_config(tmp_path):
    """Создаёт изолированную конфигурацию для этого теста."""
    return PipelineConfig.for_testing(str(tmp_path), date="2024-01-15")


@pytest.fixture
def bronze_data(spark, pipeline_config):
    """
    Записывает тестовые Bronze данные в локальный tmp_path.
    Имитирует то, что приходит от источника.
    """
    data = spark.createDataFrame(
        [# Нормальные заказы
         ("ORD-001", "CUST-01", 2, 10.0, 0.1, "2024-01-15 10:00:00"),
         ("ORD-002", "CUST-02", 1, 50.0, 0.0, "2024-01-15 11:00:00"),
         ("ORD-003", "CUST-01", 3, 20.0, 0.25, "2024-01-15 12:00:00"),
         # Дубль ORD-001 - более свежая версия (updated_at позже)
         ("ORD-001", "CUST-01", 2, 10.0, 0.1, "2024-01-15 14:00:00"),
         # NULL order_id - невалидная строка, должна быть отфильтрована
         (None, "CUST-03", 5, 15.0, 0.0, "2024-01-15 13:00:00"),
         # qty = 0 - невалидная строка
         ("ORD-004", "CUST-02", 0, 30.0, 0.0, "2024-01-15 15:00:00"),
         ],
        schema="order_id STRING, customer_id STRING, quantity INT, "
               "unit_price DOUBLE, discount DOUBLE, updated_at STRING"
    )
    data.write.parquet(pipeline_config.bronze_path)
    return data


@pytest.fixture
def customers_data(spark, pipeline_config):
    """Тестовые данные клиентов."""
    return spark.createDataFrame(
        [("CUST-01", "Alice Smith",  "Premium"),
         ("CUST-02", "Bob Johnson",  "Standard"),
         # CUST-03 намеренно отсутствует → проверяем LEFT JOIN поведение
         ],
        schema="customer_id STRING, customer_name STRING, segment STRING"
    )


def test_full_pipeline_produces_correct_silver(
    spark, pipeline_config, bronze_data, customers_data
):
    """
    End-to-end тест: Bronze → Silver трансформации корректны.

    Проверяем:
    1. NULL order_id отфильтрован
    2. qty=0 отфильтрован
    3. Дубль ORD-001 дедуплицирован (остаётся одна строка)
    4. order_total вычислен корректно
    5. Обогащение customer_name работает для известных клиентов
    6. Неизвестный клиент (CUST-03) → NULL customer_name (LEFT JOIN)
    """
    orders_raw = spark.read.parquet(pipeline_config.bronze_path)

    # Применяем пайплайн трансформаций
    orders_clean    = clean_raw_orders(orders_raw)
    orders_deduped  = deduplicate_orders(orders_clean)
    orders_enriched = enrich_with_customer(orders_deduped, customers_data)
    orders_final    = calculate_order_total(orders_enriched)

    # Записываем Silver в tmp_path
    orders_final.write.mode("overwrite").parquet(pipeline_config.silver_path)

    # Читаем Silver и проверяем
    silver = spark.read.parquet(pipeline_config.silver_path)

    # 6 исходных строк:
    # -1 NULL order_id
    # -1 qty=0
    # -1 дубль ORD-001 (старая версия)
    # = 3 строки должны остаться
    assert silver.count() == 3, \
        f"Expected 3 rows after cleaning and dedup, got {silver.count()}"

    # Проверяем расчёт для ORD-001: 2 * 10 * (1 - 0.1) = 18.0
    ord1 = silver.filter("order_id = 'ORD-001'").select("order_total").collect()
    assert len(ord1) == 1, "ORD-001 should appear exactly once after dedup"
    assert abs(ord1[0]["order_total"] - 18.0) < 0.01

    # Проверяем обогащение: CUST-01 → Alice Smith
    alice_order = silver.filter("order_id = 'ORD-001'").select("customer_name").collect()
    assert alice_order[0]["customer_name"] == "Alice Smith"

    # Убеждаемся, что NULL order_id не попал в Silver
    assert silver.filter("order_id IS NULL").count() == 0


def test_gold_kpi_aggregation(spark, pipeline_config, bronze_data, customers_data):
    """
    End-to-end тест: Silver → Gold агрегация корректна.
    """
    orders_raw      = spark.read.parquet(pipeline_config.bronze_path)
    orders_clean    = clean_raw_orders(orders_raw)
    orders_deduped  = deduplicate_orders(orders_clean)
    orders_enriched = enrich_with_customer(orders_deduped, customers_data)
    orders_final    = calculate_order_total(orders_enriched)
    orders_with_date = orders_final.withColumn(
        "order_date", F.lit("2024-01-15")
    )

    kpi = aggregate_daily_kpi(orders_with_date)

    # Premium сегмент: ORD-001 (18.0) + ORD-003 (45.0) = 63.0
    # ORD-003: 3 * 20 * (1 - 0.25) = 45.0
    premium_kpi = kpi.filter("segment = 'Premium'").collect()
    assert len(premium_kpi) == 1

    premium_revenue = premium_kpi[0]["revenue"]
    assert abs(premium_revenue - 63.0) < 0.01, \
        f"Premium revenue should be 63.0 (18+45), got {premium_revenue}"

    premium_orders = premium_kpi[0]["order_count"]
    assert premium_orders == 2, \
        f"Premium should have 2 orders (ORD-001 and ORD-003), got {premium_orders}"


def test_pipeline_detects_injected_bug(spark):
    """
    Тест с намеренным дефектом: показывает, как тесты ловят баги.

    БАГ: неправильная формула расчёта total
    Было: qty * price * (1 - discount)
    Стало: qty * price - discount  ← неверно! discount = проценты, не сумма
    """
    def calculate_total_BUGGY(df):
        return df.withColumn(
            "order_total",
            F.round(F.col("quantity") * F.col("unit_price") - F.col("discount"), 2)
        )

    orders = spark.createDataFrame(
        [("ORD-001", 2, 10.0, 0.1)],  # 10% скидка
        schema="order_id STRING, quantity INT, unit_price DOUBLE, discount DOUBLE"
    )

    # Правильный результат: 2 * 10 * (1 - 0.1) = 18.0
    # Багованный результат: 2 * 10 - 0.1 = 19.9

    buggy_result = calculate_total_BUGGY(orders)
    total = buggy_result.select("order_total").collect()[0]["order_total"]

    # Этот assert ПРОВАЛИТСЯ, показывая где баг
    assert abs(total - 18.0) < 0.01, \
        f"ORDER TOTAL BUG DETECTED!\n" \
        f"Expected 18.0 (qty * price * (1 - discount%)),\n" \
        f"got {total}.\n" \
        f"Check calculate_order_total: discount is a RATIO (0.0-1.0), " \
        f"not an absolute amount. Use (1 - discount), not -discount."

"""
Чек-лист: проверяем качество тест-сьюта.
"""

ISOLATION_CHECKLIST = {
    "no_mock_spark": {
        "check": "Нет MagicMock для SparkSession, DataFrame, RDD",
        "verify": "grep -r 'MagicMock\\|@patch' tests/ | grep -i 'spark\\|dataframe'",
    },
    "no_external_io": {
        "check": "Unit-тесты не обращаются к S3, HDFS, PostgreSQL, Kafka",
        "verify": "В unit/ нет read.parquet('s3://...') или .format('jdbc')",
    },
    "tmp_path_for_integration": {
        "check": "Integration-тесты используют tmp_path для IO",
        "verify": "Все write() / read() в integration/ используют tmp_path",
    },
    "deterministic_data": {
        "check": "Тестовые данные явные и детерминированные",
        "verify": "Нет random(), current_timestamp(), uuid() в тестовых данных",
    },
    "isolated_checkpoints": {
        "check": "Каждый стриминговый тест использует свой tmp_path для checkpoint",
        "verify": "Нет shared checkpoint-путей между тестами",
    },
    "configurable_paths": {
        "check": "Пайплайн принимает пути как параметры (не хардкодит)",
        "verify": "Есть PipelineConfig или аналог с явными параметрами пути",
    },
    "testcontainers_not_staging": {
        "check": "JDBC-тесты используют testcontainers, не staging-БД",
        "verify": "Нет JDBC URL типа staging.company.internal в тестах",
    },
    "pure_functions_in_unit": {
        "check": "Unit-тесты тестируют только чистые transform-функции",
        "verify": "В unit/ нет spark.read / spark.write",
    },
}