# Install e2e test frameworkcurl -sSL https://releases.cleanstart.dev/e2e-tests-latest.tar.gz | tar xzchmod +x ./e2e-tests # Verify./e2e-tests --version # Install dependenciespip install pytest pytest-asyncio pyyaml docker # Set environment variablesexport REGISTRY_URL="gcr.io/my-project"export STAGING_CLUSTER="gke-staging"export PROD_CLUSTER="gke-prod-us-central1"export DOCKER_CONFIG="$HOME/.docker"

version: "1.0"project: my-appimage_name: appenvironment:  staging:    cluster: gke-staging    namespace: staging    registry: gcr.io/my-project    replicas: 2  production:    cluster: gke-prod-us-central1    namespace: production    registry: gcr.io/my-project    replicas: 5    canary_percentage: 20  # Canary: 20% of traffic initially tests:  baseline:    enabled: true    timeout: 300  # 5 minutes  api:    enabled: true    timeout: 600    endpoints:      - path: /health        method: GET        expected_status: 200      - path: /api/v1/items        method: GET        expected_status: 200      - path: /api/v1/items        method: POST        body: '{"name": "test"}'        expected_status: 201  database:    enabled: true    timeout: 300    connection_string: postgresql://user:pass@postgres:5432/mydb  performance:    enabled: true    timeout: 900    thresholds:      p95_response_time_ms: 200      throughput_rps: 1000      error_rate_percent: 0.1

# test_pipeline_trigger.pyimport pytestimport subprocessimport timefrom datetime import datetime def test_github_webhook_triggers_build():    """Verify GitHub webhook triggers CI/CD pipeline"""    # Create test commit in feature branch    subprocess.run(["git", "checkout", "-b", "e2e-test-trigger"], check=True)     with open("test-file.txt", "w") as f:        f.write(f"E2E test {datetime.now()}")     subprocess.run(["git", "add", "test-file.txt"], check=True)    subprocess.run(["git", "commit", "-m", "E2E: Trigger pipeline test"], check=True)    subprocess.run(["git", "push", "origin", "e2e-test-trigger"], check=True)     # Poll for build start (GitHub Actions API)    max_retries = 60  # 5 minutes    build_found = False     for _ in range(max_retries):        result = subprocess.run(            ["gh", "run", "list", "--branch", "e2e-test-trigger", "--limit", "1", "--json", "status"],            capture_output=True,            text=True        )         if "in_progress" in result.stdout or "completed" in result.stdout:            build_found = True            break         time.sleep(5)     assert build_found, "Build was not triggered by webhook"     # Cleanup    subprocess.run(["git", "checkout", "main"], check=True)    subprocess.run(["git", "branch", "-D", "e2e-test-trigger"], check=True)

# test_image_build.pyimport subprocessimport dockerimport json def test_image_builds_successfully():    """Verify image builds without errors"""    result = subprocess.run(        ["docker", "build", "-t", "test-app:e2e", "."],        capture_output=True,        text=True    )     assert result.returncode == 0, f"Build failed: {result.stderr}"     client = docker.from_env()    image = client.images.get("test-app:e2e")    assert image is not None def test_image_signed_correctly():    """Verify image is signed with correct key"""    image_uri = f"{REGISTRY_URL}/app:latest"     result = subprocess.run(        ["cosign", "verify", "--key", "cosign.pub", image_uri],        capture_output=True,        text=True    )     assert result.returncode == 0, "Image signature verification failed"    assert "Signature verified" in result.stdout def test_sbom_attached_to_image():    """Verify SBOM attestation is attached"""    image_uri = f"{REGISTRY_URL}/app:latest"     result = subprocess.run(        ["cosign", "verify-attestation", "--type", "sbom", image_uri],        capture_output=True,        text=True    )     assert result.returncode == 0, "SBOM attestation not found"     # Parse SBOM    attestation_json = json.loads(result.stdout)    sbom_data = json.loads(attestation_json['payload'])     assert "packages" in sbom_data, "SBOM missing packages"    assert len(sbom_data["packages"]) > 0, "SBOM has no packages" def test_image_pushed_to_registry():    """Verify image is in registry and accessible"""    image_uri = f"{REGISTRY_URL}/app:latest"     result = subprocess.run(        ["gcloud", "container", "images", "describe", image_uri],        capture_output=True,        text=True    )     assert result.returncode == 0, f"Image not found in registry: {image_uri}"

# test_staging_deployment.pyimport subprocessimport timeimport requestsfrom kubernetes import client, config def test_staging_image_pull():    """Verify image can be pulled in staging cluster"""    config.load_kube_config(context="gke-staging")    v1 = client.CoreV1Api()     # Get imagePullSecrets from namespace    ns = v1.read_namespace("staging")    assert ns is not None def test_staging_pod_deployment():    """Verify pods deploy successfully in staging"""    # Trigger deployment    result = subprocess.run(        ["kubectl", "set", "image", "deployment/app",         f"app={REGISTRY_URL}/app:latest", "-n", "staging"],        capture_output=True,        text=True    )     assert result.returncode == 0     # Wait for rollout    result = subprocess.run(        ["kubectl", "rollout", "status", "deployment/app", "-n", "staging", "--timeout=5m"],        capture_output=True,        text=True    )     assert result.returncode == 0, "Deployment rollout failed" def test_staging_health_check():    """Verify health endpoints in staging"""    # Port forward to staging service    pf_process = subprocess.Popen(        ["kubectl", "port-forward", "svc/app", "8080:80", "-n", "staging"],        stdout=subprocess.PIPE    )     time.sleep(3)  # Wait for port forward to establish     try:        # Test health endpoint        response = requests.get("http://localhost:8080/health", timeout=10)        assert response.status_code == 200    finally:        pf_process.terminate() def test_staging_readiness_probe():    """Verify readiness probes pass in staging"""    config.load_kube_config(context="gke-staging")    v1 = client.CoreV1Api()     pods = v1.list_namespaced_pod("staging", label_selector="app=app")     for pod in pods.items:        # Check all containers are ready        for container_status in pod.status.container_statuses:            assert container_status.ready, f"Container {container_status.name} not ready"

# test_api_integration.pyimport requestsimport pytestimport json BASE_URL = "http://localhost:8000"  # After port-forward def test_api_health_endpoint():    """Test health endpoint"""    response = requests.get(f"{BASE_URL}/health")    assert response.status_code == 200    assert response.json()["status"] == "ok" def test_api_list_items():    """Test GET /api/v1/items"""    response = requests.get(f"{BASE_URL}/api/v1/items")    assert response.status_code == 200    assert isinstance(response.json(), list) def test_api_create_item():    """Test POST /api/v1/items"""    payload = {"name": "Test Item", "description": "E2E test"}    response = requests.post(f"{BASE_URL}/api/v1/items", json=payload)    assert response.status_code == 201    data = response.json()    assert "id" in data    assert data["name"] == "Test Item" def test_api_get_item():    """Test GET /api/v1/items/{id}"""    # Create item first    create_response = requests.post(        f"{BASE_URL}/api/v1/items",        json={"name": "Test"}    )    item_id = create_response.json()["id"]     # Get item    response = requests.get(f"{BASE_URL}/api/v1/items/{item_id}")    assert response.status_code == 200    assert response.json()["id"] == item_id def test_api_update_item():    """Test PATCH /api/v1/items/{id}"""    # Create item    create_response = requests.post(        f"{BASE_URL}/api/v1/items",        json={"name": "Original"}    )    item_id = create_response.json()["id"]     # Update item    response = requests.patch(        f"{BASE_URL}/api/v1/items/{item_id}",        json={"name": "Updated"}    )    assert response.status_code == 200    assert response.json()["name"] == "Updated" def test_api_delete_item():    """Test DELETE /api/v1/items/{id}"""    # Create item    create_response = requests.post(        f"{BASE_URL}/api/v1/items",        json={"name": "To Delete"}    )    item_id = create_response.json()["id"]     # Delete item    response = requests.delete(f"{BASE_URL}/api/v1/items/{item_id}")    assert response.status_code == 204     # Verify deleted    response = requests.get(f"{BASE_URL}/api/v1/items/{item_id}")    assert response.status_code == 404

# test_e2e_performance.pyimport requestsimport timeimport statisticsfrom concurrent.futures import ThreadPoolExecutor BASE_URL = "http://localhost:8000" def test_sustained_throughput():    """Test sustained throughput of 1000 RPS"""    duration = 60  # Test for 60 seconds    response_times = []    errors = 0     def make_request():        try:            start = time.time()            response = requests.get(f"{BASE_URL}/health", timeout=5)            elapsed = (time.time() - start) * 1000  # ms             if response.status_code == 200:                response_times.append(elapsed)                return True            else:                return False        except:            return False     start_time = time.time()    requests_count = 0     with ThreadPoolExecutor(max_workers=100) as executor:        while time.time() - start_time < duration:            futures = [executor.submit(make_request) for _ in range(100)]             for future in futures:                if not future.result():                    errors += 1                requests_count += 1     # Calculate metrics    throughput = requests_count / duration    p95_latency = statistics.quantiles(response_times, n=20)[18] if response_times else 0    p99_latency = statistics.quantiles(response_times, n=100)[98] if response_times else 0    error_rate = (errors / requests_count) * 100 if requests_count > 0 else 0     assert throughput >= 1000, f"Throughput {throughput:.0f} RPS < 1000 RPS target"    assert p95_latency <= 200, f"P95 latency {p95_latency:.0f}ms > 200ms target"    assert error_rate < 0.1, f"Error rate {error_rate:.2f}% > 0.1% target" def test_response_time_slo():    """Verify p95 and p99 response times meet SLO"""    response_times = []     for _ in range(1000):        start = time.time()        response = requests.get(f"{BASE_URL}/api/v1/items")        elapsed = (time.time() - start) * 1000         assert response.status_code == 200        response_times.append(elapsed)     p95 = statistics.quantiles(response_times, n=20)[18]    p99 = statistics.quantiles(response_times, n=100)[98]     assert p95 <= 200, f"P95 {p95:.0f}ms > 200ms SLO"    assert p99 <= 500, f"P99 {p99:.0f}ms > 500ms SLO"

# Run all E2E tests./e2e-tests run \  --config e2e-config.yaml \  --image gcr.io/my-project/app:v1.2.3 \  --staging-cluster gke-staging \  --prod-cluster gke-prod-us-central1 \  --report e2e-results.html # Run specific test phase./e2e-tests run \  --config e2e-config.yaml \  --phase staging-deployment  # Just staging tests # Run with detailed logging./e2e-tests run \  --config e2e-config.yaml \  --verbose \  --log-file e2e.log

# .github/workflows/e2e.ymlname: End-to-End Pipeline Tests on:  push:    branches: [main] jobs:  e2e:    runs-on: ubuntu-latest    permissions:      id-token: write      contents: read     steps:      - uses: actions/checkout@v4       - name: Authenticate to GCP        uses: google-github-actions/auth@v1        with:          workload_identity_provider: ${{ secrets.WIF_PROVIDER }}          service_account: ${{ secrets.WIF_SERVICE_ACCOUNT }}       - name: Set up GKE kubeconfig        run: |          gcloud container clusters get-credentials gke-staging --region us-central1          gcloud container clusters get-credentials gke-prod-us-central1 --region us-central1       - name: Install E2E test framework        run: |          curl -sSL https://releases.cleanstart.dev/e2e-tests-latest.tar.gz | tar xz          chmod +x ./e2e-tests       - name: Run E2E Pipeline Tests        run: |          ./e2e-tests run \            --config e2e-config.yaml \            --image gcr.io/${{ env.GCP_PROJECT }}/app:${{ github.sha }} \            --report e2e-results.html        env:          GCP_PROJECT: ${{ secrets.GCP_PROJECT_ID }}       - name: Upload E2E Results        uses: actions/upload-artifact@v3        with:          name: e2e-results          path: e2e-results.html       - name: Comment on PR with Results        if: github.event_name == 'pull_request'        uses: actions/github-script@v7        with:          script: |            const fs = require('fs');            const results = fs.readFileSync('e2e-results.html', 'utf8');            github.rest.issues.createComment({              issue_number: context.issue.number,              owner: context.repo.owner,              repo: context.repo.repo,              body: `## E2E Pipeline Tests\n✅ All tests passed\n[View full report](${process.env.ARTIFACT_URL})`            });

# View E2E test dashboard./e2e-tests dashboard \  --config e2e-config.yaml \  --days 30 # Analyze failure trends./e2e-tests analyze-failures \  --config e2e-config.yaml \  --days 7 # Compare E2E performance baselines./e2e-tests compare-baselines \  --baseline ./baseline-v1.0.0.json \  --current ./current-results.json