Blue-Green Deployment Testing: Strategies for Zero-Risk Releases

import httpx
import asyncio
from typing import List, Tuple

async def check_endpoint(client: httpx.AsyncClient, url: str) -> Tuple[str, bool, int]:
    try:
        response = await client.get(url, timeout=10.0)
        return url, response.status_code < 400, response.status_code
    except Exception as e:
        return url, False, 0

async def validate_green_environment(green_base_url: str) -> bool:
    endpoints = [
        f"{green_base_url}/health",
        f"{green_base_url}/api/v1/health",
        f"{green_base_url}/api/v2/health",
        f"{green_base_url}/",
    ]
    
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[
            check_endpoint(client, url) for url in endpoints
        ])
    
    all_healthy = all(healthy for _, healthy, _ in results)
    
    for url, healthy, status in results:
        status_icon = "✓" if healthy else "✗"
        print(f"  {status_icon} {url} → {status}")
    
    return all_healthy

# smoke_tests.py
import pytest
import httpx

GREEN_BASE_URL = os.environ.get("GREEN_BASE_URL")

def test_homepage_loads():
    response = httpx.get(f"{GREEN_BASE_URL}/")
    assert response.status_code == 200
    assert "HelpMeTest" in response.text

def test_api_authentication():
    response = httpx.post(
        f"{GREEN_BASE_URL}/api/v1/auth/token",
        json={"email": TEST_USER_EMAIL, "password": TEST_USER_PASSWORD}
    )
    assert response.status_code == 200
    assert "token" in response.json()

def test_critical_user_flow():
    """Test the most business-critical user journey."""
    token = get_test_token(GREEN_BASE_URL)
    
    # Create a test
    response = httpx.post(
        f"{GREEN_BASE_URL}/api/v1/tests",
        headers={"Authorization": f"Bearer {token}"},
        json={"name": "Smoke test", "url": "https://example.com"}
    )
    assert response.status_code == 201
    test_id = response.json()["id"]
    
    # Verify it's accessible
    response = httpx.get(
        f"{GREEN_BASE_URL}/api/v1/tests/{test_id}",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    
    # Clean up
    httpx.delete(
        f"{GREEN_BASE_URL}/api/v1/tests/{test_id}",
        headers={"Authorization": f"Bearer {token}"}
    )

def validate_database_migration(green_db_connection):
    """Verify database schema is correct after migration."""
    
    # Check schema version
    version = green_db_connection.execute(
        "SELECT version FROM schema_migrations ORDER BY applied_at DESC LIMIT 1"
    ).scalar()
    assert version == EXPECTED_SCHEMA_VERSION, \
        f"Schema version mismatch: {version} vs {EXPECTED_SCHEMA_VERSION}"
    
    # Verify critical tables exist
    for table in CRITICAL_TABLES:
        exists = green_db_connection.execute(
            f"SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name='{table}')"
        ).scalar()
        assert exists, f"Critical table {table} missing after migration"
    
    # Verify data integrity post-migration
    sample_user = green_db_connection.execute(
        "SELECT * FROM users LIMIT 1"
    ).fetchone()
    assert sample_user is not None, "No users found after migration"
    
    # If migration adds columns, verify they have default values
    null_count = green_db_connection.execute(
        "SELECT COUNT(*) FROM users WHERE new_required_column IS NULL"
    ).scalar()
    assert null_count == 0, "Migration left NULL values in required column"

#!/bin/bash
# blue-green-switch.sh

set -e

GREEN_URL="$1"
LOAD_BALANCER_ARN="$2"
TARGET_GROUP_GREEN_ARN="$3"

echo "🔍 Validating green environment..."
python smoke_tests.py --base-url "$GREEN_URL"

if [ $? -ne 0 ]; then
    echo "❌ Smoke tests failed. Not switching traffic."
    exit 1
fi

echo "✅ Smoke tests passed. Switching traffic to green..."
aws elbv2 modify-listener \
    --listener-arn "$LOAD_BALANCER_ARN" \
    --default-actions Type=forward,TargetGroupArn="$TARGET_GROUP_GREEN_ARN"

echo "⏳ Waiting for traffic switch to propagate..."
sleep 30

echo "🔍 Validating production traffic on green..."
python scripts/validate_production.py --max-error-rate 0.01

if [ $? -ne 0 ]; then
    echo "⚠️  Production validation failed. Initiating rollback..."
    ./rollback.sh
    exit 1
fi

echo "🎉 Deployment successful!"

# rollout.yaml
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: my-app
spec:
  strategy:
    blueGreen:
      activeService: my-app-active
      previewService: my-app-preview
      autoPromotionEnabled: false
      prePromotionAnalysis:
        templates:
          - templateName: smoke-tests
        args:
          - name: service-name
            value: my-app-preview
      postPromotionAnalysis:
        templates:
          - templateName: error-rate-check

---
# analysis-template.yaml
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: smoke-tests
spec:
  args:
    - name: service-name
  metrics:
    - name: smoke-test-job
      provider:
        job:
          spec:
            template:
              spec:
                containers:
                  - name: smoke-tests
                    image: my-smoke-tests:latest
                    env:
                      - name: TARGET_URL
                        value: "http://{{args.service-name}}"

def test_rollback_switches_traffic():
    """Verify rollback redirects traffic back to blue within SLA."""
    # Establish baseline: traffic on green
    assert_traffic_on_version("green")
    
    # Trigger rollback
    start = time.time()
    trigger_rollback_to_blue()
    
    # Poll until traffic is on blue
    deadline = time.time() + 30  # 30 second SLA
    while time.time() < deadline:
        if get_active_version() == "blue":
            break
        time.sleep(1)
    
    rollback_time = time.time() - start
    
    assert get_active_version() == "blue", "Rollback failed to switch traffic"
    assert rollback_time < 30, f"Rollback took {rollback_time:.0f}s, SLA is 30s"

def test_blue_environment_healthy_for_rollback():
    """Blue environment must remain healthy during green deployment."""
    # While deploying to green, blue should still be serving fine
    
    # Check blue health during green deployment
    blue_health = check_environment_health("blue")
    assert blue_health.healthy, "Blue environment degraded during green deployment"
    
    # Verify blue can handle production traffic if switched back
    blue_load_test = run_load_test(
        target_url=BLUE_URL,
        duration_seconds=60,
        requests_per_second=100
    )
    assert blue_load_test.error_rate < 0.01
    assert blue_load_test.p95_latency_ms < 500

def test_new_code_works_with_new_schema():
    """Green version should work correctly with migrated schema."""
    run_database_migration()
    
    response = httpx.post(
        f"{GREEN_URL}/api/v1/users",
        json={"name": "Test User", "email": "test@example.com"}
    )
    assert response.status_code == 201

def test_old_code_works_with_new_schema():
    """Blue version must continue working after database migration.
    This ensures instant rollback is safe."""
    run_database_migration()
    
    # Blue should still function with the new schema
    response = httpx.get(f"{BLUE_URL}/api/v1/users/1")
    assert response.status_code == 200
    
def test_feature_toggle_enables_new_schema_columns():
    """New schema columns should only be used when feature flag is on."""
    # During deployment: blue is live, green is being tested
    # The migration has run, but new columns shouldn't affect blue
    
    old_user = get_user_via_blue(user_id=1)
    new_user = get_user_via_green(user_id=1)
    
    # Both should return the same data for existing fields
    assert old_user['name'] == new_user['name']
    assert old_user['email'] == new_user['email']

def post_switch_monitoring(switch_time: datetime, duration_minutes: int = 10):
    """Monitor key metrics for duration_minutes after switching to green."""
    
    metrics_to_watch = [
        "http_error_rate",
        "p95_latency_ms",
        "requests_per_second",
        "database_connection_pool_usage"
    ]
    
    baseline = get_metrics_baseline(before=switch_time, window_minutes=30)
    
    for minute in range(duration_minutes):
        time.sleep(60)
        current = get_current_metrics()
        
        issues = []
        for metric in metrics_to_watch:
            if current[metric] > baseline[metric] * 1.5:
                issues.append(
                    f"{metric}: {current[metric]:.2f} vs baseline {baseline[metric]:.2f}"
                )
        
        if issues:
            logger.warning(f"Metrics degraded at minute {minute + 1}: {issues}")
            if current["http_error_rate"] > 0.05:  # >5% errors = rollback
                trigger_automatic_rollback()
                return False
    
    return True