Backup and Disaster Recovery Testing: Automated Validation of RTO/RPO with AWS, Azure, and Terraform

# Create a restore testing plan
aws backup create-restore-testing-plan \
  --restore-testing-plan-name "daily-validation" \
  --schedule-expression "cron(0 6 * * ? *)" \
  --start-window-hours 1 \
  --recovery-point-selection '{
    "Algorithm": "LATEST_WITHIN_WINDOW",
    "RecoveryPointTypes": ["SNAPSHOT"],
    "SelectionWindowDays": 1
  }'

# Add RDS to testing plan
aws backup create-restore-testing-selection \
  --restore-testing-plan-name "daily-validation" \
  --restore-testing-selection-name "rds-validation" \
  --protected-resource-type "RDS" \
  --iam-role-arn "arn:aws:iam::123456789012:role/BackupRestoreRole" \
  --protected-resource-conditions '{
    "StringEquals": [
      {"Key": "aws:ResourceTag/Environment", "Value": "production"}
    ]
  }'

# modules/backup-testing/main.tf

resource "aws_backup_plan" "production" {
  name = "production-backup-plan"

  rule {
    rule_name         = "daily-backups"
    target_vault_name = aws_backup_vault.production.name
    schedule          = "cron(0 5 * * ? *)"

    lifecycle {
      delete_after = 90  # SOC 2 compliance
    }

    copy_action {
      destination_vault_arn = aws_backup_vault.dr_region.arn
      lifecycle {
        delete_after = 90
      }
    }
  }

  rule {
    rule_name         = "hourly-backups"
    target_vault_name = aws_backup_vault.production.name
    schedule          = "cron(0 * * * ? *)"

    lifecycle {
      delete_after = 7
    }
  }
}

resource "aws_backup_selection" "production_rds" {
  iam_role_arn = aws_iam_role.backup.arn
  name         = "production-rds"
  plan_id      = aws_backup_plan.production.id

  selection_tag {
    type  = "STRINGEQUALS"
    key   = "Backup"
    value = "true"
  }

  resources = [
    "arn:aws:rds:*:*:db:*",
    "arn:aws:rds:*:*:cluster:*"
  ]
}

# Restore testing plan
resource "aws_backup_restore_testing_plan" "validation" {
  name = "automated-restore-validation"

  recovery_point_selection {
    algorithm             = "LATEST_WITHIN_WINDOW"
    recovery_point_types  = ["SNAPSHOT"]
    selection_window_days = 1
  }

  schedule_expression = "cron(0 8 * * ? *)"
}

import boto3
import time
from datetime import datetime, timedelta

class BackupValidator:
    def __init__(self, region='us-east-1'):
        self.backup = boto3.client('backup', region_name=region)
        self.rds = boto3.client('rds', region_name=region)

    def validate_rds_backup(self, db_identifier, max_age_hours=24):
        """Validate RDS backup exists and is restorable."""
        # Get latest snapshot
        snapshots = self.rds.describe_db_snapshots(
            DBInstanceIdentifier=db_identifier,
            SnapshotType='automated'
        )['DBSnapshots']

        if not snapshots:
            return {'status': 'FAILED', 'reason': 'No snapshots found'}

        latest = max(snapshots, key=lambda x: x['SnapshotCreateTime'])
        age = datetime.now(latest['SnapshotCreateTime'].tzinfo) - latest['SnapshotCreateTime']

        if age > timedelta(hours=max_age_hours):
            return {
                'status': 'WARNING',
                'reason': f'Latest snapshot is {age.total_seconds()/3600:.1f} hours old'
            }

        # Verify snapshot is available
        if latest['Status'] != 'available':
            return {'status': 'FAILED', 'reason': f"Snapshot status: {latest['Status']}"}

        return {
            'status': 'PASSED',
            'snapshot_id': latest['DBSnapshotIdentifier'],
            'snapshot_age_hours': age.total_seconds() / 3600
        }

    def test_restore(self, snapshot_id, test_db_identifier):
        """Perform actual restore test."""
        try:
            # Start restore
            self.rds.restore_db_instance_from_db_snapshot(
                DBInstanceIdentifier=test_db_identifier,
                DBSnapshotIdentifier=snapshot_id,
                DBInstanceClass='db.t3.micro',  # Minimal for testing
                PubliclyAccessible=False,
                Tags=[{'Key': 'Purpose', 'Value': 'DR-Test'}]
            )

            # Wait for restore
            start_time = time.time()
            waiter = self.rds.get_waiter('db_instance_available')
            waiter.wait(DBInstanceIdentifier=test_db_identifier)
            restore_time = time.time() - start_time

            return {
                'status': 'PASSED',
                'restore_time_seconds': restore_time,
                'meets_rto': restore_time < 7200  # 2 hour RTO
            }

        finally:
            # Cleanup test instance
            try:
                self.rds.delete_db_instance(
                    DBInstanceIdentifier=test_db_identifier,
                    SkipFinalSnapshot=True,
                    DeleteAutomatedBackups=True
                )
            except Exception:
                pass

# Usage in pytest
def test_production_db_backup():
    validator = BackupValidator()
    result = validator.validate_rds_backup('production-db')
    assert result['status'] in ['PASSED', 'WARNING'], f"Backup validation failed: {result}"

# Create a recovery plan
az site-recovery recovery-plan create \
  --resource-group rg-dr \
  --vault-name vault-dr \
  --name "webapp-recovery-plan" \
  --primary-zone "East US" \
  --recovery-zone "West US" \
  --failover-deployment-model "Resource"

# Run test failover (non-disruptive)
az site-recovery recovery-plan test-failover \
  --resource-group rg-dr \
  --vault-name vault-dr \
  --name "webapp-recovery-plan" \
  --failover-direction "PrimaryToRecovery" \
  --network-id "/subscriptions/.../test-vnet"

# Validate test environment
az vm list --resource-group rg-dr-test --output table

# Cleanup test failover
az site-recovery recovery-plan test-failover-cleanup \
  --resource-group rg-dr \
  --vault-name vault-dr \
  --name "webapp-recovery-plan"

# modules/azure-dr/main.tf

resource "azurerm_recovery_services_vault" "dr" {
  name                = "vault-dr-${var.environment}"
  location            = var.dr_region
  resource_group_name = azurerm_resource_group.dr.name
  sku                 = "Standard"

  soft_delete_enabled = true
}

resource "azurerm_site_recovery_fabric" "primary" {
  name                = "fabric-primary"
  resource_group_name = azurerm_resource_group.dr.name
  recovery_vault_name = azurerm_recovery_services_vault.dr.name
  location            = var.primary_region
}

resource "azurerm_site_recovery_fabric" "secondary" {
  name                = "fabric-secondary"
  resource_group_name = azurerm_resource_group.dr.name
  recovery_vault_name = azurerm_recovery_services_vault.dr.name
  location            = var.dr_region
}

resource "azurerm_site_recovery_replication_policy" "policy" {
  name                                                 = "replication-policy"
  resource_group_name                                  = azurerm_resource_group.dr.name
  recovery_vault_name                                  = azurerm_recovery_services_vault.dr.name
  recovery_point_retention_in_minutes                  = 1440  # 24 hours
  application_consistent_snapshot_frequency_in_minutes = 60    # 1 hour RPO
}

# Backup policy for VMs
resource "azurerm_backup_policy_vm" "daily" {
  name                = "daily-backup-policy"
  resource_group_name = azurerm_resource_group.dr.name
  recovery_vault_name = azurerm_recovery_services_vault.dr.name

  backup {
    frequency = "Daily"
    time      = "23:00"
  }

  retention_daily {
    count = 30
  }

  retention_weekly {
    count    = 12
    weekdays = ["Sunday"]
  }

  retention_monthly {
    count    = 12
    weekdays = ["Sunday"]
    weeks    = ["First"]
  }
}

# dr-test/main.tf - Spin up, test, destroy

variable "run_dr_test" {
  description = "Set to true to run DR test"
  type        = bool
  default     = false
}

# Only create resources during DR test
resource "aws_db_instance" "dr_test" {
  count = var.run_dr_test ? 1 : 0

  identifier     = "dr-test-${formatdate("YYYYMMDD", timestamp())}"
  instance_class = "db.t3.medium"

  # Restore from latest snapshot
  snapshot_identifier = data.aws_db_snapshot.latest.id

  vpc_security_group_ids = [aws_security_group.dr_test[0].id]
  db_subnet_group_name   = aws_db_subnet_group.dr_test[0].name

  skip_final_snapshot = true
  deletion_protection = false

  tags = {
    Purpose   = "DR-Test"
    AutoClean = "true"
  }
}

data "aws_db_snapshot" "latest" {
  db_instance_identifier = var.production_db_identifier
  most_recent            = true
}

# Test application connectivity
resource "null_resource" "validate_dr" {
  count = var.run_dr_test ? 1 : 0

  depends_on = [aws_db_instance.dr_test]

  provisioner "local-exec" {
    command = <<-EOT
      # Wait for DB to be ready
      aws rds wait db-instance-available \
        --db-instance-identifier ${aws_db_instance.dr_test[0].identifier}

      # Run connectivity test
      python3 scripts/validate_dr.py \
        --endpoint ${aws_db_instance.dr_test[0].endpoint} \
        --expected-tables 50 \
        --expected-rows 1000000

      # Record RTO
      echo "DR Test completed. RTO: $(cat /tmp/dr_rto.txt)"
    EOT
  }
}

output "dr_test_results" {
  value = var.run_dr_test ? {
    db_endpoint  = aws_db_instance.dr_test[0].endpoint
    restore_time = timestamp()
    status       = "Test environment ready"
  } : null
}

name: Quarterly DR Test

on:
  schedule:

    - cron: '0 6 1 */3 *'  # First day of quarter at 6 AM
  workflow_dispatch:
    inputs:
      test_type:
        description: 'Type of DR test'
        required: true
        default: 'backup-restore'
        type: choice
        options:

          - backup-restore
          - failover
          - full-dr

jobs:
  dr-test:
    runs-on: ubuntu-latest
    environment: dr-test

    steps:

      - uses: actions/checkout@v4

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.DR_TEST_ROLE_ARN }}
          aws-region: us-west-2  # DR region

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

      - name: Start DR Test
        id: start_test
        run: |
          START_TIME=$(date +%s)
          echo "start_time=$START_TIME" >> $GITHUB_OUTPUT
          echo "DR Test started at $(date)"

      - name: Deploy DR Environment
        run: |
          cd terraform/dr-test
          terraform init
          terraform apply -auto-approve -var="run_dr_test=true"

      - name: Validate Recovery
        id: validate
        run: |
          cd terraform/dr-test
          DB_ENDPOINT=$(terraform output -raw dr_test_db_endpoint)

          # Test database connectivity
          python3 scripts/validate_dr.py \
            --endpoint "$DB_ENDPOINT" \
            --test-queries \
            --output-file dr-results.json

          # Calculate RTO
          END_TIME=$(date +%s)
          START_TIME=${{ steps.start_test.outputs.start_time }}
          RTO=$((END_TIME - START_TIME))
          echo "rto_seconds=$RTO" >> $GITHUB_OUTPUT

      - name: Cleanup DR Test
        if: always()
        run: |
          cd terraform/dr-test
          terraform destroy -auto-approve -var="run_dr_test=true"

      - name: Generate Report
        run: |
          cat << EOF > dr-report.md
          # DR Test Report - $(date +%Y-%m-%d)

          ## Results
          - **Test Type**: ${{ inputs.test_type || 'backup-restore' }}
          - **Actual RTO**: ${{ steps.validate.outputs.rto_seconds }} seconds
          - **Target RTO**: 7200 seconds (2 hours)
          - **Status**: $([[ ${{ steps.validate.outputs.rto_seconds }} -lt 7200 ]] && echo "PASSED" || echo "FAILED")

          ## Validation Details
          $(cat dr-results.json | jq -r '.summary')
          EOF

      - name: Upload Report
        uses: actions/upload-artifact@v4
        with:
          name: dr-test-report
          path: dr-report.md

      - name: Notify on Failure
        if: failure()
        run: |
          # Send alert if DR test failed
          aws sns publish \
            --topic-arn ${{ secrets.ALERTS_TOPIC_ARN }} \
            --message "DR Test FAILED. Review required."