Terraform: Стратегии тестирования и валидации - Полное руководство для DevOps

# Это невинное на вид изменение уничтожило production
resource "aws_s3_bucket" "data" {
  bucket = "company-production-data"
  # Разработчик думал, что просто добавляет версионирование...
  force_destroy = true  # ⚠️ ОПАСНОСТЬ: Удаляет все объекты при destroy!
}

# Базовая проверка синтаксиса и внутренней согласованности
terraform validate

# Пример вывода для ошибок:
# Error: Unsupported argument
#   on main.tf line 12:
#   12:   instance_types = "t2.micro"
# An argument named "instance_types" is not expected here.

# Установка tflint
curl -s https://raw.githubusercontent.com/terraform-linters/tflint/master/install_linux.sh | bash

# Создание конфигурации .tflint.hcl
cat > .tflint.hcl <<EOF
plugin "aws" {
  enabled = true
  version = "0.27.0"
  source  = "github.com/terraform-linters/tflint-ruleset-aws"
}

rule "terraform_deprecated_interpolation" {
  enabled = true
}

rule "terraform_unused_declarations" {
  enabled = true
}

rule "terraform_naming_convention" {
  enabled = true
  format  = "snake_case"
}

rule "aws_instance_invalid_type" {
  enabled = true
}
EOF

# Запуск tflint
tflint --init
tflint

# Установка checkov
pip3 install checkov

# Сканирование Terraform файлов
checkov -d . --framework terraform

# Запуск конкретных проверок
checkov -d . --check CKV_AWS_8  # Проверка шифрования EBS

# Вывод в JSON для интеграции с CI/CD
checkov -d . -o json > security-report.json

# До - нарушения безопасности
resource "aws_s3_bucket" "logs" {
  bucket = "company-logs"
  # Отсутствует: версионирование, шифрование, блокировка публичного доступа
}

# После - соответствие требованиям безопасности
resource "aws_s3_bucket" "logs" {
  bucket = "company-logs"
}

resource "aws_s3_bucket_versioning" "logs" {
  bucket = aws_s3_bucket.logs.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "logs" {
  bucket = aws_s3_bucket.logs.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = "AES256"
    }
  }
}

resource "aws_s3_bucket_public_access_block" "logs" {
  bucket = aws_s3_bucket.logs.id

  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# Генерация плана и сохранение в файл
terraform plan -out=tfplan

# Конвертация бинарного плана в JSON для анализа
terraform show -json tfplan > tfplan.json

# Анализ плана с помощью jq
cat tfplan.json | jq -r '
  .resource_changes[] |
  select(.change.actions[] | contains("delete")) |
  "⚠️  DELETE: \(.address)"
'

# Пример вывода:
# ⚠️  DELETE: aws_instance.old_server
# ⚠️  DELETE: aws_security_group.deprecated

# validate_plan.py - Предотвращение опасных изменений
import json
import sys

def validate_terraform_plan(plan_file):
    """Валидация Terraform плана на опасные операции"""
    with open(plan_file) as f:
        plan = json.load(f)

    errors = []
    warnings = []

    for change in plan.get('resource_changes', []):
        address = change['address']
        actions = change['change']['actions']

        # Проверка удаления критических ресурсов
        if 'delete' in actions:
            if 'database' in address or 'rds' in address:
                errors.append(f"🚨 БЛОКИРОВАНО: Попытка удаления базы данных: {address}")
            elif 's3_bucket' in address and 'backup' in address:
                errors.append(f"🚨 БЛОКИРОВАНО: Попытка удаления бэкап-бакета: {address}")
            else:
                warnings.append(f"⚠️  Предупреждение: Удаление ресурса: {address}")

        # Проверка пересоздания (replace)
        if 'delete' in actions and 'create' in actions:
            if 'aws_instance' in address:
                warnings.append(f"⚠️  Инстанс будет пересоздан: {address}")

        # Проверка изменений правил security group
        if 'aws_security_group' in address or 'aws_security_group_rule' in address:
            if change['change'].get('after', {}).get('ingress'):
                for rule in change['change']['after']['ingress']:
                    if rule.get('cidr_blocks') == ['0.0.0.0/0']:
                        errors.append(f"🚨 БЛОКИРОВАНО: Security group разрешает публичный доступ: {address}")

    # Вывод результатов
    if errors:
        print("\n❌ ВАЛИДАЦИЯ НЕ ПРОЙДЕНА - Обнаружены критические проблемы:\n")
        for error in errors:
            print(error)
        return False

    if warnings:
        print("\n⚠️  Предупреждения (проверьте перед применением):\n")
        for warning in warnings:
            print(warning)

    print("\n✅ Валидация плана пройдена")
    return True

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Использование: python validate_plan.py tfplan.json")
        sys.exit(1)

    success = validate_terraform_plan(sys.argv[1])
    sys.exit(0 if success else 1)

# .github/workflows/terraform.yml
name: Terraform Validation
on: [pull_request]

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup Terraform
        uses: hashicorp/setup-terraform@v2

      - name: Terraform Init
        run: terraform init

      - name: Terraform Validate
        run: terraform validate

      - name: TFLint
        uses: terraform-linters/setup-tflint@v3
        run: |
          tflint --init
          tflint

      - name: Checkov Security Scan
        uses: bridgecrewio/checkov-action@master
        with:
          directory: .
          framework: terraform

      - name: Terraform Plan
        run: |
          terraform plan -out=tfplan
          terraform show -json tfplan > tfplan.json

      - name: Validate Plan
        run: python3 validate_plan.py tfplan.json

// test/terraform_aws_example_test.go
package test

import (
    "testing"
    "time"

    "github.com/gruntwork-io/terratest/modules/aws"
    "github.com/gruntwork-io/terratest/modules/http-helper"
    "github.com/gruntwork-io/terratest/modules/terraform"
    "github.com/stretchr/testify/assert"
)

func TestTerraformWebServer(t *testing.T) {
    t.Parallel()

    // Настройка опций terraform
    terraformOptions := terraform.WithDefaultRetryableErrors(t, &terraform.Options{
        TerraformDir: "../examples/web-server",

        Vars: map[string]interface{}{
            "instance_type": "t2.micro",
            "environment":   "test",
        },

        EnvVars: map[string]string{
            "AWS_DEFAULT_REGION": "us-east-1",
        },
    })

    // Очистка ресурсов в конце
    defer terraform.Destroy(t, terraformOptions)

    // Развертывание инфраструктуры
    terraform.InitAndApply(t, terraformOptions)

    // Валидация выходных данных
    instanceID := terraform.Output(t, terraformOptions, "instance_id")
    publicIP := terraform.Output(t, terraformOptions, "public_ip")

    // Проверка существования и работы инстанса
    instance := aws.GetEc2Instance(t, instanceID, "us-east-1")
    assert.Equal(t, "running", instance.State.Name)
    assert.Equal(t, "t2.micro", instance.InstanceType)

    // Проверка отклика веб-сервера
    url := "http://" + publicIP + ":8080"
    http_helper.HttpGetWithRetry(
        t,
        url,
        nil,
        200,
        "Hello, World",
        30,
        3*time.Second,
    )
}

// test/terraform_module_test.go
func TestVPCModule(t *testing.T) {
    t.Parallel()

    terraformOptions := &terraform.Options{
        TerraformDir: "../modules/vpc",
        Vars: map[string]interface{}{
            "vpc_cidr": "10.0.0.0/16",
            "azs":      []string{"us-east-1a", "us-east-1b"},
        },
    }

    defer terraform.Destroy(t, terraformOptions)
    terraform.InitAndApply(t, terraformOptions)

    // Валидация создания VPC
    vpcID := terraform.Output(t, terraformOptions, "vpc_id")
    vpc := aws.GetVpcById(t, vpcID, "us-east-1")

    assert.Equal(t, "10.0.0.0/16", vpc.CidrBlock)
    assert.True(t, vpc.EnableDnsHostnames)
    assert.True(t, vpc.EnableDnsSupport)

    // Валидация подсетей
    publicSubnetIDs := terraform.OutputList(t, terraformOptions, "public_subnet_ids")
    assert.Equal(t, 2, len(publicSubnetIDs))

    for _, subnetID := range publicSubnetIDs {
        subnet := aws.GetSubnetById(t, subnetID, "us-east-1")
        assert.True(t, subnet.MapPublicIpOnLaunch)
    }
}

// test/terraform_disaster_recovery_test.go
func TestDatabaseFailover(t *testing.T) {
    terraformOptions := &terraform.Options{
        TerraformDir: "../examples/rds-multi-az",
    }

    defer terraform.Destroy(t, terraformOptions)
    terraform.InitAndApply(t, terraformOptions)

    dbEndpoint := terraform.Output(t, terraformOptions, "db_endpoint")
    dbInstanceID := terraform.Output(t, terraformOptions, "db_instance_id")

    // Проверка доступности базы данных
    err := testDatabaseConnection(dbEndpoint, "admin", "password123")
    assert.NoError(t, err)

    // Симуляция failover
    aws.RebootRdsInstance(t, dbInstanceID, "us-east-1")

    // Ожидание завершения failover
    maxRetries := 10
    timeBetweenRetries := 30 * time.Second

    for i := 0; i < maxRetries; i++ {
        err = testDatabaseConnection(dbEndpoint, "admin", "password123")
        if err == nil {
            t.Logf("База данных восстановлена после %d попыток", i+1)
            return
        }
        time.Sleep(timeBetweenRetries)
    }

    t.Fatal("База данных не восстановилась после failover")
}

// Тестирование нескольких сценариев
func TestAWSVPCModule(t *testing.T) {
    testCases := []struct {
        name     string
        vars     map[string]interface{}
        validate func(*testing.T, *terraform.Options)
    }{
        {
            name: "SingleNAT",
            vars: map[string]interface{}{
                "enable_nat_gateway":   true,
                "single_nat_gateway":   true,
            },
            validate: validateSingleNAT,
        },
        {
            name: "MultiNATHighAvailability",
            vars: map[string]interface{}{
                "enable_nat_gateway":   true,
                "single_nat_gateway":   false,
                "one_nat_gateway_per_az": true,
            },
            validate: validateMultiNAT,
        },
    }

    for _, tc := range testCases {
        tc := tc // Захват переменной range
        t.Run(tc.name, func(t *testing.T) {
            t.Parallel()
            // Логика теста здесь
        })
    }
}

// test/state_management_test.go
func TestStateConsistency(t *testing.T) {
    terraformOptions := &terraform.Options{
        TerraformDir: "../infrastructure",
        BackendConfig: map[string]interface{}{
            "bucket": "spotify-terraform-state-test",
            "key":    fmt.Sprintf("test-%d/terraform.tfstate", time.Now().Unix()),
            "region": "us-east-1",
        },
    }

    // Применение инфраструктуры
    terraform.InitAndApply(t, terraformOptions)

    // Получение текущего состояния
    state1 := terraform.Show(t, terraformOptions)

    // Повторное применение (не должно быть изменений)
    terraform.Apply(t, terraformOptions)
    state2 := terraform.Show(t, terraformOptions)

    // Состояния должны быть идентичными
    assert.Equal(t, state1, state2, "Состояние изменилось при повторном apply (обнаружен drift)")

    // Очистка
    terraform.Destroy(t, terraformOptions)
}

# test_cost_estimate.py
import json
import subprocess

def estimate_terraform_cost(plan_file):
    """Оценка стоимости с помощью Infracost"""
    result = subprocess.run(
        ['infracost', 'breakdown', '--path', plan_file, '--format', 'json'],
        capture_output=True,
        text=True
    )

    return json.loads(result.stdout)

def test_monthly_cost_under_budget():
    """Проверка, что изменения инфраструктуры не превышают бюджет"""
    # Генерация плана
    subprocess.run(['terraform', 'plan', '-out=tfplan'], check=True)

    # Оценка стоимости
    cost_data = estimate_terraform_cost('tfplan')
    monthly_cost = cost_data['projects'][0]['breakdown']['totalMonthlyCost']

    # Лимит бюджета
    MAX_MONTHLY_COST = 10000.00

    assert float(monthly_cost) <= MAX_MONTHLY_COST, \
        f"Месячная стоимость ${monthly_cost} превышает бюджет ${MAX_MONTHLY_COST}"

def test_cost_increase_reasonable():
    """Проверка, что изменения не вызывают неожиданных всплесков стоимости"""
    # Получение текущей стоимости инфраструктуры
    current_cost = get_current_monthly_cost()

    # Получение новой стоимости инфраструктуры
    subprocess.run(['terraform', 'plan', '-out=tfplan'], check=True)
    cost_data = estimate_terraform_cost('tfplan')
    new_cost = float(cost_data['projects'][0]['breakdown']['totalMonthlyCost'])

    # Увеличение стоимости должно быть < 20%
    max_increase = current_cost * 1.20

    assert new_cost <= max_increase, \
        f"Слишком большое увеличение стоимости: ${current_cost} -> ${new_cost}"

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/antonbabenko/pre-commit-terraform
    rev: v1.81.0
    hooks:
      - id: terraform_fmt
      - id: terraform_validate
      - id: terraform_docs
      - id: terraform_tflint
        args:
          - --args=--config=__GIT_WORKING_DIR__/.tflint.hcl
      - id: terraform_checkov
        args:
          - --args=--quiet
          - --args=--framework terraform
      - id: terraform_tfsec

# Установка pre-commit
pip3 install pre-commit

# Установка хуков
pre-commit install

# Ручной запуск
pre-commit run --all-files

# environments/staging/main.tf
module "infrastructure" {
  source = "../../modules/infrastructure"

  environment = "staging"

  # Использование меньших инстансов для экономии
  instance_type = "t3.small"

  # Включение всего логирования для отладки
  enable_detailed_monitoring = true
  log_retention_days         = 7

  # Использование той же структуры конфигурации, что и в production
  # но с уменьшенными ресурсами
}

#!/bin/bash
# validate-staging.sh

set -e

echo "🧪 Тестирование в Staging окружении"

cd environments/staging

# 1. Валидация конфигурации
terraform validate

# 2. Сканирование безопасности
checkov -d . --quiet

# 3. План и сохранение
terraform plan -out=staging.tfplan

# 4. Применение в staging
terraform apply staging.tfplan

# 5. Запуск smoke тестов
./smoke-tests.sh

# 6. Запуск интеграционных тестов
go test -v ../test/integration_test.go

# 7. Мониторинг в течение 10 минут
echo "⏰ Мониторинг в течение 10 минут..."
./monitor-health.sh 600

echo "✅ Валидация staging завершена"

// test/drift_detection_test.go
func TestNoDrift(t *testing.T) {
    terraformOptions := &terraform.Options{
        TerraformDir: "../production",
    }

    // Не применять, только проверить на drift
    planOutput := terraform.InitAndPlan(t, terraformOptions)

    // Парсинг плана для обнаружения изменений
    planStruct := terraform.ParsePlanOutput(planOutput)

    resourcesChanged := planStruct.Add + planStruct.Change + planStruct.Destroy

    if resourcesChanged > 0 {
        t.Errorf("Обнаружен drift: %d ресурсов будут изменены", resourcesChanged)
        t.Logf("Вывод плана:\n%s", planOutput)
    }
}

# .github/workflows/drift-detection.yml
name: Drift Detection
on:
  schedule:
    - cron: '0 */6 * * *'  # Каждые 6 часов

jobs:
  detect-drift:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup Terraform
        uses: hashicorp/setup-terraform@v2

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v2
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: us-east-1

      - name: Check for Drift
        run: |
          cd environments/production
          terraform init
          terraform plan -detailed-exitcode || {
            echo "⚠️ ОБНАРУЖЕН DRIFT В PRODUCTION"
            # Отправка уведомления
            curl -X POST ${{ secrets.SLACK_WEBHOOK }} \
              -H 'Content-Type: application/json' \
              -d '{"text":"🚨 Обнаружен Terraform drift в production!"}'
            exit 1
          }

# Тестирование с новой версией модуля
module "vpc_test" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "5.0.0"  # Тестирование обновления с 4.x

  # ... конфигурация
}

#!/bin/bash
# test-module-upgrade.sh

OLD_VERSION="4.0.0"
NEW_VERSION="5.0.0"

echo "Тестирование обновления: $OLD_VERSION -> $NEW_VERSION"

# Создание тестового окружения со старой версией
cat > test_old.tf <<EOF
module "test" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "$OLD_VERSION"

  name = "upgrade-test"
  cidr = "10.0.0.0/16"
}
EOF

terraform init
terraform apply -auto-approve

# Захват состояния
OLD_STATE=$(terraform show -json)

# Обновление до новой версии
cat > test_new.tf <<EOF
module "test" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "$NEW_VERSION"

  name = "upgrade-test"
  cidr = "10.0.0.0/16"
}
EOF

terraform init -upgrade
terraform plan -out=upgrade.tfplan

# Проверка неожиданных изменений
python3 validate_plan.py upgrade.tfplan

terraform apply upgrade.tfplan

echo "✅ Обновление успешно"

// ПЛОХО - Жестко заданные тестовые значения
func TestInstance(t *testing.T) {
    terraformOptions := &terraform.Options{
        TerraformDir: "../",
        Vars: map[string]interface{}{
            "instance_type": "t2.micro",
            "ami":           "ami-12345678",
        },
    }
    // ...
}

// ХОРОШО - Реалистичные, регион-зависимые значения
func TestInstance(t *testing.T) {
    region := aws.GetRandomStableRegion(t, nil, nil)
    ami := aws.GetAmazonLinuxAmi(t, region)

    terraformOptions := &terraform.Options{
        TerraformDir: "../",
        Vars: map[string]interface{}{
            "instance_type": "t3.small",  // Текущее поколение
            "ami":           ami,
            "region":        region,
        },
    }
    // ...
}

func TestCompleteLifecycle(t *testing.T) {
    terraformOptions := &terraform.Options{
        TerraformDir: "../",
    }

    // Тест создания
    terraform.InitAndApply(t, terraformOptions)

    // Проверка существования ресурсов
    instanceID := terraform.Output(t, terraformOptions, "instance_id")
    instance := aws.GetEc2Instance(t, instanceID, "us-east-1")
    assert.NotNil(t, instance)

    // Тест удаления
    terraform.Destroy(t, terraformOptions)

    // Проверка удаления ресурсов
    _, err := aws.GetEc2InstanceE(t, instanceID, "us-east-1")
    assert.Error(t, err, "Инстанс не должен существовать после destroy")
}

# test_state_file.py
import json
import boto3

def test_state_file_integrity():
    """Проверка валидности и согласованности Terraform state файла"""
    s3 = boto3.client('s3')

    # Загрузка state файла
    response = s3.get_object(
        Bucket='terraform-state-bucket',
        Key='production/terraform.tfstate'
    )

    state = json.loads(response['Body'].read())

    # Валидация структуры
    assert 'version' in state
    assert 'terraform_version' in state
    assert 'resources' in state

    # Проверка на пустые ресурсы (обычно проблема)
    assert len(state['resources']) > 0, "State файл не содержит ресурсов"

    # Валидация целостности ресурсов
    for resource in state['resources']:
        assert 'type' in resource
        assert 'name' in resource
        assert 'instances' in resource

        for instance in resource['instances']:
            assert 'attributes' in instance
            # Проверка наличия критических атрибутов
            if resource['type'] == 'aws_instance':
                assert 'id' in instance['attributes']
                assert 'ami' in instance['attributes']

# Минимальная но эффективная настройка
terraform validate
tflint
checkov -d .

# Добавление интеграционного тестирования
terraform validate
tflint
checkov -d .
go test -v ./test/...  # Terratest

# Полный пайплайн валидации
- terraform validate
- terraform fmt -check
- tflint
- checkov -d .
- terrascan scan
- infracost breakdown --path .
- sentinel apply policy/  # При использовании Terraform Cloud
- go test -v -timeout 30m ./test/...
- drift detection (по расписанию)