Estrategias de Testing y Validación en Terraform: Guía Completa de DevOps

# Este cambio aparentemente inocente destruyó producción
resource "aws_s3_bucket" "data" {
  bucket = "company-production-data"
  # El desarrollador pensó que esto solo añadiría versionado...
  force_destroy = true  # ⚠️ PELIGRO: ¡Elimina todos los objetos al destruir!
}

# Verificación básica de sintaxis y consistencia interna
terraform validate

# Ejemplo de salida para errores:
# Error: Unsupported argument
#   on main.tf line 12:
#   12:   instance_types = "t2.micro"
# An argument named "instance_types" is not expected here.

# Instalar tflint
curl -s https://raw.githubusercontent.com/terraform-linters/tflint/master/install_linux.sh | bash

# Crear configuración .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

# Ejecutar tflint
tflint --init
tflint

# Instalar checkov
pip3 install checkov

# Escanear archivos Terraform
checkov -d . --framework terraform

# Ejecutar verificaciones específicas
checkov -d . --check CKV_AWS_8  # Asegurar que EBS esté cifrado

# Salida en JSON para integración CI/CD
checkov -d . -o json > security-report.json

# Antes - Violaciones de seguridad
resource "aws_s3_bucket" "logs" {
  bucket = "company-logs"
  # Falta: versionado, cifrado, bloqueo de acceso público
}

# Después - Cumplimiento de seguridad
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
}

# Generar plan y guardar en archivo
terraform plan -out=tfplan

# Convertir plan binario a JSON para análisis
terraform show -json tfplan > tfplan.json

# Analizar plan con jq
cat tfplan.json | jq -r '
  .resource_changes[] |
  select(.change.actions[] | contains("delete")) |
  "⚠️  DELETE: \(.address)"
'

# Ejemplo de salida:
# ⚠️  DELETE: aws_instance.old_server
# ⚠️  DELETE: aws_security_group.deprecated

# validate_plan.py - Prevenir cambios peligrosos
import json
import sys

def validate_terraform_plan(plan_file):
    """Validar plan de Terraform para operaciones peligrosas"""
    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']

        # Verificar eliminaciones de recursos críticos
        if 'delete' in actions:
            if 'database' in address or 'rds' in address:
                errors.append(f"🚨 BLOQUEADO: Intentando eliminar base de datos: {address}")
            elif 's3_bucket' in address and 'backup' in address:
                errors.append(f"🚨 BLOQUEADO: Intentando eliminar bucket de respaldo: {address}")
            else:
                warnings.append(f"⚠️  Advertencia: Eliminando recurso: {address}")

        # Verificar recreación (reemplazo)
        if 'delete' in actions and 'create' in actions:
            if 'aws_instance' in address:
                warnings.append(f"⚠️  La instancia será recreada: {address}")

        # Verificar cambios en reglas de 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"🚨 BLOQUEADO: Security group permite acceso público: {address}")

    # Imprimir resultados
    if errors:
        print("\n❌ VALIDACIÓN FALLIDA - Problemas Críticos Encontrados:\n")
        for error in errors:
            print(error)
        return False

    if warnings:
        print("\n⚠️  Advertencias (revisar antes de aplicar):\n")
        for warning in warnings:
            print(warning)

    print("\n✅ Validación de plan aprobada")
    return True

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Uso: 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()

    // Construir opciones de 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",
        },
    })

    // Limpiar recursos al final
    defer terraform.Destroy(t, terraformOptions)

    // Desplegar infraestructura
    terraform.InitAndApply(t, terraformOptions)

    // Validar outputs
    instanceID := terraform.Output(t, terraformOptions, "instance_id")
    publicIP := terraform.Output(t, terraformOptions, "public_ip")

    // Verificar que la instancia existe y está corriendo
    instance := aws.GetEc2Instance(t, instanceID, "us-east-1")
    assert.Equal(t, "running", instance.State.Name)
    assert.Equal(t, "t2.micro", instance.InstanceType)

    // Verificar que el servidor web responde
    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)

    // Validar que el VPC fue creado
    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)

    // Validar subnets
    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")

    // Verificar que la base de datos es accesible
    err := testDatabaseConnection(dbEndpoint, "admin", "password123")
    assert.NoError(t, err)

    // Simular failover
    aws.RebootRdsInstance(t, dbInstanceID, "us-east-1")

    // Esperar a que se complete el failover
    maxRetries := 10
    timeBetweenRetries := 30 * time.Second

    for i := 0; i < maxRetries; i++ {
        err = testDatabaseConnection(dbEndpoint, "admin", "password123")
        if err == nil {
            t.Logf("Base de datos recuperada después de %d intentos", i+1)
            return
        }
        time.Sleep(timeBetweenRetries)
    }

    t.Fatal("La base de datos no se recuperó después del failover")
}

// Testear múltiples escenarios
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 // Capturar variable de rango
        t.Run(tc.name, func(t *testing.T) {
            t.Parallel()
            // Lógica de prueba aquí
        })
    }
}

// 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",
        },
    }

    // Aplicar infraestructura
    terraform.InitAndApply(t, terraformOptions)

    // Obtener estado actual
    state1 := terraform.Show(t, terraformOptions)

    // Aplicar de nuevo (no debería haber cambios)
    terraform.Apply(t, terraformOptions)
    state2 := terraform.Show(t, terraformOptions)

    // Los estados deberían ser idénticos
    assert.Equal(t, state1, state2, "Estado cambió en re-aplicación (drift detectado)")

    // Limpieza
    terraform.Destroy(t, terraformOptions)
}

# test_cost_estimate.py
import json
import subprocess

def estimate_terraform_cost(plan_file):
    """Estimar costos usando 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():
    """Asegurar que los cambios de infraestructura no excedan el presupuesto"""
    # Generar plan
    subprocess.run(['terraform', 'plan', '-out=tfplan'], check=True)

    # Estimar costo
    cost_data = estimate_terraform_cost('tfplan')
    monthly_cost = cost_data['projects'][0]['breakdown']['totalMonthlyCost']

    # Límite de presupuesto
    MAX_MONTHLY_COST = 10000.00

    assert float(monthly_cost) <= MAX_MONTHLY_COST, \
        f"Costo mensual ${monthly_cost} excede presupuesto ${MAX_MONTHLY_COST}"

def test_cost_increase_reasonable():
    """Asegurar que los cambios no causen picos de costo inesperados"""
    # Obtener costo actual de infraestructura
    current_cost = get_current_monthly_cost()

    # Obtener nuevo costo de infraestructura
    subprocess.run(['terraform', 'plan', '-out=tfplan'], check=True)
    cost_data = estimate_terraform_cost('tfplan')
    new_cost = float(cost_data['projects'][0]['breakdown']['totalMonthlyCost'])

    # El aumento de costo debería ser < 20%
    max_increase = current_cost * 1.20

    assert new_cost <= max_increase, \
        f"Aumento de costo demasiado grande: ${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

# Instalar pre-commit
pip3 install pre-commit

# Instalar hooks
pre-commit install

# Ejecutar manualmente
pre-commit run --all-files

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

  environment = "staging"

  # Usar instancias más pequeñas para ahorrar costos
  instance_type = "t3.small"

  # Habilitar todo el logging para debugging
  enable_detailed_monitoring = true
  log_retention_days         = 7

  # Usar la misma estructura de configuración que producción
  # pero con recursos reducidos
}

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

set -e

echo "🧪 Testeando en Ambiente de Staging"

cd environments/staging

# 1. Validar configuración
terraform validate

# 2. Escaneo de seguridad
checkov -d . --quiet

# 3. Plan y guardar
terraform plan -out=staging.tfplan

# 4. Aplicar a staging
terraform apply staging.tfplan

# 5. Ejecutar smoke tests
./smoke-tests.sh

# 6. Ejecutar tests de integración
go test -v ../test/integration_test.go

# 7. Monitorear por 10 minutos
echo "⏰ Monitoreando por 10 minutos..."
./monitor-health.sh 600

echo "✅ Validación de staging completa"

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

    // No aplicar, solo verificar drift
    planOutput := terraform.InitAndPlan(t, terraformOptions)

    // Parsear plan para detectar cambios
    planStruct := terraform.ParsePlanOutput(planOutput)

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

    if resourcesChanged > 0 {
        t.Errorf("Drift detectado: %d recursos cambiarían", resourcesChanged)
        t.Logf("Salida del plan:\n%s", planOutput)
    }
}

# .github/workflows/drift-detection.yml
name: Drift Detection
on:
  schedule:
    - cron: '0 */6 * * *'  # Cada 6 horas

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 DETECTADO EN PRODUCCIÓN"
            # Enviar alerta
            curl -X POST ${{ secrets.SLACK_WEBHOOK }} \
              -H 'Content-Type: application/json' \
              -d '{"text":"🚨 Drift de Terraform detectado en producción!"}'
            exit 1
          }

# Testear con nueva versión del módulo
module "vpc_test" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "5.0.0"  # Testeando actualización desde 4.x

  # ... configuración
}

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

OLD_VERSION="4.0.0"
NEW_VERSION="5.0.0"

echo "Testeando actualización: $OLD_VERSION -> $NEW_VERSION"

# Crear ambiente de test con versión antigua
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

# Capturar estado
OLD_STATE=$(terraform show -json)

# Actualizar a nueva versión
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

# Verificar cambios inesperados
python3 validate_plan.py upgrade.tfplan

terraform apply upgrade.tfplan

echo "✅ Actualización exitosa"

// MAL - Valores de test hardcoded
func TestInstance(t *testing.T) {
    terraformOptions := &terraform.Options{
        TerraformDir: "../",
        Vars: map[string]interface{}{
            "instance_type": "t2.micro",
            "ami":           "ami-12345678",
        },
    }
    // ...
}

// BIEN - Valores realistas conscientes de la región
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",  // Generación actual
            "ami":           ami,
            "region":        region,
        },
    }
    // ...
}

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

    // Testear create
    terraform.InitAndApply(t, terraformOptions)

    // Verificar que los recursos existen
    instanceID := terraform.Output(t, terraformOptions, "instance_id")
    instance := aws.GetEc2Instance(t, instanceID, "us-east-1")
    assert.NotNil(t, instance)

    // Testear destroy
    terraform.Destroy(t, terraformOptions)

    // Verificar que los recursos han desaparecido
    _, err := aws.GetEc2InstanceE(t, instanceID, "us-east-1")
    assert.Error(t, err, "La instancia no debería existir después de destroy")
}

# test_state_file.py
import json
import boto3

def test_state_file_integrity():
    """Verificar que el archivo de estado de Terraform es válido y consistente"""
    s3 = boto3.client('s3')

    # Descargar archivo de estado
    response = s3.get_object(
        Bucket='terraform-state-bucket',
        Key='production/terraform.tfstate'
    )

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

    # Validar estructura
    assert 'version' in state
    assert 'terraform_version' in state
    assert 'resources' in state

    # Verificar recursos vacíos (usualmente un problema)
    assert len(state['resources']) > 0, "El archivo de estado no tiene recursos"

    # Validar integridad de recursos
    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
            # Verificar que existen atributos críticos
            if resource['type'] == 'aws_instance':
                assert 'id' in instance['attributes']
                assert 'ami' in instance['attributes']

# Configuración mínima pero efectiva
terraform validate
tflint
checkov -d .

# Añadir testing de integración
terraform validate
tflint
checkov -d .
go test -v ./test/...  # Terratest

# Pipeline de validación completo
- terraform validate
- terraform fmt -check
- tflint
- checkov -d .
- terrascan scan
- infracost breakdown --path .
- sentinel apply policy/  # Si usas Terraform Cloud
- go test -v -timeout 30m ./test/...
- drift detection (programado)