Skip to content

Latest commit

 

History

History
1059 lines (852 loc) · 27.3 KB

File metadata and controls

1059 lines (852 loc) · 27.3 KB

CI/CD Testing Integration 🚀

Integrating testing into Continuous Integration and Continuous Deployment (CI/CD) pipelines is essential for maintaining code quality and ensuring reliable deployments in DevOps workflows.

Table of Contents

CI/CD Testing Overview

The Testing Pyramid in CI/CD

        /\
       /  \
      /E2E \     <- End-to-End Tests (Few, Slow, Expensive)
     /______\
    /        \
   /Integration\ <- Integration Tests (Some, Medium Speed)
  /____________\
 /              \
/   Unit Tests   \ <- Unit Tests (Many, Fast, Cheap)
/________________\

Testing Phases in CI/CD Pipeline

  1. Pre-commit: Local testing, linting, formatting
  2. Commit: Unit tests, static analysis
  3. Build: Integration tests, security scanning
  4. Deploy: End-to-end tests, smoke tests
  5. Post-deploy: Monitoring, health checks

GitHub Actions Integration

Basic Testing Workflow

Create .github/workflows/test.yml:

name: Test Suite

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

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: [3.8, 3.9, 3.10, 3.11]

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

      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-test.txt

      - name: Lint with flake8
        run: |
          # Stop build if there are Python syntax errors or undefined names
          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
          # Exit-zero treats all errors as warnings
          flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics

      - name: Test with pytest
        run: |
          pytest tests/ --cov=src --cov-report=xml --cov-report=html

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml
          flags: unittests
          name: codecov-umbrella

Advanced Testing Workflow with Services

name: Advanced Test Suite

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:13
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: test_db
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

      redis:
        image: redis:6
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 6379:6379

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

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

      - name: Cache dependencies
        uses: actions/cache@v3
        with:
          path: ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements*.txt') }}
          restore-keys: |
            ${{ runner.os }}-pip-

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-test.txt

      - name: Wait for services
        run: |
          until pg_isready -h localhost -p 5432; do sleep 1; done
          until redis-cli -h localhost -p 6379 ping; do sleep 1; done

      - name: Run database migrations
        run: |
          python manage.py migrate
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0

      - name: Run unit tests
        run: |
          pytest tests/unit/ -v --cov=src --cov-branch
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0

      - name: Run integration tests
        run: |
          pytest tests/integration/ -v --cov=src --cov-append
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0

      - name: Generate coverage report
        run: |
          coverage html
          coverage xml

      - name: Upload coverage reports
        uses: actions/upload-artifact@v3
        with:
          name: coverage-report
          path: htmlcov/

Multi-Stage Pipeline with Quality Gates

name: Multi-Stage Pipeline

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

jobs:
  lint:
    name: Code Quality Check
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: 3.11

      - name: Install linting tools
        run: |
          pip install flake8 black isort mypy

      - name: Check formatting with black
        run: black --check .

      - name: Check imports with isort
        run: isort --check-only .

      - name: Lint with flake8
        run: flake8 .

      - name: Type check with mypy
        run: mypy src/

  unit-tests:
    name: Unit Tests
    runs-on: ubuntu-latest
    needs: lint
    steps:
      - 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 -r requirements-test.txt

      - name: Run unit tests
        run: |
          pytest tests/unit/ --cov=src --cov-report=xml --cov-fail-under=80

      - name: Upload unit test results
        uses: actions/upload-artifact@v3
        if: always()
        with:
          name: unit-test-results
          path: coverage.xml

  integration-tests:
    name: Integration Tests
    runs-on: ubuntu-latest
    needs: unit-tests
    services:
      postgres:
        image: postgres:13
        env:
          POSTGRES_PASSWORD: postgres
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - 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 -r requirements-test.txt

      - name: Run integration tests
        run: |
          pytest tests/integration/ -v
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres

  security-scan:
    name: Security Scan
    runs-on: ubuntu-latest
    needs: lint
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: 3.11

      - name: Install safety
        run: pip install safety bandit

      - name: Check for vulnerabilities
        run: |
          safety check
          bandit -r src/

  build-and-test:
    name: Build and Test Container
    runs-on: ubuntu-latest
    needs: [unit-tests, integration-tests, security-scan]
    steps:
      - uses: actions/checkout@v3

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

      - name: Run container tests
        run: |
          # Start container
          docker run -d --name test-container -p 8080:8080 myapp:${{ github.sha }}

          # Wait for container to be ready
          sleep 30

          # Run health check
          curl -f http://localhost:8080/health || exit 1

          # Stop container
          docker stop test-container

Test Automation Strategies

Test Categories and Execution Strategy

# pytest.ini
[tool:pytest]
testpaths = tests
markers =
    unit: Unit tests (fast, isolated)
    integration: Integration tests (medium speed, external dependencies)
    e2e: End-to-end tests (slow, full system)
    smoke: Smoke tests (quick validation after deployment)
    security: Security tests
    performance: Performance tests

addopts =
    --strict-markers
    --tb=short
    --cov=src
    --cov-branch
    --cov-report=html
    --cov-report=xml
    --cov-report=term-missing
# conftest.py - Shared test configuration
import pytest
import os
from unittest.mock import Mock

def pytest_configure(config):
    """Configure pytest with custom markers and settings."""
    config.addinivalue_line("markers", "unit: Unit tests")
    config.addinivalue_line("markers", "integration: Integration tests")
    config.addinivalue_line("markers", "e2e: End-to-end tests")
    config.addinivalue_line("markers", "smoke: Smoke tests")
    config.addinivalue_line("markers", "slow: Slow running tests")

@pytest.fixture(scope="session")
def test_environment():
    """Determine test environment and configuration."""
    env = os.getenv("TEST_ENV", "local")
    return {
        "environment": env,
        "database_url": os.getenv("DATABASE_URL", "sqlite:///:memory:"),
        "redis_url": os.getenv("REDIS_URL", "redis://localhost:6379/0"),
        "api_base_url": os.getenv("API_BASE_URL", "http://localhost:8000")
    }

@pytest.fixture
def mock_external_api():
    """Mock external API calls for unit tests."""
    mock_api = Mock()
    mock_api.get_data.return_value = {"status": "success", "data": []}
    return mock_api
# tests/test_smoke.py - Smoke tests for critical functionality
import pytest
import requests

@pytest.mark.smoke
def test_application_health_endpoint(test_environment):
    """Test that application health endpoint responds correctly."""
    health_url = f"{test_environment['api_base_url']}/health"
    response = requests.get(health_url, timeout=10)

    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

@pytest.mark.smoke
def test_database_connection(test_environment):
    """Test that database connection is working."""
    # This would test actual database connectivity
    # Implementation depends on your database setup
    pass

@pytest.mark.smoke
def test_critical_user_workflow():
    """Test the most critical user workflow."""
    # Test the most important user journey
    pass

Conditional Test Execution

# tests/test_conditional.py
import pytest
import os
import sys

# Skip tests based on environment
@pytest.mark.skipif(
    os.getenv("CI") != "true",
    reason="Integration tests only run in CI environment"
)
def test_ci_only_functionality():
    """Test that only runs in CI environment."""
    pass

# Skip tests based on Python version
@pytest.mark.skipif(
    sys.version_info < (3, 9),
    reason="Requires Python 3.9 or higher"
)
def test_modern_python_feature():
    """Test that requires newer Python features."""
    pass

# Skip tests based on available services
@pytest.mark.skipif(
    not os.getenv("DATABASE_URL"),
    reason="Database not available"
)
def test_database_functionality():
    """Test that requires database connection."""
    pass

# Expected failure (for known issues)
@pytest.mark.xfail(reason="Known issue with external API")
def test_known_failing_feature():
    """Test that is expected to fail due to known issue."""
    pass

Quality Gates

Coverage Requirements

# setup.cfg or pyproject.toml
[tool:pytest]
addopts = --cov-fail-under=85

# Or in pytest command
# pytest --cov=src --cov-fail-under=85
# In GitHub Actions
- name: Run tests with coverage requirement
  run: |
    pytest --cov=src --cov-fail-under=80 --cov-report=xml

- name: Coverage comment
  uses: py-cov-action/python-coverage-comment-action@v3
  with:
    GITHUB_TOKEN: ${{ github.token }}
    MINIMUM_GREEN: 80
    MINIMUM_ORANGE: 70

Performance Thresholds

# tests/test_performance.py
import pytest
import time
from performance_monitor import measure_time

@pytest.mark.performance
def test_api_response_time():
    """Ensure API responses are under acceptable threshold."""
    with measure_time() as timer:
        # Make API call
        response = make_api_request()

    assert timer.elapsed < 2.0  # 2 second threshold
    assert response.status_code == 200

@pytest.mark.performance
def test_database_query_performance():
    """Ensure database queries are optimized."""
    start_time = time.time()

    # Execute database query
    results = execute_complex_query()

    end_time = time.time()
    query_time = end_time - start_time

    assert query_time < 0.5  # 500ms threshold
    assert len(results) > 0

Security Quality Gates

# tests/test_security.py
import pytest
import requests
from security_scanner import scan_for_vulnerabilities

@pytest.mark.security
def test_no_sensitive_data_in_logs():
    """Ensure no sensitive data is logged."""
    # Trigger logging
    perform_user_login("test@example.com", "password123")

    # Check logs don't contain sensitive data
    with open("app.log", "r") as f:
        log_content = f.read()

    assert "password123" not in log_content
    assert "secret_key" not in log_content

@pytest.mark.security
def test_dependency_vulnerabilities():
    """Check for known vulnerabilities in dependencies."""
    vulnerabilities = scan_for_vulnerabilities()

    # Fail if high severity vulnerabilities found
    high_severity = [v for v in vulnerabilities if v.severity == "high"]
    assert len(high_severity) == 0, f"High severity vulnerabilities found: {high_severity}"

@pytest.mark.security
def test_api_rate_limiting():
    """Test that API rate limiting is enforced."""
    api_url = "http://localhost:8000/api/data"

    # Make rapid requests
    responses = []
    for i in range(20):
        response = requests.get(api_url)
        responses.append(response)

    # Should get rate limited
    rate_limited = [r for r in responses if r.status_code == 429]
    assert len(rate_limited) > 0, "Rate limiting not working"

Test Reporting

JUnit XML Reports

# Generate JUnit XML for CI/CD systems
pytest --junit-xml=test-results.xml
# GitHub Actions - Upload test results
- name: Run tests
  run: |
    pytest --junit-xml=test-results.xml

- name: Publish test results
  uses: EnricoMi/publish-unit-test-result-action@v2
  if: always()
  with:
    files: test-results.xml

HTML Reports

# pytest-html plugin
# pip install pytest-html

# Generate HTML report
# pytest --html=report.html --self-contained-html

Custom Test Reporting

# tests/conftest.py
import pytest
import json
from datetime import datetime

class TestResultCollector:
    def __init__(self):
        self.results = []

    def add_result(self, test_name, outcome, duration, error=None):
        self.results.append({
            "test_name": test_name,
            "outcome": outcome,
            "duration": duration,
            "timestamp": datetime.now().isoformat(),
            "error": str(error) if error else None
        })

    def save_results(self, filename):
        with open(filename, 'w') as f:
            json.dump(self.results, f, indent=2)

@pytest.fixture(scope="session")
def test_collector():
    return TestResultCollector()

@pytest.hookimpl(tryfirst=True)
def pytest_runtest_makereport(item, call):
    """Hook to collect test results."""
    if call.when == "call":
        test_collector = item.session.config.getoption("--test-collector", None)
        if test_collector:
            outcome = "passed" if call.excinfo is None else "failed"
            duration = call.duration
            error = call.excinfo if call.excinfo else None

            # This would need to be properly integrated with pytest
            # test_collector.add_result(item.name, outcome, duration, error)

def pytest_sessionfinish(session, exitstatus):
    """Save test results at end of session."""
    # Save custom test results
    pass

Docker Testing

Testing Containerized Applications

# Dockerfile.test - Multi-stage build for testing
FROM python:3.11-slim as base

WORKDIR /app

# Install dependencies
COPY requirements*.txt ./
RUN pip install -r requirements.txt
RUN pip install -r requirements-test.txt

# Copy source code
COPY src/ ./src/
COPY tests/ ./tests/

# Test stage
FROM base as test
CMD ["pytest", "tests/", "--cov=src", "--cov-report=xml", "--junit-xml=test-results.xml"]

# Production stage
FROM base as production
CMD ["python", "-m", "src.main"]
# Build and run tests in container
docker build --target test -t myapp:test .
docker run --rm -v $(pwd)/reports:/app/reports myapp:test

# Test production container
docker build --target production -t myapp:prod .
docker run -d --name test-container -p 8080:8080 myapp:prod

# Run health checks
curl -f http://localhost:8080/health

# Cleanup
docker stop test-container
docker rm test-container

Docker Compose Testing

# docker-compose.test.yml
version: "3.8"

services:
  app:
    build:
      context: .
      target: test
    depends_on:
      - postgres
      - redis
    environment:
      - DATABASE_URL=postgresql://user:password@postgres:5432/testdb
      - REDIS_URL=redis://redis:6379/0
    volumes:
      - ./reports:/app/reports

  postgres:
    image: postgres:13
    environment:
      POSTGRES_DB: testdb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    tmpfs:
      - /var/lib/postgresql/data

  redis:
    image: redis:6
    tmpfs:
      - /data
# Run tests with Docker Compose
docker-compose -f docker-compose.test.yml up --build --abort-on-container-exit
docker-compose -f docker-compose.test.yml down

Infrastructure Testing

Testing Infrastructure as Code

# tests/test_infrastructure.py
import pytest
import boto3
from moto import mock_ec2, mock_s3
import yaml

@mock_ec2
def test_ec2_instance_creation():
    """Test EC2 instance creation with proper configuration."""
    ec2 = boto3.client('ec2', region_name='us-east-1')

    # Create instance
    response = ec2.run_instances(
        ImageId='ami-12345678',
        MinCount=1,
        MaxCount=1,
        InstanceType='t2.micro',
        SecurityGroupIds=['sg-12345678'],
        TagSpecifications=[
            {
                'ResourceType': 'instance',
                'Tags': [
                    {'Key': 'Environment', 'Value': 'test'},
                    {'Key': 'Application', 'Value': 'myapp'}
                ]
            }
        ]
    )

    instance_id = response['Instances'][0]['InstanceId']

    # Verify instance configuration
    instances = ec2.describe_instances(InstanceIds=[instance_id])
    instance = instances['Reservations'][0]['Instances'][0]

    assert instance['InstanceType'] == 't2.micro'
    assert instance['State']['Name'] == 'running'

    # Verify tags
    tags = {tag['Key']: tag['Value'] for tag in instance.get('Tags', [])}
    assert tags['Environment'] == 'test'
    assert tags['Application'] == 'myapp'

def test_kubernetes_deployment_yaml():
    """Test Kubernetes deployment configuration."""
    with open('k8s/deployment.yaml', 'r') as f:
        deployment = yaml.safe_load(f)

    # Validate deployment structure
    assert deployment['kind'] == 'Deployment'
    assert deployment['metadata']['name'] == 'myapp'

    # Validate container configuration
    containers = deployment['spec']['template']['spec']['containers']
    app_container = next(c for c in containers if c['name'] == 'myapp')

    assert app_container['image'].startswith('myapp:')
    assert app_container['ports'][0]['containerPort'] == 8080

    # Validate resource limits
    resources = app_container.get('resources', {})
    assert 'limits' in resources
    assert 'requests' in resources

Testing Ansible Playbooks

# tests/test_ansible.py
import pytest
import ansible_runner
import tempfile
import os

def test_ansible_playbook_syntax():
    """Test Ansible playbook syntax is valid."""
    result = ansible_runner.run(
        playbook='playbooks/deploy.yml',
        inventory='inventory/test',
        check=True,  # Dry run mode
        verbosity=1
    )

    assert result.status == 'successful'
    assert result.rc == 0

def test_ansible_playbook_execution():
    """Test Ansible playbook execution in test environment."""
    with tempfile.TemporaryDirectory() as temp_dir:
        # Create test inventory
        inventory_content = """
        [test]
        test-server ansible_host=localhost ansible_connection=local
        """

        inventory_path = os.path.join(temp_dir, 'inventory')
        with open(inventory_path, 'w') as f:
            f.write(inventory_content)

        # Run playbook
        result = ansible_runner.run(
            playbook='playbooks/setup.yml',
            inventory=inventory_path,
            extravars={'target_env': 'test'}
        )

        assert result.status == 'successful'

        # Verify specific tasks completed
        for event in result.events:
            if event['event'] == 'runner_on_ok':
                # Check that required tasks completed successfully
                pass

Deployment Testing

Blue-Green Deployment Testing

# tests/test_deployment.py
import pytest
import requests
import time

class TestBlueGreenDeployment:
    def test_green_environment_health(self):
        """Test that green environment is healthy before switch."""
        green_url = "http://green.example.com/health"
        response = requests.get(green_url, timeout=10)

        assert response.status_code == 200
        assert response.json()["status"] == "healthy"

    def test_traffic_switch(self):
        """Test traffic switching between blue and green."""
        # Test current active environment
        active_url = "http://api.example.com/version"
        initial_response = requests.get(active_url)
        initial_version = initial_response.json()["version"]

        # Perform traffic switch (this would be done by deployment script)
        # switch_traffic_to_green()

        # Verify new version is active
        time.sleep(5)  # Allow for DNS propagation
        new_response = requests.get(active_url)
        new_version = new_response.json()["version"]

        assert new_version != initial_version

    def test_rollback_capability(self):
        """Test ability to rollback to previous version."""
        # Get current version
        active_url = "http://api.example.com/version"
        current_response = requests.get(active_url)
        current_version = current_response.json()["version"]

        # Simulate rollback
        # rollback_to_previous_version()

        # Verify rollback worked
        time.sleep(5)
        rollback_response = requests.get(active_url)
        rollback_version = rollback_response.json()["version"]

        # Should be different from current (back to previous)
        assert rollback_version != current_version

Canary Deployment Testing

# tests/test_canary.py
import pytest
import requests
import statistics

class TestCanaryDeployment:
    def test_canary_traffic_distribution(self):
        """Test that canary receives appropriate traffic percentage."""
        api_url = "http://api.example.com/version"
        versions = []

        # Make multiple requests to check traffic distribution
        for _ in range(100):
            response = requests.get(api_url)
            version = response.json()["version"]
            versions.append(version)

        # Count version distribution
        version_counts = {}
        for version in versions:
            version_counts[version] = version_counts.get(version, 0) + 1

        # Canary should receive ~10% of traffic
        canary_percentage = (version_counts.get("v2.0.0", 0) / len(versions)) * 100
        assert 5 <= canary_percentage <= 15  # Allow for some variance

    def test_canary_performance_metrics(self):
        """Test that canary version meets performance requirements."""
        canary_url = "http://canary.example.com/api/heavy-operation"
        response_times = []

        # Collect performance metrics
        for _ in range(10):
            start_time = time.time()
            response = requests.get(canary_url)
            end_time = time.time()

            assert response.status_code == 200
            response_times.append(end_time - start_time)

        # Verify performance is acceptable
        avg_response_time = statistics.mean(response_times)
        max_response_time = max(response_times)

        assert avg_response_time < 2.0  # Average under 2 seconds
        assert max_response_time < 5.0  # Max under 5 seconds

Best Practices for CI/CD Testing

1. Test Pyramid Implementation

  • Unit Tests (70%): Fast, isolated, mock dependencies
  • Integration Tests (20%): Medium speed, test component interactions
  • E2E Tests (10%): Slow, test complete user workflows

2. Fail Fast Principle

# Run fast tests first
jobs:
  lint: # Fastest - catches syntax/style issues
  unit: # Fast - catches logic issues
  integration: # Medium - catches integration issues
  e2e: # Slowest - catches workflow issues

3. Test Environment Management

# Environment-specific test configuration
@pytest.fixture
def test_config():
    env = os.getenv("TEST_ENV", "local")

    configs = {
        "local": {
            "database_url": "sqlite:///:memory:",
            "external_api_url": "http://localhost:8000"
        },
        "ci": {
            "database_url": "postgresql://postgres:postgres@postgres:5432/test",
            "external_api_url": "https://staging-api.example.com"
        },
        "staging": {
            "database_url": os.getenv("STAGING_DATABASE_URL"),
            "external_api_url": "https://staging-api.example.com"
        }
    }

    return configs[env]

4. Test Data Management

# Fixture for test data setup/teardown
@pytest.fixture(scope="function")
def clean_database():
    """Clean database before each test."""
    # Setup
    db.create_all()
    yield db
    # Teardown
    db.drop_all()

@pytest.fixture
def sample_users():
    """Provide consistent test data."""
    return [
        {"name": "Alice", "email": "alice@test.com"},
        {"name": "Bob", "email": "bob@test.com"}
    ]

5. Monitoring and Alerting

# Add monitoring to CI/CD pipeline
- name: Monitor test trends
  run: |
    python scripts/analyze_test_trends.py

- name: Alert on test failures
  if: failure()
  uses: 8398a7/action-slack@v3
  with:
    status: failure
    channel: "#dev-alerts"

Next Steps

Continue your CI/CD testing journey with:

  1. Security Testing Integration - SAST, DAST, dependency scanning
  2. Performance Testing in CI/CD - Load testing, benchmarking
  3. Contract Testing - API contract validation between services
  4. Chaos Engineering - Testing system resilience

Effective CI/CD testing ensures reliable, fast, and confident deployments while maintaining high code quality.