✅ Step-by-Step: Run All Tests & Get Coverage
🧱 Project Structure Example
my_project/
├── my_app/ # Your actual code (module/package)
│ ├── __init__.py
│ ├── s3_service.py
│ └── db_service.py
├── tests/ # Your test files
│ ├── __init__.py
│ ├── test_s3_service.py
│ └── test_db_service.py
✅ Run All Tests with Coverage:
pytest --cov=my_app --cov-report=term --cov-report=html
✅ What This Does:
Part | Purpose |
--cov=my_app | Tells pytest to measure test coverage for all code inside my_app/ |
--cov-report=term | Shows summary in the terminal |
--cov-report=html | Generates a full visual report in htmlcov/index.html |
✅ Install if Needed:
pip install pytest pytest-cov
Tool | Purpose |
pytest | Python testing framework |
patch() | Replace real AWS calls with mock objects |
MagicMock | Simulate AWS response data |
pytest-cov | Shows test coverage |
+++++++++++++++++++++++
Scenario: Internal API Validation
You’re writing a function that:
Calls an internal user validation service (http://internal.service/api/user/<user_id>)
The response returns user info like:
{ "id": 101, "status": "active", "role": "admin" }
Your function should:
✅ Return True if the user is active and role is admin
❌ Return False otherwise
🔧 Python Code (e.g. user_check.py)
import requests
def is_valid_admin(user_id):
url = f"http://internal.service/api/user/{user_id}"
response = requests.get(url)
if response.status_code != 200:
return False
data = response.json()
return data.get("status") == "active" and data.get("role") == "admin"
🧪 Unit Tests with Mock (e.g. test_user_check.py)
from user_check import is_valid_admin
from unittest.mock import patch
# ✅ Case 1: Valid active admin user
def test_valid_admin_user():
with patch('user_check.requests.get') as mock_get:
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"id": 101,
"status": "active",
"role": "admin"
}
assert is_valid_admin(101) is True
# ❌ Case 2: Inactive user
def test_inactive_user():
with patch('user_check.requests.get') as mock_get:
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"id": 102,
"status": "inactive",
"role": "admin"
}
assert is_valid_admin(102) is False
# ❌ Case 3: Wrong role
def test_wrong_role_user():
with patch('user_check.requests.get') as mock_get:
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"id": 103,
"status": "active",
"role": "user"
}
assert is_valid_admin(103) is False
# ❌ Case 4: API error (non-200 response)
def test_api_error():
with patch('user_check.requests.get') as mock_get:
mock_get.return_value.status_code = 500
assert is_valid_admin(104) is False
✅ What You’ve Covered
Scenario | What It Tests |
Active admin | ✅ Valid user, returns True |
Inactive user | ❌ Invalid status, returns False |
Wrong role | ❌ Invalid role, returns False |
API failure | ❌ API error, returns False |
🧪 Run the Tests
Install pytest if needed:
bash
pip install pytest
Run tests:
bash
pytest test_user_check.py
++++++++++++++++++++++++++++++++
Imagine this function in db_service.py:
def get_user_name(user_id, db):
user = db.query(User).filter(User.id == user_id).first()
if user:
return user.name
return None
db is a database session object passed in.
You want to test this function without a real database.
Simple test example (test_db_service.py):
from unittest.mock import MagicMock
from db_service import get_user_name
def test_get_user_name_found():
fake_db = MagicMock()
# Chain the mocks: db.query().filter().first()
fake_db.query.return_value.filter.return_value.first.return_value.name = "Alice"
result = get_user_name(1, fake_db)
assert result == "Alice"
def test_get_user_name_not_found():
fake_db = MagicMock()
# Chain returns None on first() to simulate user not found
fake_db.query.return_value.filter.return_value.first.return_value = None
result = get_user_name(2, fake_db)
assert result is None
Explanation
fake_db.query.return_value is what db.query() returns.
Then .filter() is called, so fake_db.query.return_value.filter.return_value mocks that.
Then .first() is called, so .first.return_value is either a user object or None.
When user object is returned, .name attribute is set to "Alice".
Why no patch() here?
Because:
Your function accepts db as an argument.
You can create a mock db object directly and control its methods.
No need to patch imports or globals.
When to use patch()?
If your function creates the db session inside or imports it globally, then you patch where it’s used.
Example (if db is imported in db_service module):
from unittest.mock import patch
@patch('db_service.db')
def test_get_user_name_with_patch(mock_db):
mock_db.query.return_value.filter.return_value.first.return_value.name = "Alice"
result = get_user_name(1, mock_db)
assert result == "Alice"
Summary
Situation | Use |
db passed as argument | Create MagicMock() for db, no patch needed |
db created or imported inside module | Use patch() to replace it in tests |
+++++++++++++++++++++++++++
Imagine you have this function in s3_service.py:
import boto3
def list_s3_objects(bucket_name):
s3 = boto3.client('s3') # Create S3 client
response = s3.list_objects_v2(Bucket=bucket_name) # Call AWS API
if 'Contents' in response:
return [obj['Key'] for obj in response['Contents']] # Return list of file names
return []
You want to test this function without calling real AWS, so you mock the boto3 client.
How to write a test that mocks boto3 client?
Create a test file, say test_s3_service.py:
from unittest.mock import patch, MagicMock
from s3_service import list_s3_objects
@patch('s3_service.boto3.client') # Mock boto3.client inside s3_service.py
def test_list_s3_objects(mock_boto_client):
# Step 1: Create a fake S3 client object
mock_s3 = MagicMock()
# Step 2: Tell the fake S3 client what to return when list_objects_v2 is called
mock_s3.list_objects_v2.return_value = {
'Contents': [
{'Key': 'file1.txt'},
{'Key': 'file2.txt'}
]
}
# Step 3: Make boto3.client() return your fake S3 client
mock_boto_client.return_value = mock_s3
# Step 4: Call your function (it will use the mocked client)
files = list_s3_objects('my-bucket')
# Step 5: Check if the function returned what we expect
assert files == ['file1.txt', 'file2.txt']
# Step 6: Also check if the list_objects_v2 was called correctly
mock_s3.list_objects_v2.assert_called_once_with(Bucket='my-bucket')
What happens here?
Step | Explanation |
@patch(...) | Replace boto3.client in your module with a mock |
mock_s3 = MagicMock() | Create a fake S3 client object |
mock_s3.list_objects_v2.return_value = {...} | Define fake response data from AWS |
mock_boto_client.return_value = mock_s3 | When code calls boto3.client('s3'), return this fake client |
Call function | Calls your function; it uses the fake client, not real AWS |
Assertions | Check your function returns expected data and API was called |
++++++++++++++++
Let's create a BDD (Behavior-Driven Development) testing setup for EKS (Elastic Kubernetes Service) using feature files and Python step definitions.
✅ Project Layout
CopyEdit
eks-bdd-tests/
├── features/
│ └── eks_service.feature
├── tests/
│ └── test_eks_steps.py
├── requirements.txt
📄 features/eks_service.feature
gherkin
CopyEdit
Feature: EKS Service Validation
Scenario: Verify EKS cluster is active
Given I have access to the EKS cluster "my-cluster"
Then the cluster status should be "ACTIVE"
Scenario: Verify pods are running in namespace
Given I am connected to EKS cluster "my-cluster"
When I check pods in namespace "default"
Then all pods should be in "Running" state
Scenario: Verify service is reachable
Given I have service "my-service" in namespace "default"
Then it should be accessible on port "80"
🧪 tests/test_eks_steps.py
import boto3
import pytest
from pytest_bdd import scenarios, given, when, then
from kubernetes import client, config
# Load all scenarios from the feature file
scenarios('../features/eks_service.feature')
# Globals (to share across steps)
CLUSTER_NAME = None
NAMESPACE = None
###############################
# Step Definitions
###############################
@given('I have access to the EKS cluster "<cluster_name>"')
def check_eks_cluster(cluster_name):
global CLUSTER_NAME
CLUSTER_NAME = cluster_name
eks = boto3.client('eks')
response = eks.describe_cluster(name=cluster_name)
assert 'cluster' in response
assert response['cluster']['status'] in ['ACTIVE', 'CREATING', 'UPDATING'] # validate it's a known state
@then('the cluster status should be "<expected_status>"')
def verify_cluster_status(expected_status):
eks = boto3.client('eks')
response = eks.describe_cluster(name=CLUSTER_NAME)
actual_status = response['cluster']['status']
assert actual_status == expected_status, f"Expected {expected_status}, got {actual_status}"
@given('I am connected to EKS cluster "<cluster_name>"')
def connect_k8s(cluster_name):
global CLUSTER_NAME
CLUSTER_NAME = cluster_name
config.load_kube_config() # Or load_incluster_config() if running inside the cluster
@when('I check pods in namespace "<namespace>"')
def check_pods(namespace):
config.load_kube_config()
v1 = client.CoreV1Api()
pods = v1.list_namespaced_pod(namespace=namespace)
pytest.pods = pods.items
@then('all pods should be in "Running" state')
def assert_pods_running():
for pod in pytest.pods:
assert pod.status.phase == "Running", f"Pod {pod.metadata.name} is in {pod.status.phase}"
@given('I have service "<service_name>" in namespace "<namespace>"')
def get_service(service_name, namespace):
config.load_kube_config()
v1 = client.CoreV1Api()
svc = v1.read_namespaced_service(service_name, namespace)
pytest.service = svc
@then('it should be accessible on port "<port>"')
def check_service_port(port):
ports = [p.port for p in pytest.service.spec.ports]
assert int(port) in ports, f"Port {port} not found in service ports: {ports}"
📦 requirements.txt
pytest
pytest-bdd
boto3
kubernetes
Install with:
pip install -r requirements.txt
▶️ How to Run
pytest tests/
✅ Prerequisites
You must have ~/.kube/config configured with access to your EKS cluster:
aws eks update-kubeconfig --name my-cluster
Ensure AWS credentials (IAM) allow access to EKS
🧠 What This Tests
Scenario | What it Does |
Cluster is Active | Uses boto3 to verify EKS status |
Pods Running | Uses K8s Python client to list and check pod states |
Service Exposed on Port | Verifies service has a port exposed (e.g. 80, 443) |
No comments:
Post a Comment