๐ 4.4 CI/CD Pipeline Components
What is CI/CD?
CI/CD stands for Continuous Integration and Continuous Deployment/Delivery. It's a method to frequently deliver apps by introducing automation into development stages.
- Continuous Integration (CI): Automatically build and test code changes
- Continuous Delivery (CD): Automatically prepare code for release
- Continuous Deployment (CD): Automatically deploy to production
Key CI/CD Pipeline Components
1. Source Control Version Control System
Source control manages code versions and triggers the CI/CD pipeline.
Key Elements
- Git Repository: GitHub, GitLab, Bitbucket, Azure Repos
- Branching Strategy: Feature branches, main/master
- Webhooks: Trigger pipeline on push, merge, tag
- Pull Requests: Code review before merge
# .git workflow example # Developer makes changes git checkout -b feature/new-api git add . git commit -m "Add new API endpoint" git push origin feature/new-api # Create Pull Request (PR) # Code review happens # PR merged to main branch # Webhook triggers CI/CD pipeline automatically
2. Build Compile and Package
Build stage compiles source code, resolves dependencies, and creates executable artifacts.
Build Activities
- Dependency Resolution: Install required libraries (pip, npm, maven)
- Compilation: Compile code (for compiled languages)
- Linting: Check code quality (pylint, eslint)
- Static Analysis: Security scanning, code smell detection
- Artifact Creation: Create deployable package
# Example build steps (Python)
# Install dependencies
pip install -r requirements.txt
# Run linter
pylint src/
# Run security scan
bandit -r src/
# Build package
python setup.py sdist bdist_wheel
# Build Docker image
docker build -t myapp:${VERSION} .
3. Test Automated Testing
Testing stage runs automated tests to ensure code quality and functionality.
Test Types
| Test Type | Purpose | Tools |
|---|---|---|
| Unit Tests | Test individual functions/methods | unittest, pytest, Jest |
| Integration Tests | Test component interactions | pytest, Postman, Newman |
| End-to-End Tests | Test complete user workflows | Selenium, Cypress, Playwright |
| Performance Tests | Test speed and scalability | JMeter, Locust, k6 |
| Security Tests | Vulnerability scanning | OWASP ZAP, SonarQube |
# Example test execution # Run unit tests pytest tests/ -v --cov=src # Run integration tests pytest tests/integration/ -v # Generate coverage report pytest --cov=src --cov-report=html # If tests fail, pipeline stops # If tests pass, continue to next stage
4. Package/Artifact Storage Store Build Artifacts
Artifact storage saves build outputs for deployment.
Artifact Repositories
- Docker Registry: Docker Hub, Amazon ECR, Harbor
- Package Registry: PyPI, npm, Maven Central
- Artifact Storage: Artifactory, Nexus, Azure Artifacts
- Version Tagging: Semantic versioning (v1.2.3)
# Push Docker image to registry docker tag myapp:1.0 registry.example.com/myapp:1.0 docker push registry.example.com/myapp:1.0 # Tag with version docker tag myapp:1.0 registry.example.com/myapp:latest docker push registry.example.com/myapp:latest
5. Deployment Release to Environments
Deployment stage releases the application to target environments.
Deployment Strategies
| Strategy | Description | Risk |
|---|---|---|
| Rolling Deployment | Gradually replace instances | โ ๏ธ Medium |
| Blue/Green | Two identical environments, switch traffic | โ Low |
| Canary | Deploy to small subset first | โ Low |
| A/B Testing | Split traffic between versions | โ Low |
Environments
- Development (Dev): Developer testing
- Staging/QA: Pre-production testing
- Production (Prod): Live environment
6. Monitoring and Feedback
Monitoring tracks application performance and provides feedback to developers.
- Logging: ELK Stack, Splunk, CloudWatch
- Metrics: Prometheus, Grafana, Datadog
- Alerts: PagerDuty, Slack notifications
- APM: New Relic, AppDynamics
CI/CD Tools
| Tool | Type | Description |
|---|---|---|
| Jenkins | CI/CD Server | Open-source automation server |
| GitLab CI/CD | Integrated | Built into GitLab |
| GitHub Actions | Integrated | Built into GitHub |
| CircleCI | Cloud-based | SaaS CI/CD platform |
| Travis CI | Cloud-based | GitHub integration |
| Azure DevOps | Platform | Microsoft DevOps suite |
Example CI/CD Pipeline Configuration
# .gitlab-ci.yml - GitLab CI/CD Pipeline
stages:
- build
- test
- package
- deploy
variables:
APP_NAME: myapp
DOCKER_REGISTRY: registry.example.com
# Build Stage
build_job:
stage: build
script:
- echo "Building application..."
- pip install -r requirements.txt
- python -m py_compile src/*.py
artifacts:
paths:
- src/
expire_in: 1 hour
# Test Stage
test_job:
stage: test
script:
- echo "Running tests..."
- pip install -r requirements.txt
- pip install pytest pytest-cov
- pytest tests/ -v --cov=src --cov-report=term
coverage: '/TOTAL.*\s+(\d+%)$/'
# Package Stage
package_job:
stage: package
script:
- echo "Building Docker image..."
- docker build -t $DOCKER_REGISTRY/$APP_NAME:$CI_COMMIT_SHA .
- docker tag $DOCKER_REGISTRY/$APP_NAME:$CI_COMMIT_SHA $DOCKER_REGISTRY/$APP_NAME:latest
- docker push $DOCKER_REGISTRY/$APP_NAME:$CI_COMMIT_SHA
- docker push $DOCKER_REGISTRY/$APP_NAME:latest
only:
- main
# Deploy to Staging
deploy_staging:
stage: deploy
script:
- echo "Deploying to staging..."
- kubectl set image deployment/$APP_NAME $APP_NAME=$DOCKER_REGISTRY/$APP_NAME:$CI_COMMIT_SHA -n staging
- kubectl rollout status deployment/$APP_NAME -n staging
environment:
name: staging
url: https://staging.example.com
only:
- main
# Deploy to Production (manual trigger)
deploy_production:
stage: deploy
script:
- echo "Deploying to production..."
- kubectl set image deployment/$APP_NAME $APP_NAME=$DOCKER_REGISTRY/$APP_NAME:$CI_COMMIT_SHA -n production
- kubectl rollout status deployment/$APP_NAME -n production
environment:
name: production
url: https://example.com
when: manual
only:
- main
- Faster Delivery: Automated deployment reduces time to market
- Early Bug Detection: Tests run on every commit
- Reduced Risk: Small, frequent changes easier to troubleshoot
- Consistent Process: Same steps every time
- Better Collaboration: Shared pipeline and standards
- Rapid Rollback: Quick revert if issues occur
๐งช 4.5 Construct Python Unit Tests
What is Unit Testing?
Unit testing is testing individual units (functions, methods, classes) of code in isolation to ensure they work correctly.
- Test Case: Individual test for specific functionality
- Test Suite: Collection of test cases
- Assertion: Check if expected result matches actual
- Test Runner: Framework that executes tests
Python Testing Frameworks
| Framework | Style | Description |
|---|---|---|
| unittest | Built-in | Standard library, class-based tests |
| pytest | Third-party | Simple, powerful, fixture-based (most popular) |
| nose2 | Third-party | Extension of unittest |
Method 1: Using unittest (Built-in)
unittest Framework - Standard Library
Example: Testing a Calculator Class
#!/usr/bin/env python3
"""
Calculator Module - Code to Test
"""
class Calculator:
"""Simple calculator for demonstration"""
def add(self, a, b):
"""Add two numbers"""
return a + b
def subtract(self, a, b):
"""Subtract b from a"""
return a - b
def multiply(self, a, b):
"""Multiply two numbers"""
return a * b
def divide(self, a, b):
"""Divide a by b"""
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
def is_even(self, number):
"""Check if number is even"""
return number % 2 == 0
Unit Tests Using unittest
#!/usr/bin/env python3
"""
test_calculator.py - Unit Tests for Calculator
Using unittest framework
"""
import unittest
from calculator import Calculator
class TestCalculator(unittest.TestCase):
"""Test cases for Calculator class"""
def setUp(self):
"""Set up test fixtures - runs before each test"""
self.calc = Calculator()
print(f"\nRunning test: {self._testMethodName}")
def tearDown(self):
"""Clean up after each test"""
self.calc = None
# Test addition
def test_add_positive_numbers(self):
"""Test adding two positive numbers"""
result = self.calc.add(5, 3)
self.assertEqual(result, 8)
def test_add_negative_numbers(self):
"""Test adding negative numbers"""
result = self.calc.add(-5, -3)
self.assertEqual(result, -8)
def test_add_zero(self):
"""Test adding zero"""
result = self.calc.add(5, 0)
self.assertEqual(result, 5)
# Test subtraction
def test_subtract(self):
"""Test subtraction"""
result = self.calc.subtract(10, 3)
self.assertEqual(result, 7)
def test_subtract_negative_result(self):
"""Test subtraction with negative result"""
result = self.calc.subtract(3, 10)
self.assertEqual(result, -7)
# Test multiplication
def test_multiply(self):
"""Test multiplication"""
result = self.calc.multiply(4, 5)
self.assertEqual(result, 20)
def test_multiply_by_zero(self):
"""Test multiplication by zero"""
result = self.calc.multiply(5, 0)
self.assertEqual(result, 0)
# Test division
def test_divide(self):
"""Test normal division"""
result = self.calc.divide(10, 2)
self.assertEqual(result, 5.0)
def test_divide_by_zero(self):
"""Test that dividing by zero raises ValueError"""
with self.assertRaises(ValueError):
self.calc.divide(10, 0)
def test_divide_negative(self):
"""Test division with negative numbers"""
result = self.calc.divide(-10, 2)
self.assertEqual(result, -5.0)
# Test is_even
def test_is_even_true(self):
"""Test even number returns True"""
self.assertTrue(self.calc.is_even(4))
self.assertTrue(self.calc.is_even(0))
self.assertTrue(self.calc.is_even(-2))
def test_is_even_false(self):
"""Test odd number returns False"""
self.assertFalse(self.calc.is_even(3))
self.assertFalse(self.calc.is_even(-1))
# Run tests
if __name__ == '__main__':
# Run with verbose output
unittest.main(verbosity=2)
# Command line execution:
# python -m unittest test_calculator.py
# python -m unittest test_calculator.TestCalculator
# python -m unittest test_calculator.TestCalculator.test_add_positive_numbers
Common unittest Assertions
| Assertion | Purpose | Example |
|---|---|---|
assertEqual(a, b) |
Check a == b | self.assertEqual(result, 10) |
assertNotEqual(a, b) |
Check a != b | self.assertNotEqual(result, 0) |
assertTrue(x) |
Check x is True | self.assertTrue(is_valid) |
assertFalse(x) |
Check x is False | self.assertFalse(has_error) |
assertIsNone(x) |
Check x is None | self.assertIsNone(result) |
assertIn(a, b) |
Check a in b | self.assertIn('error', message) |
assertRaises(exc) |
Check exception raised | self.assertRaises(ValueError) |
Method 2: Using pytest (Popular Choice)
pytest Framework - Simple and Powerful
pip install pytest pytest-cov
#!/usr/bin/env python3
"""
test_calculator_pytest.py - Unit Tests Using pytest
"""
import pytest
from calculator import Calculator
# Fixture - reusable test setup
@pytest.fixture
def calc():
"""Create Calculator instance for tests"""
return Calculator()
# Test functions (no class needed with pytest)
def test_add_positive(calc):
"""Test adding positive numbers"""
assert calc.add(5, 3) == 8
def test_add_negative(calc):
"""Test adding negative numbers"""
assert calc.add(-5, -3) == -8
def test_subtract(calc):
"""Test subtraction"""
assert calc.subtract(10, 3) == 7
def test_multiply(calc):
"""Test multiplication"""
assert calc.multiply(4, 5) == 20
def test_divide(calc):
"""Test division"""
assert calc.divide(10, 2) == 5.0
def test_divide_by_zero(calc):
"""Test division by zero raises error"""
with pytest.raises(ValueError):
calc.divide(10, 0)
def test_is_even(calc):
"""Test is_even method"""
assert calc.is_even(4) == True
assert calc.is_even(3) == False
assert calc.is_even(0) == True
# Parametrized tests - run same test with different inputs
@pytest.mark.parametrize("a,b,expected", [
(5, 3, 8),
(10, 20, 30),
(-5, 5, 0),
(0, 0, 0),
])
def test_add_parametrized(calc, a, b, expected):
"""Test addition with multiple parameter sets"""
assert calc.add(a, b) == expected
@pytest.mark.parametrize("number,expected", [
(2, True),
(3, False),
(0, True),
(-2, True),
(-3, False),
])
def test_is_even_parametrized(calc, number, expected):
"""Test is_even with multiple values"""
assert calc.is_even(number) == expected
# Run tests with command:
# pytest test_calculator_pytest.py -v
# pytest test_calculator_pytest.py -v --cov=calculator
# pytest test_calculator_pytest.py -v --cov=calculator --cov-report=html
Testing REST API Functions
Unit Testing API Client Code
#!/usr/bin/env python3
"""
api_client.py - Simple API client to test
"""
import requests
class APIClient:
"""REST API client"""
def __init__(self, base_url):
self.base_url = base_url
def get_user(self, user_id):
"""Get user by ID"""
response = requests.get(f"{self.base_url}/users/{user_id}")
if response.status_code == 200:
return response.json()
elif response.status_code == 404:
return None
else:
response.raise_for_status()
def create_user(self, name, email):
"""Create new user"""
data = {"name": name, "email": email}
response = requests.post(f"{self.base_url}/users", json=data)
if response.status_code == 201:
return response.json()
else:
response.raise_for_status()
def is_valid_response(self, response):
"""Check if response is valid"""
return 200 <= response.status_code < 300
Unit Tests for API Client (with Mocking)
#!/usr/bin/env python3
"""
test_api_client.py - Unit tests with mocking
"""
import unittest
from unittest.mock import Mock, patch
from api_client import APIClient
class TestAPIClient(unittest.TestCase):
"""Test API client"""
def setUp(self):
"""Setup test client"""
self.client = APIClient("https://api.example.com")
@patch('api_client.requests.get')
def test_get_user_success(self, mock_get):
"""Test successful user retrieval"""
# Mock response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
'id': 1,
'name': 'John Doe',
'email': 'john@example.com'
}
mock_get.return_value = mock_response
# Execute
user = self.client.get_user(1)
# Assert
self.assertIsNotNone(user)
self.assertEqual(user['name'], 'John Doe')
self.assertEqual(user['email'], 'john@example.com')
# Verify API was called correctly
mock_get.assert_called_once_with(
"https://api.example.com/users/1"
)
@patch('api_client.requests.get')
def test_get_user_not_found(self, mock_get):
"""Test user not found (404)"""
mock_response = Mock()
mock_response.status_code = 404
mock_get.return_value = mock_response
user = self.client.get_user(999)
self.assertIsNone(user)
@patch('api_client.requests.post')
def test_create_user(self, mock_post):
"""Test user creation"""
mock_response = Mock()
mock_response.status_code = 201
mock_response.json.return_value = {
'id': 2,
'name': 'Jane Doe',
'email': 'jane@example.com'
}
mock_post.return_value = mock_response
user = self.client.create_user('Jane Doe', 'jane@example.com')
self.assertEqual(user['id'], 2)
self.assertEqual(user['name'], 'Jane Doe')
# Verify POST was called with correct data
mock_post.assert_called_once()
def test_is_valid_response(self):
"""Test response validation"""
# Test with mock objects
response_200 = Mock(status_code=200)
response_404 = Mock(status_code=404)
response_500 = Mock(status_code=500)
self.assertTrue(self.client.is_valid_response(response_200))
self.assertFalse(self.client.is_valid_response(response_404))
self.assertFalse(self.client.is_valid_response(response_500))
if __name__ == '__main__':
unittest.main(verbosity=2)
Testing Network Functions
Example: Testing IP Address Validation
#!/usr/bin/env python3
"""
network_utils.py - Network utility functions
"""
import ipaddress
def is_valid_ip(ip_string):
"""Check if string is valid IPv4 address"""
try:
ipaddress.IPv4Address(ip_string)
return True
except ValueError:
return False
def is_private_ip(ip_string):
"""Check if IP is in private range"""
try:
ip = ipaddress.IPv4Address(ip_string)
return ip.is_private
except ValueError:
return False
def get_network_info(ip_with_prefix):
"""Get network information from CIDR"""
try:
network = ipaddress.IPv4Network(ip_with_prefix, strict=False)
return {
'network': str(network.network_address),
'netmask': str(network.netmask),
'broadcast': str(network.broadcast_address),
'hosts': network.num_addresses - 2
}
except ValueError:
return None
Unit Tests for Network Functions
#!/usr/bin/env python3
"""
test_network_utils.py - Test network utility functions
"""
import unittest
from network_utils import is_valid_ip, is_private_ip, get_network_info
class TestNetworkUtils(unittest.TestCase):
"""Test network utility functions"""
# Test is_valid_ip
def test_valid_ip_addresses(self):
"""Test valid IP addresses"""
valid_ips = [
"192.168.1.1",
"10.0.0.1",
"8.8.8.8",
"255.255.255.255",
"0.0.0.0"
]
for ip in valid_ips:
with self.subTest(ip=ip):
self.assertTrue(is_valid_ip(ip))
def test_invalid_ip_addresses(self):
"""Test invalid IP addresses"""
invalid_ips = [
"256.1.1.1", # Out of range
"192.168.1", # Incomplete
"192.168.1.1.1", # Too many octets
"abc.def.ghi.jkl", # Non-numeric
"192.168.-1.1" # Negative
]
for ip in invalid_ips:
with self.subTest(ip=ip):
self.assertFalse(is_valid_ip(ip))
# Test is_private_ip
def test_private_ip_ranges(self):
"""Test private IP detection"""
private_ips = [
"192.168.1.1", # 192.168.0.0/16
"10.0.0.1", # 10.0.0.0/8
"172.16.0.1", # 172.16.0.0/12
]
for ip in private_ips:
with self.subTest(ip=ip):
self.assertTrue(is_private_ip(ip))
def test_public_ip_addresses(self):
"""Test public IP detection"""
public_ips = [
"8.8.8.8", # Google DNS
"1.1.1.1", # Cloudflare DNS
"4.2.2.2" # Level 3 DNS
]
for ip in public_ips:
with self.subTest(ip=ip):
self.assertFalse(is_private_ip(ip))
# Test get_network_info
def test_get_network_info(self):
"""Test network information calculation"""
result = get_network_info("192.168.1.0/24")
self.assertIsNotNone(result)
self.assertEqual(result['network'], "192.168.1.0")
self.assertEqual(result['netmask'], "255.255.255.0")
self.assertEqual(result['broadcast'], "192.168.1.255")
self.assertEqual(result['hosts'], 254)
def test_get_network_info_different_masks(self):
"""Test with different subnet masks"""
# /30 network
result = get_network_info("10.1.1.0/30")
self.assertEqual(result['hosts'], 2)
# /16 network
result = get_network_info("172.16.0.0/16")
self.assertEqual(result['hosts'], 65534)
def test_get_network_info_invalid(self):
"""Test with invalid CIDR"""
result = get_network_info("invalid")
self.assertIsNone(result)
if __name__ == '__main__':
unittest.main(verbosity=2)
Test Coverage and Best Practices
- Test One Thing: Each test should verify one specific behavior
- Independent Tests: Tests shouldn't depend on each other
- Descriptive Names: Test name should describe what it tests
- Arrange-Act-Assert: Setup, execute, verify pattern
- Test Edge Cases: Test boundary conditions and errors
- Use Mocking: Mock external dependencies (APIs, databases)
- Aim for Coverage: Target 80%+ code coverage
- Fast Execution: Unit tests should run quickly
Running Tests
# unittest - Run all tests in file python -m unittest test_calculator.py -v # unittest - Run specific test class python -m unittest test_calculator.TestCalculator -v # unittest - Run specific test method python -m unittest test_calculator.TestCalculator.test_add_positive_numbers # unittest - Discover and run all tests in directory python -m unittest discover -s tests/ -v # pytest - Run all tests pytest # pytest - Run with verbose output pytest -v # pytest - Run specific file pytest test_calculator.py -v # pytest - Run specific test pytest test_calculator.py::test_add_positive -v # pytest - Run with coverage report pytest --cov=src --cov-report=html # pytest - Run with coverage and show missing lines pytest --cov=src --cov-report=term-missing
- Red: Write a failing test first
- Green: Write minimal code to make test pass
- Refactor: Improve code while keeping tests passing
- Repeat: Continue cycle for each feature
๐ Interactive Python Interpreter
- Write Python code in the editor below
- Click "โถ Run Code" to execute
- Practice writing unit tests