13 KiB
Testing Guide
This document covers testing strategies and best practices for Fission Python functions.
Table of Contents
- Test Types
- Dependencies
- Unit Testing
- Integration Testing
- Test Database
- Mocking
- Fixtures
- Coverage
- Running Tests
- CI/CD Integration
Test Types
Unit Tests
Test individual functions in isolation, mocking external dependencies:
- Database calls
- HTTP requests
- File I/O
- External services
Goal: Verify business logic correctness without infrastructure.
Integration Tests
Test the function with real (or test) dependencies:
- Actual database queries
- End-to-end request/response flow
- Real configuration loading
Goal: Verify integration points work correctly.
Dependencies
Install test dependencies:
pip install -r test/requirements.txt
# Or for dev (includes both runtime and test deps):
pip install -r dev-requirements.txt
Required packages:
pytest- Test frameworkpytest-mock- Mocking utilities (providesmockerfixture)requests- For integration tests making HTTP calls
Unit Testing
Example Test Structure
# test/test_my_function.py
import pytest
from unittest.mock import patch, MagicMock
from src.my_function import create_item
from exceptions import ValidationError
def test_create_item_success():
"""Test successful item creation."""
# Arrange
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchone.return_value = ("item-id", "Item Name", "active")
# Mock init_db_connection to return our mock
with patch("src.my_function.init_db_connection", return_value=mock_conn):
# Create a mock Flask request
with patch("src.my_function.request") as mock_request:
mock_request.get_json.return_value = {
"name": "Test Item",
"status": "active"
}
mock_request.view_args = {}
# Act
result = create_item({}, {})
# Assert
assert result["id"] == "item-id"
assert result["name"] == "Test Item"
mock_cursor.execute.assert_called_once()
mock_conn.commit.assert_called_once()
def test_create_item_validation_error():
"""Test validation of missing required fields."""
with patch("src.my_function.request") as mock_request:
mock_request.get_json.return_value = {"name": ""} # Empty name
with pytest.raises(ValidationError) as exc_info:
create_item({}, {})
assert "validation" in str(exc_info.value.error_msg).lower()
Mocking Helpers
Use patch to replace dependencies:
# Mock helpers.get_secret
@patch("src.my_function.helpers.get_secret")
def test_with_mocked_secret(mock_get_secret):
mock_get_secret.return_value = "localhost"
# Test code...
# Mock entire module
@patch("src.my_function.helpers.init_db_connection")
def test_with_mocked_db(mock_init_db):
mock_conn = MagicMock()
mock_init_db.return_value = mock_conn
# Test code...
Mocking Flask Request
from flask import Request
def test_with_flask_request():
with patch("src.my_function.request") as mock_request:
mock_request.get_json.return_value = {"key": "value"}
mock_request.args.getlist.return_value = []
mock_request.headers.get.return_value = "user-123"
# Test code...
Integration Testing
Test Database Setup
Use a separate test database:
# Create test database
createdb fission_test
# Or with Docker:
docker run -d -p 5433:5432 -e POSTGRES_PASSWORD=test postgres:15
Set environment variables for test database:
export PG_HOST=localhost
export PG_PORT=5433
export PG_DB=fission_test
export PG_USER=postgres
export PG_PASS=test
pytest Fixtures for Database
# conftest.py (placed in test/ directory)
import pytest
import psycopg2
from helpers import init_db_connection
@pytest.fixture(scope="session")
def db_connection():
"""Create a database connection for the entire test session."""
conn = init_db_connection()
yield conn
conn.close()
@pytest.fixture(scope="function")
def db_cursor(db_connection):
"""Create a cursor for each test, with transaction rollback."""
conn = db_connection
cursor = conn.cursor()
# Start a transaction that will be rolled back
conn.rollback()
yield cursor
# Rollback after each test to keep DB clean
conn.rollback()
Example Integration Test
# test/test_integration.py
def test_create_and_retrieve_item_integration(db_connection):
"""Test full CRUD cycle with real database."""
from src.models import ItemCreateRequest
from src.functions import create_item, get_item
# Insert test data
cursor = db_connection.cursor()
cursor.execute("DELETE FROM items WHERE name = 'Integration Test'")
db_connection.commit()
# Create item via function
with patch("src.functions.request") as mock_request:
mock_request.get_json.return_value = {
"name": "Integration Test",
"description": "Test item"
}
mock_request.view_args = {}
result = create_item({}, {})
item_id = result["id"]
assert result["name"] == "Integration Test"
# Retrieve same item
with patch("src.functions.request") as mock_request:
with patch("src.functions.request.view_args", {"id": item_id}):
result = get_item({"path": f"/items/{item_id}"}, {})
assert result["id"] == item_id
# Cleanup
cursor.execute("DELETE FROM items WHERE id = %s", (item_id,))
db_connection.commit()
Test Database Migrations
Apply migrations before integration tests:
# conftest.py
import subprocess
def apply_migrations():
"""Apply all SQL migrations to test database."""
import os
migrates_dir = os.path.join(os.path.dirname(__file__), "..", "migrates")
for file in sorted(os.listdir(migrates_dir)):
if file.endswith(".sql"):
path = os.path.join(migrates_dir, file)
subprocess.run(
["psql", "-d", "fission_test", "-f", path],
check=True
)
@pytest.fixture(scope="session", autouse=True)
def setup_database():
"""Run migrations before any tests."""
apply_migrations()
yield
# Optionally drop and recreate after tests
Mocking
Built-in unittest.mock
from unittest.mock import patch, MagicMock, mock_open
# Simple patch
with patch("module.function") as mock_func:
mock_func.return_value = "mocked"
# call code that uses module.function
# Assert called with specific args
mock_func.assert_called_once_with("arg1", "arg2")
# Mock context manager
with patch("builtins.open", mock_open(read_data="file content")) as mock_file:
# code that opens file
mock_file.assert_called_with("path/to/file", "r")
pytest-mock Fixture
Simpler syntax using mocker fixture:
def test_with_mocker(mocker):
mock_func = mocker.patch("src.function.helper")
mock_func.return_value = {"key": "value"}
# test code...
Fixtures
Create reusable fixtures in conftest.py:
# test/conftest.py
import pytest
@pytest.fixture
def sample_item_data():
"""Provide sample item data for tests."""
return {
"name": "Test Item",
"description": "A test item",
"status": "active"
}
@pytest.fixture
def mock_db_connection(mocker):
"""Provide a mocked database connection."""
mock_conn = mocker.MagicMock()
mock_cursor = mocker.MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchone.return_value = None
return mock_conn
Fixtures are automatically available to all tests in the directory.
Coverage
Measure test coverage with pytest-cov:
# Install
pip install pytest-cov
# Run with coverage
pytest --cov=src
# HTML report
pytest --cov=src --cov-report=html
open htmlcov/index.html
# Show missing lines
pytest --cov=src --cov-report=term-missing
Aim for high coverage of business logic (80%+). Don't worry about 100% coverage of trivial getters/setters.
Excluding Files
Add to pytest.ini:
[pytest]
addopts = --cov=src --cov-exclude=src/vault.py
Or use .coveragerc:
[run]
omit = src/vault.py
Running Tests
Basic Commands
# Run all tests
pytest
# Verbose
pytest -v
# Run specific test file
pytest test/test_my_function.py
# Run specific test function
pytest test/test_my_function.py::test_create_item_success
# Run with markers
pytest -m "integration" # if using @pytest.mark.integration
# Stop on first failure
pytest -x
# Show print statements
pytest -s
Environment Setup
Create test/.env or set environment variables before tests:
# For integration tests
export PG_HOST=localhost
export PG_PORT=5432
export PG_DB=fission_test
Or use a pytest fixture to load from .env:
# conftest.py
import os
from dotenv import load_dotenv
@pytest.fixture(scope="session", autouse=True)
def load_env():
env_path = os.path.join(os.path.dirname(__file__), ".env")
load_dotenv(env_path)
Markers
Mark tests as unit/integration/slow:
import pytest
@pytest.mark.unit
def test_quick_unit():
pass
@pytest.mark.integration
def test_full_workflow():
pass
@pytest.mark.slow
def test_long_running():
pass
Run only unit tests:
pytest -m "unit"
Skip tests:
pytest -m "not slow"
CI/CD Integration
GitHub Actions Example
# .github/workflows/test.yaml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRES_PASSWORD: test
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 dev-requirements.txt
- name: Setup database
run: |
createdb -h localhost -U postgres fission_test
psql -h localhost -U postgres fission_test -f migrates/001_schema.sql
env:
PGPASSWORD: test
- name: Run tests
run: |
pytest --cov=src --cov-report=xml
env:
PG_HOST: localhost
PG_PORT: 5432
PG_DB: fission_test
PG_USER: postgres
PG_PASS: test
- name: Upload coverage
uses: codecov/codecov-action@v3
Best Practices
- One assertion per test - Keep tests focused
- Use descriptive names -
test_create_item_validation_error_for_missing_name - Arrange-Act-Assert - Structure tests clearly
- Mock external dependencies - Don't rely on network or external services
- Test error cases - Don't just test happy paths
- Use fixtures - Reuse setup/teardown code
- Keep tests independent - No shared state between tests
- Test edge cases - Empty inputs, null values, boundary conditions
- Don't test libraries - Don't write tests for Flask/Pydantic themselves
- Clean up resources - Use fixtures to ensure cleanup
Common Patterns
Testing Exceptions
def test_raises_not_found():
with pytest.raises(NotFoundError) as exc:
get_item("nonexistent-id")
assert exc.value.http_status == 404
Parametrized Tests
import pytest
@pytest.mark.parametrize("input,expected", [
("true", True),
("false", False),
("", None),
(None, None),
])
def test_str_to_bool(input, expected):
from helpers import str_to_bool
assert str_to_bool(input) == expected
Temporary Files/Directories
def test_with_temp_file(tmp_path):
# tmp_path is a pathlib.Path to a temporary directory
file = tmp_path / "test.txt"
file.write_text("content")
assert file.read_text() == "content"
Troubleshooting
Tests Fail with Database Errors
- Check test database is running:
pg_isready -h localhost -p 5432 - Verify migrations applied:
psql -l | grep fission_test - Check environment variables:
echo $PG_HOST
Mock Not Working
- Ensure you're patching the correct import location (where it's used, not where it's defined)
# Wrong: patching where it's defined @patch("helpers.get_secret") # Right: patching where it's used in your function module @patch("src.my_function.helpers.get_secret")
Import Errors
Ensure PYTHONPATH includes project root:
export PYTHONPATH=/path/to/project:$PYTHONPATH
Or use pytest's pythonpath option in pytest.ini:
[pytest]
pythonpath = .