Continuous Testing in DevOps: Quality Gates and CI/CD Integration

# .github/workflows/ci-cd.yml

name: CI/CD Pipeline with Continuous Testing

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  build-and-test:
    runs-on: ubuntu-latest

    steps:

    - name: Checkout code
      uses: actions/checkout@v3

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'

    - name: Install dependencies
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov pylint

    - name: Static Analysis
      run: |
        pylint src/ --fail-under=8.0
        echo "Static analysis passed"

    - name: Unit Tests with Coverage
      run: |
        pytest tests/unit --cov=src --cov-report=xml --cov-report=term
        echo "Unit tests passed"

    - name: Integration Tests
      run: |
        pytest tests/integration -v
        echo "Integration tests passed"

    - name: Build Docker Image
      run: |
        docker build -t myapp:${{ github.sha }} .

    - name: Deploy to Staging
      run: |
        # Deploy to staging environment
        ./scripts/deploy-staging.sh

    - name: Smoke Tests on Staging
      run: |
        pytest tests/smoke --base-url=https://staging.example.com
        echo "Smoke tests passed on staging"

    - name: E2E Tests
      run: |
        pytest tests/e2e --base-url=https://staging.example.com
        echo "E2E tests passed"

    - name: Security Scan
      uses: aquasecurity/trivy-action@master
      with:
        image-ref: myapp:${{ github.sha }}
        format: 'sarif'
        output: 'trivy-results.sarif'

    - name: Quality Gate Check
      run: |
        python scripts/quality-gate.py \
          --coverage-threshold=80 \
          --test-pass-rate=100
        echo "Quality gate passed"

    - name: Deploy to Production
      if: github.ref == 'refs/heads/main'
      run: |
        ./scripts/deploy-production.sh
        echo "Deployed to production"

# tests/unit/test_cart.py
import pytest
from src.shopping_cart import ShoppingCart, Product

class TestShoppingCart:
    def test_empty_cart_total_is_zero(self):
        cart = ShoppingCart()
        assert cart.total() == 0

    def test_add_single_product(self):
        cart = ShoppingCart()
        product = Product("Book", 15.99)
        cart.add(product)
        assert cart.total() == 15.99
        assert cart.item_count() == 1

    def test_add_multiple_products(self):
        cart = ShoppingCart()
        cart.add(Product("Book", 15.99))
        cart.add(Product("Pen", 2.50))
        assert cart.total() == 18.49
        assert cart.item_count() == 2

    def test_remove_product(self):
        cart = ShoppingCart()
        product = Product("Book", 15.99)
        cart.add(product)
        cart.remove(product)
        assert cart.total() == 0
        assert cart.item_count() == 0

    def test_apply_discount_code(self):
        cart = ShoppingCart()
        cart.add(Product("Book", 100.00))
        cart.apply_discount("SAVE20")  # 20% discount
        assert cart.total() == 80.00

# Run: pytest tests/unit -v
# Execution time: ~1 second for 5 tests

# tests/integration/test_checkout_api.py
import pytest
import requests

BASE_URL = "https://api.staging.example.com"

class TestCheckoutAPI:
    def test_complete_checkout_flow(self):
        # 1. Create cart
        response = requests.post(f"{BASE_URL}/api/cart", json={
            "user_id": "test_user_123"
        })
        assert response.status_code == 201
        cart_id = response.json()["cart_id"]

        # 2. Add products to cart
        response = requests.post(f"{BASE_URL}/api/cart/{cart_id}/items", json={
            "product_id": "BOOK001",
            "quantity": 2
        })
        assert response.status_code == 200

        # 3. Apply discount
        response = requests.post(f"{BASE_URL}/api/cart/{cart_id}/discount", json={
            "code": "SAVE20"
        })
        assert response.status_code == 200

        # 4. Get cart total
        response = requests.get(f"{BASE_URL}/api/cart/{cart_id}")
        assert response.status_code == 200
        cart_data = response.json()
        assert cart_data["total"] > 0
        assert cart_data["discount_applied"] == True

        # 5. Checkout
        response = requests.post(f"{BASE_URL}/api/checkout", json={
            "cart_id": cart_id,
            "payment_method": "credit_card",
            "card_token": "test_token_valid"
        })
        assert response.status_code == 200
        assert response.json()["order_status"] == "confirmed"

# Run: pytest tests/integration -v
# Execution time: ~10 seconds for full checkout flow test

# tests/e2e/test_user_journey.py
import pytest
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

class TestUserJourney:
    @pytest.fixture
    def browser(self):
        driver = webdriver.Chrome()
        driver.implicitly_wait(10)
        yield driver
        driver.quit()

    def test_complete_purchase_journey(self, browser):
        # 1. Navigate to homepage
        browser.get("https://staging.example.com")
        assert "Example Store" in browser.title

        # 2. Search for product
        search_box = browser.find_element(By.ID, "search")
        search_box.send_keys("Python Book")
        search_box.submit()

        # 3. Add product to cart
        wait = WebDriverWait(browser, 10)
        add_to_cart_btn = wait.until(
            EC.element_to_be_clickable((By.CSS_SELECTOR, ".add-to-cart"))
        )
        add_to_cart_btn.click()

        # 4. Verify cart count updated
        cart_count = browser.find_element(By.ID, "cart-count").text
        assert cart_count == "1"

        # 5. Proceed to checkout
        checkout_btn = browser.find_element(By.ID, "checkout")
        checkout_btn.click()

        # 6. Fill checkout form
        browser.find_element(By.ID, "email").send_keys("test@example.com")
        browser.find_element(By.ID, "card-number").send_keys("4111111111111111")
        browser.find_element(By.ID, "cvv").send_keys("123")

        # 7. Complete purchase
        browser.find_element(By.ID, "place-order").click()

        # 8. Verify order confirmation
        confirmation = wait.until(
            EC.presence_of_element_located((By.CLASS_NAME, "order-confirmation"))
        )
        assert "Thank you for your order" in confirmation.text

# Run: pytest tests/e2e -v --headless
# Execution time: ~45 seconds for complete user journey

# scripts/quality-gate.py
import sys
import json
import argparse

class QualityGate:
    def __init__(self, config):
        self.config = config
        self.checks_passed = []
        self.checks_failed = []

    def check_code_coverage(self, coverage_file):
        """Check if code coverage meets threshold"""
        with open(coverage_file, 'r') as f:
            coverage_data = json.load(f)
            coverage_percent = coverage_data['totals']['percent_covered']

        threshold = self.config['coverage_threshold']
        if coverage_percent >= threshold:
            self.checks_passed.append(f"Coverage: {coverage_percent}% (>= {threshold}%)")
            return True
        else:
            self.checks_failed.append(f"Coverage: {coverage_percent}% (< {threshold}%)")
            return False

    def check_test_results(self, test_results_file):
        """Check if all tests passed"""
        with open(test_results_file, 'r') as f:
            results = json.load(f)

        total = results['total']
        passed = results['passed']
        pass_rate = (passed / total) * 100

        required_rate = self.config['test_pass_rate']
        if pass_rate >= required_rate:
            self.checks_passed.append(f"Test Pass Rate: {pass_rate}% ({passed}/{total})")
            return True
        else:
            self.checks_failed.append(f"Test Pass Rate: {pass_rate}% (< {required_rate}%)")
            return False

    def check_security_vulnerabilities(self, security_report):
        """Check for critical security vulnerabilities"""
        with open(security_report, 'r') as f:
            vulns = json.load(f)

        critical = vulns.get('critical', 0)
        high = vulns.get('high', 0)

        if critical == 0 and high == 0:
            self.checks_passed.append(f"Security: No critical/high vulnerabilities")
            return True
        else:
            self.checks_failed.append(f"Security: {critical} critical, {high} high vulnerabilities")
            return False

    def check_code_quality(self, quality_report):
        """Check static analysis score"""
        with open(quality_report, 'r') as f:
            quality = json.load(f)

        score = quality['overall_score']
        threshold = self.config['quality_score_threshold']

        if score >= threshold:
            self.checks_passed.append(f"Code Quality: {score}/10 (>= {threshold})")
            return True
        else:
            self.checks_failed.append(f"Code Quality: {score}/10 (< {threshold})")
            return False

    def evaluate(self):
        """Run all quality gate checks"""
        print("=" * 60)
        print("Quality Gate Evaluation")
        print("=" * 60)

        all_checks = [
            self.check_code_coverage('coverage.json'),
            self.check_test_results('test-results.json'),
            self.check_security_vulnerabilities('security-report.json'),
            self.check_code_quality('quality-report.json')
        ]

        print("\n✓ Passed Checks:")
        for check in self.checks_passed:
            print(f"  - {check}")

        if self.checks_failed:
            print("\n✗ Failed Checks:")
            for check in self.checks_failed:
                print(f"  - {check}")

        passed = all(all_checks)
        print("\n" + "=" * 60)
        if passed:
            print("QUALITY GATE: PASSED ✓")
            print("=" * 60)
            return 0
        else:
            print("QUALITY GATE: FAILED ✗")
            print("=" * 60)
            return 1

# Configuration
config = {
    'coverage_threshold': 80,      # Minimum 80% code coverage
    'test_pass_rate': 100,         # All tests must pass
    'quality_score_threshold': 7.5, # Minimum 7.5/10 quality score
}

if __name__ == '__main__':
    gate = QualityGate(config)
    sys.exit(gate.evaluate())

# scripts/notify-slack.py
import requests
import sys
import json

def send_slack_notification(webhook_url, message, status):
    """Send formatted test results to Slack"""

    color = "#36a64f" if status == "success" else "#ff0000"
    icon = ":white_check_mark:" if status == "success" else ":x:"

    payload = {
        "attachments": [
            {
                "color": color,
                "title": f"{icon} CI/CD Pipeline {status.upper()}",
                "fields": [
                    {
                        "title": "Branch",
                        "value": message['branch'],
                        "short": True
                    },
                    {
                        "title": "Commit",
                        "value": message['commit'][:8],
                        "short": True
                    },
                    {
                        "title": "Test Results",
                        "value": f"Passed: {message['tests_passed']}/{message['tests_total']}",
                        "short": True
                    },
                    {
                        "title": "Coverage",
                        "value": f"{message['coverage']}%",
                        "short": True
                    }
                ],
                "footer": "CI/CD Pipeline",
                "ts": message['timestamp']
            }
        ]
    }

    response = requests.post(webhook_url, json=payload)
    return response.status_code == 200

# Usage in CI/CD pipeline
message = {
    "branch": "main",
    "commit": "a7b3c4d5e6f7g8h9",
    "tests_passed": 487,
    "tests_total": 500,
    "coverage": 85.2,
    "timestamp": 1672531200
}

send_slack_notification(
    "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
    message,
    "success"  # or "failure"
)

## Test Speed Guidelines

- Unit tests: < 5 seconds total
- Integration tests: < 2 minutes total
- E2E tests: < 30 minutes total
- Full pipeline: < 45 minutes total

**Strategies:**

- Run tests in parallel
- Mock external dependencies
- Use test containers for isolation
- Cache dependencies
- Optimize slow tests

## Reducing Test Flakiness

**Common causes:**

- Race conditions and timing issues
- Test data dependencies
- External service instability
- Environment differences

**Solutions:**

- Explicit waits instead of sleeps
- Isolated test data per test
- Mock external services
- Use containers for consistency
- Retry flaky tests automatically (max 2 retries)

# Good test: Clear, focused, maintainable

def test_user_can_checkout_with_discount():
    # Arrange: Set up test data
    cart = create_cart_with_products([
        Product("Book", 50.00),
        Product("Pen", 5.00)
    ])
    discount = DiscountCode("SAVE20", percentage=20)

    # Act: Perform action
    final_total = cart.apply_discount(discount)

    # Assert: Verify outcome
    assert final_total == 44.00  # (50 + 5) * 0.8
    assert cart.discount_applied == True
    assert cart.discount_amount == 11.00

## Progressive Test Execution

1. **Commit Stage** (2-5 min)
   - Unit tests
   - Static analysis
   - Fast, immediate feedback

2. **Acceptance Stage** (10-20 min)
   - Integration tests
   - API contract tests
   - Component interaction validation

3. **Staging Stage** (20-45 min)
   - E2E tests
   - Performance tests (subset)
   - Security scans

4. **Production Stage** (Continuous)
   - Smoke tests post-deployment
   - Synthetic monitoring
   - Production health checks