dbt Tests и Source Freshness: generic tests, singular tests, data contracts

                  Стоимость исправления растёт →→→

[Ingestion]  →  [Bronze]  →  [Silver]  →  [Gold]  →  [Report]
    ↑              ↑             ↑            ↑           ↑
  Дёшево         Дёшево       Дороже      Дорого    Очень дорого

-- Пример: тест unique для колонки order_id
SELECT order_id
FROM my_schema.fct_orders
WHERE order_id IS NOT NULL
GROUP BY order_id
HAVING COUNT(*) > 1

-- Если этот SELECT вернул строки - есть дубликаты → тест упал
-- Если SELECT вернул 0 строк - дублей нет → тест прошёл

# models/staging/sources.yml
version: 2

sources:
  - name: bronze_transactions
    description: >
      Bronze-слой с транзакциями из PostgreSQL через CDC-репликацию.
      Обновляется каждые 15 минут через Debezium + Kafka.
    database: hive_metastore
    schema: bronze

    # Колонка, по которой измеряем свежесть.
    # Это должен быть timestamp момента загрузки в Bronze,
    # а не бизнес-timestamp (created_at транзакции).
    loaded_at_field: _ingested_at

    # SLA по умолчанию для всех таблиц источника
    freshness:
      warn_after: {count: 1, period: hour}
      error_after: {count: 3, period: hours}

    tables:
      - name: raw_orders
        description: Сырые заказы из PostgreSQL orders table
        # Переопределяем freshness для этой конкретной таблицы
        freshness:
          warn_after: {count: 30, period: minutes}
          error_after: {count: 2, period: hours}

      - name: raw_order_items
        description: Позиции заказов - обновляется вместе с raw_orders
        # Наследует freshness от источника: warn=1h, error=3h

      - name: raw_users
        description: Профили пользователей - медленно меняются
        # Для справочников допустима большая задержка
        freshness:
          warn_after: {count: 6, period: hours}
          error_after: {count: 24, period: hours}

      - name: raw_product_catalog
        description: Каталог товаров
        # Полностью отключаем проверку freshness
        freshness: null

-- Что dbt генерирует для raw_orders:
SELECT
    MAX(_ingested_at) AS max_loaded_at,
    CURRENT_TIMESTAMP() AS snapshotted_at
FROM hive_metastore.bronze.raw_orders

Текущее время: 2024-01-15 14:30:00
MAX(_ingested_at): 2024-01-15 12:45:00
Разница: 1 час 45 минут

warn_after: 30 минут  → WARN (превышено)
error_after: 2 часа   → OK (не превышено)
→ Статус: WARN

# Запустить freshness-проверки для всех источников
dbt source freshness

# Только для конкретного источника
dbt source freshness --select source:bronze_transactions

# Вывод в файл для мониторинга
dbt source freshness --output sources.json

Running with dbt=1.7.0
Found 3 sources, 0 freshness errors

14:30:01  Freshness of bronze_transactions.raw_orders ........ [WARN in 0.84s]
14:30:02  Freshness of bronze_transactions.raw_order_items ... [pass in 0.71s]
14:30:03  Freshness of bronze_transactions.raw_users ......... [pass in 0.62s]

Done. PASS=2 WARN=1 ERROR=0 SKIP=0 TOTAL=3

tables:
  - name: raw_events
    loaded_at_field: event_date  # Поле партиционирования
    freshness:
      warn_after: {count: 6, period: hours}

-- dbt генерирует:
SELECT MAX(event_date) AS max_loaded_at, CURRENT_TIMESTAMP() AS snapshotted_at
FROM bronze.raw_events

# Airflow DAG
from airflow.operators.bash import BashOperator

check_freshness = BashOperator(
    task_id='check_source_freshness',
    bash_command='cd /opt/dbt && dbt source freshness --select source:bronze_transactions',
)

run_transformations = BashOperator(
    task_id='run_dbt_transformations',
    bash_command='cd /opt/dbt && dbt run --select marts.*',
)

# Трансформации запустятся только если freshness прошёл
check_freshness >> run_transformations

# models/marts/schema.yml
version: 2

models:
  - name: fct_orders
    description: >
      Финансовая витрина заказов. Одна строка - один заказ.
      Обновляется инкрементально каждый час.

    config:
      # Строгий режим: тест-ошибка блокирует downstream модели
      severity: error

    columns:
      - name: order_id
        description: Уникальный идентификатор заказа из PostgreSQL
        tests:
          - not_null
          - unique

      - name: user_id
        description: Ссылка на пользователя
        tests:
          - not_null
          - relationships:
              to: ref('dim_users')
              field: user_id

      - name: status
        description: Статус заказа
        tests:
          - not_null
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'refunded']

      - name: total_amount
        description: Итоговая сумма заказа в рублях
        tests:
          - not_null

      - name: event_date
        description: Дата создания заказа (partition key)
        tests:
          - not_null

-- target/compiled/.../not_null_fct_orders_order_id.sql
SELECT order_id
FROM my_schema.fct_orders
WHERE order_id IS NULL

-- Генерируемый SQL:
SELECT order_id
FROM my_schema.fct_orders
WHERE order_id IS NOT NULL
GROUP BY order_id
HAVING COUNT(*) > 1

- name: status
  tests:
    - accepted_values:
        values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'refunded']
        # По умолчанию quote=true (значения в кавычках - строки)
        # Для числовых значений:
        # quote: false
        # values: [1, 2, 3, 4]

-- Генерируемый SQL:
SELECT DISTINCT status
FROM my_schema.fct_orders
WHERE status NOT IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'refunded')
  AND status IS NOT NULL

- name: user_id
  tests:
    - relationships:
        to: ref('dim_users')
        field: user_id

-- Генерируемый SQL:
SELECT order_id
FROM my_schema.fct_orders
WHERE user_id IS NOT NULL
  AND user_id NOT IN (
      SELECT user_id FROM my_schema.dim_users WHERE user_id IS NOT NULL
  )

- name: user_id
  tests:
    - relationships:
        to: ref('dim_users')
        field: user_id
        severity: warn   # Предупреждение, не ошибка

- name: total_amount
  tests:
    - not_null:
        where: "status != 'cancelled'"
        # NULL в total_amount допустим только для отменённых заказов

- name: order_id
  tests:
    - unique:
        where: "event_date >= current_date() - interval 7 days"
        # Уникальность проверяем только за последние 7 дней
        # Для исторических данных уникальность уже была проверена

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique:
              store_failures: true
              # Создаст таблицу: my_schema_dbt_test__audit.unique_fct_orders_order_id

-- Смотрим дубликаты напрямую:
SELECT * FROM my_schema_dbt_test__audit.unique_fct_orders_order_id
LIMIT 100;

models:
  - name: fct_order_items
    description: Позиции заказов. Уникальность - комбинация order_id + item_id.
    tests:
      # Тест на уровне модели (не колонки)
      - unique:
          column_name: "CONCAT(order_id, '|', item_id)"
          # Или через dbt-utils (подробнее ниже)

    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns:
            - order_id
            - item_id

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1

dbt deps  # Установить пакеты

models:
  - name: fct_orders
    tests:
      # Уникальность составного ключа
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [order_id, item_id]

      # Колонка возрастает монотонно (для суррогатных ключей)
      - dbt_utils.monotonically_increasing_id:
          column_name: surrogate_key

    columns:
      - name: total_amount
        tests:
          # Произвольное условие SQL (самый гибкий generic test)
          - dbt_utils.expression_is_true:
              expression: ">= 0"
              # Проверяет: total_amount >= 0

          # Значение в диапазоне
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 10000000  # Максимальный заказ 10 млн руб

      - name: created_at
        tests:
          # Дата не в будущем
          - dbt_utils.expression_is_true:
              expression: "<= current_timestamp()"

-- tests/assert_discount_not_exceeds_total.sql
-- Тест: скидка не может превышать полную стоимость позиции

SELECT
    order_id,
    item_id,
    total_amount,
    discount_amount,
    (discount_amount - total_amount) AS overcharge
FROM {{ ref('fct_order_items') }}
WHERE discount_amount > total_amount

dbt test --select test_type:singular
dbt test --select assert_discount_not_exceeds_total

-- tests/assert_no_future_orders.sql
-- Заказы не могут быть созданы в будущем (с допуском в 1 час на clock skew)

{% set tolerance_hours = var('clock_skew_tolerance_hours', 1) %}

SELECT
    order_id,
    created_at,
    CURRENT_TIMESTAMP() AS check_time,
    TIMESTAMPDIFF(HOUR, CURRENT_TIMESTAMP(), created_at) AS hours_in_future
FROM {{ ref('fct_orders') }}
WHERE created_at > CURRENT_TIMESTAMP() + INTERVAL {{ tolerance_hours }} HOURS

dbt test --vars '{"clock_skew_tolerance_hours": 2}'

-- tests/assert_event_ordering.sql
-- Событие delivered не может предшествовать событию confirmed

SELECT
    order_id,
    confirmed_at,
    delivered_at,
    TIMESTAMPDIFF(MINUTE, confirmed_at, delivered_at) AS minutes_diff
FROM {{ ref('fct_orders') }}
WHERE status = 'delivered'
  AND delivered_at < confirmed_at

-- tests/assert_transaction_balance_reconciliation.sql
-- Сверка: изменение баланса пользователя должно совпадать
-- с суммой транзакций за тот же период

WITH balance_changes AS (
    SELECT
        user_id,
        SUM(amount) AS total_transactions
    FROM {{ ref('fct_transactions') }}
    WHERE event_date >= CURRENT_DATE() - INTERVAL 1 DAY
    GROUP BY user_id
),

balance_movements AS (
    SELECT
        user_id,
        SUM(CASE WHEN event_type = 'credit' THEN amount ELSE -amount END) AS net_movement
    FROM {{ ref('fct_balance_movements') }}
    WHERE event_date >= CURRENT_DATE() - INTERVAL 1 DAY
    GROUP BY user_id
),

reconciliation AS (
    SELECT
        t.user_id,
        t.total_transactions,
        b.net_movement,
        ABS(t.total_transactions - b.net_movement) AS discrepancy
    FROM balance_changes t
    JOIN balance_movements b ON t.user_id = b.user_id
)

SELECT *
FROM reconciliation
WHERE discrepancy > 0.01  -- Допуск 1 копейка на ошибки округления

-- tests/assert_no_gaps_in_daily_data.sql
-- Проверка: нет пропущенных дат в ежедневной таблице

WITH date_spine AS (
    -- Генерируем все даты от первой записи до вчера
    SELECT explode(sequence(
        (SELECT MIN(event_date) FROM {{ ref('fct_orders') }}),
        CURRENT_DATE() - INTERVAL 1 DAY,
        INTERVAL 1 DAY
    )) AS expected_date
),

actual_dates AS (
    SELECT DISTINCT event_date AS actual_date
    FROM {{ ref('fct_orders') }}
)

SELECT expected_date AS missing_date
FROM date_spine
LEFT JOIN actual_dates ON expected_date = actual_date
WHERE actual_date IS NULL

-- tests/assert_cdc_no_duplicate_events.sql
-- В CDC-потоке не должно быть дублей одного события

SELECT
    event_id,
    COUNT(*) AS occurrence_count
FROM {{ ref('stg_orders_cdc') }}
GROUP BY event_id
HAVING COUNT(*) > 1

-- Вариант 1: ограничить проверку свежими данными
SELECT *
FROM {{ ref('fct_orders') }}
WHERE event_date >= CURRENT_DATE() - INTERVAL 7 DAYS
  AND discount_amount > total_amount

-- Вариант 2: использовать TABLESAMPLE для приблизительной проверки
-- (не гарантирует 100% покрытие, но быстро)
SELECT *
FROM {{ ref('fct_orders') }} TABLESAMPLE (1 PERCENT)
WHERE total_amount < 0

models:
  - name: fct_orders
    description: Gold-витрина заказов с Data Contract

    # Включаем строгую проверку контракта
    config:
      contract:
        enforced: true

    columns:
      - name: order_id
        description: UUID заказа
        data_type: string
        constraints:
          - type: not_null

      - name: user_id
        description: UUID пользователя
        data_type: string
        constraints:
          - type: not_null

      - name: total_amount
        description: Итоговая сумма заказа в рублях
        data_type: decimal(18, 2)
        constraints:
          - type: not_null

      - name: status
        description: Статус заказа
        data_type: string
        constraints:
          - type: not_null

      - name: created_at
        description: Время создания заказа UTC
        data_type: timestamp
        constraints:
          - type: not_null

      - name: event_date
        description: Дата создания (partition key)
        data_type: date
        constraints:
          - type: not_null

dbt run --select fct_orders

Compilation Error in model fct_orders
  This model has an enforced contract that failed.
  Column "total_amount": data type mismatch (expected decimal(18,2), got double)

  Contract enforcement requires explicit casting in your SELECT.
  Update the SELECT to: CAST(total_amount AS DECIMAL(18, 2)) AS total_amount

-- models/marts/fct_orders.sql
{{ config(
    materialized='incremental',
    file_format='delta',
    incremental_strategy='merge',
    unique_key='order_id',
    contract={'enforced': true}
) }}

SELECT
    CAST(o.order_id     AS STRING)          AS order_id,
    CAST(o.user_id      AS STRING)          AS user_id,
    CAST(o.total_amount AS DECIMAL(18, 2))  AS total_amount,
    CAST(o.status       AS STRING)          AS status,
    CAST(o.created_at   AS TIMESTAMP)       AS created_at,
    CAST(date(o.created_at) AS DATE)        AS event_date

FROM {{ ref('int_orders_enriched') }} o
{% if is_incremental() %}
WHERE o.updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

columns:
  - name: order_id
    data_type: string
    constraints:
      - type: not_null        # Обязательное поле

  - name: total_amount
    data_type: decimal(18, 2)
    constraints:
      - type: not_null
      - type: check            # Произвольное SQL-условие
        expression: "total_amount >= 0"
        name: chk_positive_amount

  - name: status
    data_type: string
    constraints:
      - type: not_null

-- Что dbt генерирует для Delta:
ALTER TABLE my_schema.fct_orders
ADD CONSTRAINT chk_positive_amount CHECK (total_amount >= 0)

models:
  - name: fct_orders
    latest_version: 2
    versions:
      - v: 1
        defined_in: fct_orders_v1
        # Старый контракт - deprecated, но ещё в production
      - v: 2
        # Текущий контракт
        columns:
          - name: order_id
            data_type: string
          # + новые колонки

# models/staging/sources.yml (Bronze тесты)
sources:
  - name: bronze
    tables:
      - name: raw_orders
        columns:
          - name: _id          # Внутренний ID CDC-события
            tests:
              - not_null        # Единственный hard-fail тест
          - name: order_id
            tests:
              - not_null:
                  severity: warn  # Только предупреждение

# models/staging/schema.yml (Silver тесты)
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - not_null
          - unique            # Строго: дубли недопустимы после очистки

      - name: user_id
        tests:
          - not_null
          - relationships:
              to: ref('stg_users')
              field: user_id
              severity: warn  # warn: CDC может опережать

      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'refunded']

# models/marts/schema.yml (Gold тесты)
models:
  - name: fct_orders
    config:
      contract:
        enforced: true
    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [order_id]
    columns:
      - name: order_id
        data_type: string
        constraints:
          - type: not_null
        tests:
          - not_null
          - unique

      - name: total_amount
        data_type: decimal(18, 2)
        constraints:
          - type: not_null
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

dbt run    # Только выполнение моделей, тесты не запускаются
dbt test   # Только тесты, модели не пересчитываются
dbt build  # Модели + тесты в порядке DAG, с зависимостями

Source Freshness
  → stg_orders (model)
    → not_null_stg_orders_order_id (test)
    → unique_stg_orders_order_id (test)
      → int_orders_enriched (model)           ← ждёт прохождения тестов stg_orders
        → fct_orders (model)
          → assert_discount_not_exceeds_total (test)
          → not_null_fct_orders_order_id (test)

# .github/workflows/dbt_ci.yml
name: dbt CI

on:
  pull_request:
    branches: [main]

jobs:
  dbt-build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install dbt
        run: pip install dbt-spark[PyHive]==1.7.0

      - name: Check source freshness
        run: dbt source freshness
        env:
          DBT_TARGET: ci
          SPARK_HOST: ${{ secrets.SPARK_HOST }}
          SPARK_TOKEN: ${{ secrets.SPARK_TOKEN }}

      - name: Run dbt build (models + tests)
        run: dbt build --select state:modified+
        # state:modified+ = изменённые модели + все downstream зависимости
        env:
          DBT_TARGET: ci
          SPARK_HOST: ${{ secrets.SPARK_HOST }}
          SPARK_TOKEN: ${{ secrets.SPARK_TOKEN }}

      - name: Upload test results
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: dbt-test-results
          path: target/run_results.json

models/
├── staging/
│   ├── sources.yml          ← источники + freshness
│   └── stg_transactions.sql
├── intermediate/
│   └── int_transactions_enriched.sql
└── marts/
    ├── schema.yml           ← data contracts + generic tests
    ├── fct_transactions.sql
    └── dim_users.sql

tests/
├── assert_no_negative_amounts.sql
├── assert_discount_valid.sql
└── assert_transaction_balance_reconciliation.sql

# models/staging/sources.yml
version: 2

sources:
  - name: bronze_transactions
    description: CDC-репликация из PostgreSQL, обновляется каждые 15 минут
    database: hive_metastore
    schema: bronze
    loaded_at_field: _ingested_at

    freshness:
      warn_after: {count: 30, period: minutes}
      error_after: {count: 2, period: hours}

    tables:
      - name: raw_transactions
        description: Финансовые транзакции из payments-сервиса
        columns:
          - name: _ingested_at
            description: Timestamp загрузки в Bronze (UTC)
          - name: transaction_id
            description: UUID транзакции
          - name: user_id
            description: UUID пользователя
          - name: amount
            description: Сумма транзакции (может быть отрицательной для возвратов)
          - name: currency
            description: Валюта (ISO 4217)
          - name: status
            description: Статус транзакции

      - name: raw_users
        description: Профили пользователей - обновляются редко
        freshness:
          warn_after: {count: 6, period: hours}
          error_after: {count: 24, period: hours}

dbt source freshness

# Ожидаемый вывод при нормальной работе:
# 14:30:01  Freshness of bronze_transactions.raw_transactions .. [pass in 0.84s]
# 14:30:02  Freshness of bronze_transactions.raw_users ......... [pass in 0.62s]

# Симулируем задержку: обновляем _ingested_at на прошлое
# (в реальности просто ждём или останавливаем пайплайн)
dbt source freshness
# Freshness of bronze_transactions.raw_transactions .. [WARN in 0.91s]
# Elapsed time since last record: 41 minutes (warn threshold: 30 minutes)

# models/staging/schema.yml
version: 2

models:
  - name: stg_transactions
    description: Очищенные транзакции из Bronze. Одна строка - одна транзакция.

    columns:
      - name: transaction_id
        description: UUID транзакции - первичный ключ
        tests:
          - not_null
          - unique

      - name: user_id
        description: UUID пользователя
        tests:
          - not_null
          - relationships:
              to: ref('stg_users')
              field: user_id
              severity: warn  # CDC может опережать users-таблицу

      - name: amount
        description: Сумма транзакции в копейках (целое число)
        tests:
          - not_null

      - name: currency
        description: Валюта по ISO 4217
        tests:
          - not_null
          - accepted_values:
              values: ['RUB', 'USD', 'EUR', 'CNY']

      - name: status
        description: Статус транзакции
        tests:
          - not_null
          - accepted_values:
              values: ['pending', 'completed', 'failed', 'refunded', 'disputed']

      - name: created_at
        description: Время создания транзакции UTC
        tests:
          - not_null

-- tests/assert_no_negative_completed_amounts.sql
-- Завершённые транзакции не могут иметь нулевую или отрицательную сумму.
-- Отрицательные суммы допустимы только для статуса 'refunded'.

SELECT
    transaction_id,
    user_id,
    amount,
    status,
    created_at
FROM {{ ref('stg_transactions') }}
WHERE status = 'completed'
  AND amount <= 0

-- tests/assert_refund_not_exceeds_original.sql
-- Сумма возврата не может превышать сумму оригинальной транзакции.
-- Ищем пользователей, чьи возвраты за день больше пополнений.

WITH daily_by_user AS (
    SELECT
        user_id,
        event_date,
        SUM(CASE WHEN status = 'completed' THEN amount ELSE 0 END) AS total_credits,
        SUM(CASE WHEN status = 'refunded'  THEN ABS(amount) ELSE 0 END) AS total_refunds
    FROM {{ ref('fct_transactions') }}
    GROUP BY user_id, event_date
)

SELECT
    user_id,
    event_date,
    total_credits,
    total_refunds,
    (total_refunds - total_credits) AS overrefund
FROM daily_by_user
WHERE total_refunds > total_credits

# models/marts/schema.yml
version: 2

models:
  - name: fct_transactions
    description: >
      Gold-витрина финансовых транзакций.
      Строгий Data Contract - изменения схемы требуют согласования.

    config:
      contract:
        enforced: true

    columns:
      - name: transaction_id
        description: UUID транзакции
        data_type: string
        constraints:
          - type: not_null
        tests:
          - not_null
          - unique

      - name: user_id
        description: UUID пользователя
        data_type: string
        constraints:
          - type: not_null
        tests:
          - not_null

      - name: amount_rub
        description: Сумма транзакции в рублях (с копейками)
        data_type: decimal(18, 2)
        constraints:
          - type: not_null
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: -1000000   # Максимальный возврат 1 млн руб
              max_value: 10000000   # Максимальная транзакция 10 млн руб

      - name: status
        description: Финальный статус транзакции
        data_type: string
        constraints:
          - type: not_null
        tests:
          - not_null
          - accepted_values:
              values: ['completed', 'refunded', 'failed', 'disputed']

      - name: created_at
        description: Время создания транзакции UTC
        data_type: timestamp
        constraints:
          - type: not_null
        tests:
          - not_null

      - name: event_date
        description: Дата создания (partition key)
        data_type: date
        constraints:
          - type: not_null
        tests:
          - not_null

-- models/marts/fct_transactions.sql
{{ config(
    materialized='incremental',
    file_format='delta',
    incremental_strategy='merge',
    unique_key='transaction_id',
    partition_by={'field': 'event_date', 'data_type': 'date'},
    contract={'enforced': true}
) }}

SELECT
    CAST(t.transaction_id           AS STRING)          AS transaction_id,
    CAST(t.user_id                  AS STRING)          AS user_id,
    CAST(t.amount / 100.0           AS DECIMAL(18, 2))  AS amount_rub,
    CAST(t.status                   AS STRING)          AS status,
    CAST(t.created_at               AS TIMESTAMP)       AS created_at,
    CAST(date(t.created_at)         AS DATE)            AS event_date,
    CAST(current_timestamp()        AS TIMESTAMP)       AS _loaded_at

FROM {{ ref('int_transactions_enriched') }} t

{% if is_incremental() %}
WHERE t.updated_at > (SELECT MAX(_loaded_at) FROM {{ this }})
{% endif %}

# 1. Проверить свежесть источников
dbt source freshness

# 2. Запустить все модели с тестами (в порядке DAG)
dbt build

# Ожидаемый вывод:
# 14:30:01  Running 8 nodes in thread pool with 4 threads
# 14:30:02  1 of 8 START source freshness check ... [RUN]
# 14:30:02  1 of 8 PASS source freshness check ... [pass in 0.84s]
# 14:30:03  2 of 8 START model stg_transactions ... [RUN]
# 14:30:15  2 of 8 OK created model stg_transactions ... [OK in 12.4s]
# 14:30:15  3 of 8 START test not_null_stg_transactions_transaction_id ... [RUN]
# 14:30:16  3 of 8 PASS test not_null_stg_transactions_transaction_id ... [PASS in 0.91s]
# 14:30:16  4 of 8 START test unique_stg_transactions_transaction_id ... [RUN]
# 14:30:17  4 of 8 PASS test unique_stg_transactions_transaction_id ... [PASS in 1.12s]
# ...
# 14:31:45  8 of 8 START test assert_refund_not_exceeds_original ... [RUN]
# 14:31:50  8 of 8 PASS test assert_refund_not_exceeds_original ... [PASS in 4.91s]
#
# Finished running 3 models, 5 tests in 1 minutes 50.34 seconds.
# PASS=8 WARN=0 ERROR=0 SKIP=0 TOTAL=8

# 3. Посмотреть скомпилированный SQL тестов
ls target/compiled/my_project/tests/
cat target/compiled/my_project/tests/assert_refund_not_exceeds_original.sql

14:31:50  3 of 8 FAIL 2 unique_fct_transactions_transaction_id ... [FAIL 2 in 1.12s]

# Включить store_failures и перезапустить
dbt test --select fct_transactions --store-failures

# Посмотреть нарушителей
dbt run-operation run_query --args '{query: "SELECT * FROM dbt_test__audit.unique_fct_transactions_transaction_id LIMIT 10"}'

# 1. Используйте where для ограничения объёма
- name: order_id
  tests:
    - unique:
        where: "event_date >= current_date() - interval 7 days"

# 2. Настройте severity правильно:
# error - только для критичных бизнес-инвариантов (финансы, ключи)
# warn  - для мягких проверок (relationships при асинхронной загрузке)

# 3. Разделяйте тесты на быстрые и медленные:
# dbt test --select "tag:fast"
# dbt test --select "tag:slow"  (запускать реже)

# Тегирование тестов
models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - not_null:
              meta:
                tags: ['fast', 'critical']
          - unique:
              meta:
                tags: ['fast', 'critical']

tests:
  - name: assert_transaction_balance_reconciliation
    meta:
      tags: ['slow', 'financial']