Module 6: Cloud Migration Patterns

5. Multi-environment Testing

Next →

Overview

In this capsule you'll build a test suite that runs against LocalStack AND against AWS with the same code. It's the most valuable pattern of cloud migration: if your tests pass in both environments, the migration is a verifiable process — not a leap of faith. You'll use pytest fixtures that detect the environment, a conftest.py with automatic configuration, markers for tests that only apply to certain environments, and feature flags in tests for AWS-only features like SageMaker.

Context: You have environment abstraction (C02), config management (C03), and dependency injection (C04). Your code works on LocalStack and AWS without changes. But how do you verify that it works the same in both? Without multi-environment tests, your migration is "I trust it works." With multi-environment tests, your migration is "the 47 tests pass on LocalStack and the 47 tests pass on AWS — migration verified." That's the difference between a developer and a production engineer.

The Strategy: Same Suite, Multiple Backends

What multi-environment testing means

Test Suite (47 tests)
├── 35 universal tests → run on LocalStack AND AWS
│   ├── test_s3_upload_prompt
│   ├── test_s3_get_document
│   ├── test_lambda_invoke
│   └── ... (35 tests)
│
├── 8 AWS-only tests → run only when ENVIRONMENT=aws
│   ├── test_sagemaker_endpoint
│   ├── test_iam_permissions
│   ├── test_cloudwatch_logs
│   └── ... (8 tests)
│
└── 4 local-only tests → run only when ENVIRONMENT=local
    ├── test_localstack_health
    ├── test_local_performance
    └── ... (4 tests)

Execution:
$ ENVIRONMENT=local pytest          → 39 tests (35 universal + 4 local)
$ ENVIRONMENT=staging pytest        → 43 tests (35 universal + 8 AWS)
$ ENVIRONMENT=production pytest     → 43 tests (35 universal + 8 AWS)

The key piece: conftest.py

conftest.py is the file where pytest looks for shared fixtures. Here lives all the environment-detection logic, client creation, and test configuration:

tests/
├── conftest.py          ← Environment detection, fixtures, markers
├── test_s3_operations.py
├── test_document_processor.py
├── test_lambda_integration.py
├── test_migration.py
└── test_aws_only.py

conftest.py: The Control Center

Complete implementation

"""tests/conftest.py — Multi-environment fixtures for pytest."""

import os
import json
import pytest
import boto3
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor


# ---------------------------------------------------------------------------
# Environment detection
# ---------------------------------------------------------------------------

def get_test_environment() -> str:
    """Determines the testing environment."""
    return os.environ.get("ENVIRONMENT", "local")


def is_aws_environment() -> bool:
    env = get_test_environment()
    return env in ("staging", "production")


def is_local_environment() -> bool:
    return get_test_environment() == "local"


# ---------------------------------------------------------------------------
# Custom markers
# ---------------------------------------------------------------------------

def pytest_configure(config):
    """Registers custom markers for per-environment tests."""
    config.addinivalue_line("markers", "aws_only: test requires AWS environment")
    config.addinivalue_line("markers", "local_only: test requires LocalStack")
    config.addinivalue_line("markers", "sagemaker: test requires SageMaker")
    config.addinivalue_line("markers", "slow: test is slow (>10s)")


def pytest_collection_modifyitems(config, items):
    """Skips tests marked according to the current environment."""
    env = get_test_environment()

    for item in items:
        if "aws_only" in item.keywords and not is_aws_environment():
            item.add_marker(pytest.mark.skip(
                reason=f"Requires AWS (current environment: {env})"
            ))

        if "local_only" in item.keywords and not is_local_environment():
            item.add_marker(pytest.mark.skip(
                reason=f"Requires LocalStack (current environment: {env})"
            ))

        if "sagemaker" in item.keywords:
            settings = Settings()
            if not settings.feature_sagemaker_enabled:
                item.add_marker(pytest.mark.skip(
                    reason="SageMaker not enabled in this environment"
                ))


# ---------------------------------------------------------------------------
# Configuration fixtures
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session")
def test_settings() -> Settings:
    """Settings for the current testing environment."""
    env = get_test_environment()
    env_file = f".env.{env}"
    if os.path.exists(env_file):
        return Settings(_env_file=env_file)
    return Settings()


@pytest.fixture(scope="session")
def client_factory(test_settings) -> ClientFactory:
    """Client factory for the testing environment."""
    return ClientFactory(test_settings)


# ---------------------------------------------------------------------------
# S3 fixtures
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session")
def s3_client(client_factory):
    """S3 client for tests."""
    return client_factory.s3


@pytest.fixture(scope="session")
def test_bucket(test_settings, s3_client) -> str:
    """Bucket for tests — created if it doesn't exist."""
    bucket_name = f"{test_settings.s3_bucket}-test"
    try:
        s3_client.head_bucket(Bucket=bucket_name)
    except Exception:
        try:
            s3_client.create_bucket(Bucket=bucket_name)
        except s3_client.exceptions.BucketAlreadyOwnedByYou:
            pass
    return bucket_name


@pytest.fixture
def clean_bucket(s3_client, test_bucket):
    """Cleans the test bucket before each test."""
    yield test_bucket

    paginator = s3_client.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=test_bucket):
        objects = page.get("Contents", [])
        if objects:
            s3_client.delete_objects(
                Bucket=test_bucket,
                Delete={"Objects": [{"Key": o["Key"]} for o in objects]},
            )


# ---------------------------------------------------------------------------
# Service fixtures
# ---------------------------------------------------------------------------

@pytest.fixture
def processor(s3_client, test_bucket) -> DocumentProcessor:
    """DocumentProcessor configured for tests."""
    return DocumentProcessor(s3_client=s3_client, bucket=test_bucket)


# ---------------------------------------------------------------------------
# Environment info fixture (for debugging)
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session", autouse=True)
def print_environment(test_settings):
    """Prints environment info at the start of the test suite."""
    print(f"\n{'='*60}")
    print(f"TEST ENVIRONMENT: {test_settings.environment}")
    print(f"Endpoint: {test_settings.aws_endpoint_url or 'AWS default'}")
    print(f"Bucket: {test_settings.s3_bucket}")
    print(f"SageMaker: {'enabled' if test_settings.feature_sagemaker_enabled else 'disabled'}")
    print(f"{'='*60}\n")

Universal Tests: They Run in All Environments

test_s3_operations.py

"""tests/test_s3_operations.py — S3 tests that run in any environment."""

import json
import pytest


class TestS3PutAndGet:
    """Tests of basic S3 operations."""

    def test_put_and_get_text(self, s3_client, clean_bucket):
        key = "test/hello.txt"
        s3_client.put_object(
            Bucket=clean_bucket, Key=key, Body=b"hello world"
        )
        response = s3_client.get_object(Bucket=clean_bucket, Key=key)
        content = response["Body"].read().decode("utf-8")
        assert content == "hello world"

    def test_put_and_get_json(self, s3_client, clean_bucket):
        key = "test/data.json"
        data = {"name": "test", "value": 42, "tags": ["a", "b"]}
        s3_client.put_object(
            Bucket=clean_bucket,
            Key=key,
            Body=json.dumps(data).encode("utf-8"),
            ContentType="application/json",
        )
        response = s3_client.get_object(Bucket=clean_bucket, Key=key)
        result = json.loads(response["Body"].read().decode("utf-8"))
        assert result == data

    def test_object_metadata(self, s3_client, clean_bucket):
        key = "test/with-metadata.txt"
        s3_client.put_object(
            Bucket=clean_bucket,
            Key=key,
            Body=b"content",
            Metadata={"version": "v1", "author": "test-suite"},
        )
        response = s3_client.head_object(Bucket=clean_bucket, Key=key)
        assert response["Metadata"]["version"] == "v1"
        assert response["Metadata"]["author"] == "test-suite"

    def test_list_objects_by_prefix(self, s3_client, clean_bucket):
        for i in range(5):
            s3_client.put_object(
                Bucket=clean_bucket,
                Key=f"prompts/test/v{i}/system.txt",
                Body=f"prompt v{i}".encode(),
            )

        response = s3_client.list_objects_v2(
            Bucket=clean_bucket, Prefix="prompts/test/"
        )
        keys = [o["Key"] for o in response.get("Contents", [])]
        assert len(keys) == 5
        assert all(k.startswith("prompts/test/") for k in keys)

    def test_delete_object(self, s3_client, clean_bucket):
        key = "test/to-delete.txt"
        s3_client.put_object(Bucket=clean_bucket, Key=key, Body=b"delete me")
        s3_client.delete_object(Bucket=clean_bucket, Key=key)

        with pytest.raises(Exception):
            s3_client.get_object(Bucket=clean_bucket, Key=key)

    def test_nonexistent_key_raises(self, s3_client, clean_bucket):
        with pytest.raises(Exception):
            s3_client.get_object(
                Bucket=clean_bucket, Key="does/not/exist.txt"
            )

test_document_processor.py

"""tests/test_document_processor.py — Processor tests in any environment."""

import json
import pytest


class TestDocumentProcessor:
    """Tests of DocumentProcessor against the current environment."""

    @pytest.fixture(autouse=True)
    def setup_prompts(self, s3_client, test_bucket):
        """Creates the prompt templates needed for the tests."""
        s3_client.put_object(
            Bucket=test_bucket,
            Key="prompts/summarizer/v1/system.txt",
            Body="Summarize the document in 3 key points.".encode("utf-8"),
        )
        s3_client.put_object(
            Bucket=test_bucket,
            Key="prompts/classifier/v1/system.txt",
            Body="Classify the document into a category.".encode("utf-8"),
        )

    def test_get_prompt_summarizer(self, processor):
        prompt = processor.get_prompt("summarizer", "v1")
        assert "Summarize" in prompt
        assert "3 key points" in prompt

    def test_get_prompt_classifier(self, processor):
        prompt = processor.get_prompt("classifier", "v1")
        assert "Classify" in prompt

    def test_get_prompt_nonexistent_raises(self, processor):
        with pytest.raises(Exception):
            processor.get_prompt("nonexistent", "v99")

    def test_store_and_retrieve_document(self, processor, s3_client, test_bucket):
        doc_data = {
            "id": "test-doc-001",
            "title": "Test Document",
            "content": "Test content for multi-environment testing.",
        }
        key = processor.store_document("test-collection", "test-doc-001", doc_data)

        response = s3_client.get_object(Bucket=test_bucket, Key=key)
        stored = json.loads(response["Body"].read().decode("utf-8"))

        assert stored["id"] == "test-doc-001"
        assert stored["title"] == "Test Document"

    def test_save_response(self, processor, s3_client, test_bucket):
        result_data = {"status": "processed", "tokens_used": 150}
        key = processor.save_response("req-test-001", result_data)

        assert key.startswith("responses/")
        assert "req-test-001" in key

        response = s3_client.get_object(Bucket=test_bucket, Key=key)
        saved = json.loads(response["Body"].read().decode("utf-8"))

        assert saved["request_id"] == "req-test-001"
        assert saved["status"] == "processed"
        assert "timestamp" in saved

Tests with Environment Markers

AWS-only tests

"""tests/test_aws_only.py — Tests that only run on AWS."""

import pytest


@pytest.mark.aws_only
class TestAWSSpecific:
    """Tests that require real AWS (not LocalStack)."""

    def test_iam_role_exists(self, client_factory, test_settings):
        """Verifies that the Lambda IAM role exists in AWS."""
        iam = client_factory.get_client("iam")
        try:
            response = iam.get_role(RoleName="ai-processor-lambda-role")
            assert response["Role"]["RoleName"] == "ai-processor-lambda-role"
        except Exception:
            pytest.skip("IAM role not configured in this environment")

    def test_s3_bucket_encryption(self, s3_client, test_settings):
        """Verifies that the bucket has encryption enabled."""
        try:
            response = s3_client.get_bucket_encryption(
                Bucket=test_settings.s3_bucket
            )
            rules = response["ServerSideEncryptionConfiguration"]["Rules"]
            assert len(rules) > 0
        except Exception:
            pytest.skip("Encryption not configured")

    def test_cloudwatch_log_group_exists(self, client_factory, test_settings):
        """Verifies that the Lambda log group exists."""
        logs = client_factory.get_client("logs")
        log_group = f"/aws/lambda/{test_settings.lambda_function_name}"
        response = logs.describe_log_groups(logGroupNamePrefix=log_group)
        groups = response.get("logGroups", [])
        matching = [g for g in groups if g["logGroupName"] == log_group]
        assert len(matching) > 0, f"Log group {log_group} not found"


@pytest.mark.sagemaker
class TestSageMaker:
    """Tests that require SageMaker (AWS only, with the feature flag)."""

    def test_sagemaker_endpoint_exists(self, client_factory):
        """Verifies that the SageMaker endpoint exists."""
        sm = client_factory.get_client("sagemaker")
        response = sm.list_endpoints(MaxResults=10)
        endpoints = response.get("Endpoints", [])
        assert len(endpoints) >= 0  # Verify that the API responds

LocalStack-only tests

"""tests/test_local_only.py — LocalStack-specific tests."""

import pytest
import requests


@pytest.mark.local_only
class TestLocalStackSpecific:
    """Tests that only make sense on LocalStack."""

    def test_localstack_health(self):
        """Verifies that LocalStack is healthy."""
        response = requests.get("http://localhost:4566/_localstack/health")
        assert response.status_code == 200
        health = response.json()
        assert "services" in health

    def test_localstack_s3_service_running(self):
        """Verifies that the S3 service is active."""
        response = requests.get("http://localhost:4566/_localstack/health")
        services = response.json().get("services", {})
        assert services.get("s3") in ("running", "available")

    def test_local_latency_acceptable(self, s3_client, clean_bucket):
        """Verifies that local operations are fast."""
        import time

        start = time.time()
        for i in range(10):
            s3_client.put_object(
                Bucket=clean_bucket,
                Key=f"perf/test-{i}.txt",
                Body=b"performance test",
            )
        elapsed = time.time() - start

        assert elapsed < 5.0, (
            f"10 put_object took {elapsed:.2f}s — "
            f"LocalStack should be faster"
        )

pytest.ini: Test Runner Configuration

# pytest.ini
[pytest]
markers =
    aws_only: test requires AWS environment
    local_only: test requires LocalStack environment
    sagemaker: test requires SageMaker (AWS with feature flag)
    slow: test takes more than 10 seconds

testpaths = tests
python_files = test_*.py
python_functions = test_*
python_classes = Test*

addopts =
    -v
    --tb=short
    -x

log_cli = true
log_cli_level = INFO

Execution: Same Suite, Different Environments

Against LocalStack

# Start LocalStack
docker run -d --name localstack -p 4566:4566 localstack/localstack

# Run the tests
ENVIRONMENT=local pytest tests/ -v

# Expected output:
# ============================================
# TEST ENVIRONMENT: local
# Endpoint: http://localhost:4566
# SageMaker: disabled
# ============================================
#
# tests/test_s3_operations.py::TestS3PutAndGet::test_put_and_get_text PASSED
# tests/test_s3_operations.py::TestS3PutAndGet::test_put_and_get_json PASSED
# tests/test_document_processor.py::TestDocumentProcessor::test_get_prompt PASSED
# tests/test_aws_only.py::TestAWSSpecific::test_iam_role_exists SKIPPED (Requires AWS)
# tests/test_aws_only.py::TestSageMaker::test_sagemaker_endpoint SKIPPED (SageMaker not enabled)
# tests/test_local_only.py::TestLocalStackSpecific::test_localstack_health PASSED
#
# 39 passed, 8 skipped

Against AWS Staging

# Ensure AWS credentials are configured
aws sts get-caller-identity

# Run the tests
ENVIRONMENT=staging pytest tests/ -v

# Expected output:
# ============================================
# TEST ENVIRONMENT: staging
# Endpoint: AWS default
# SageMaker: enabled
# ============================================
#
# tests/test_s3_operations.py::TestS3PutAndGet::test_put_and_get_text PASSED
# tests/test_s3_operations.py::TestS3PutAndGet::test_put_and_get_json PASSED
# tests/test_document_processor.py::TestDocumentProcessor::test_get_prompt PASSED
# tests/test_aws_only.py::TestAWSSpecific::test_iam_role_exists PASSED
# tests/test_aws_only.py::TestSageMaker::test_sagemaker_endpoint PASSED
# tests/test_local_only.py::TestLocalStackSpecific::test_localstack_health SKIPPED (Requires LocalStack)
#
# 43 passed, 4 skipped

Automated comparison

"""scripts/run_migration_tests.py — Runs tests in both environments and compares."""

import subprocess
import json
import sys


def run_tests(environment: str) -> dict:
    """Runs pytest against an environment and captures the results."""
    result = subprocess.run(
        ["pytest", "tests/", "-v", "--tb=short", "-q", "--no-header"],
        capture_output=True,
        text=True,
        env={**dict(__import__("os").environ), "ENVIRONMENT": environment},
    )

    lines = result.stdout.strip().split("\n")
    summary_line = lines[-1] if lines else ""

    return {
        "environment": environment,
        "exit_code": result.returncode,
        "output": result.stdout,
        "summary": summary_line,
        "success": result.returncode == 0,
    }


def compare_results():
    """Runs tests on local and staging, compares the results."""
    print("Running tests against LocalStack...")
    local = run_tests("local")

    print("\nRunning tests against AWS staging...")
    staging = run_tests("staging")

    print("\n" + "=" * 60)
    print("MIGRATION TEST COMPARISON")
    print("=" * 60)
    print(f"\nLocalStack: {local['summary']}")
    print(f"AWS Staging: {staging['summary']}")

    if local["success"] and staging["success"]:
        print("\n✅ MIGRATION VERIFIED: Tests pass in both environments")
    else:
        print("\n❌ MIGRATION ISSUE: Tests differ between environments")
        if not local["success"]:
            print(f"  LocalStack failures:\n{local['output']}")
        if not staging["success"]:
            print(f"  AWS Staging failures:\n{staging['output']}")


if __name__ == "__main__":
    compare_results()

Troubleshooting

Problem 1: Tests pass locally but fail on AWS

S3 behavior differs slightly between LocalStack and AWS.

# Common difference: LocalStack is more permissive with bucket names
# AWS rejects names with uppercase or underscores

# ❌ Works on LocalStack, fails on AWS
bucket = "My_Test_Bucket"

# ✅ Works on both
bucket = "my-test-bucket"

# Another difference: consistency model
# LocalStack is always consistent, AWS S3 is strongly consistent
# (since 2020), but list_objects can have minimal lag

Problem 2: "Bucket already exists" when running tests

The test_bucket fixture uses scope="session" but if another developer has the bucket:

# Solution: use bucket names with a unique suffix
import hashlib
import os

username = os.environ.get("USER", "unknown")
suffix = hashlib.md5(username.encode()).hexdigest()[:8]
bucket_name = f"test-{suffix}"

Problem 3: AWS-only tests run locally

The marker isn't being applied correctly. Check conftest.py:

# Make sure pytest_collection_modifyitems is in conftest.py
# (not in another file) and that it uses the correct marker:

if "aws_only" in item.keywords and not is_aws_environment():
    item.add_marker(pytest.mark.skip(...))

# Debug: check the environment
print(f"ENVIRONMENT = {os.environ.get('ENVIRONMENT')}")

Problem 4: Session fixtures aren't cleaned up between runs

The test bucket accumulates objects between runs.

# Add a cleanup fixture at the end of the session
@pytest.fixture(scope="session", autouse=True)
def cleanup_test_bucket(s3_client, test_bucket):
    yield
    # Cleanup after all the tests
    paginator = s3_client.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=test_bucket):
        objects = page.get("Contents", [])
        if objects:
            s3_client.delete_objects(
                Bucket=test_bucket,
                Delete={"Objects": [{"Key": o["Key"]} for o in objects]},
            )

Practical Exercises

Exercise 1: Fixture parametrized by environment

Create a fixture that parametrizes the tests so they run with two bucket configurations: one with the v1/ prefix and another with the v2/ prefix, verifying that the DocumentProcessor works with both structures.

See solution 
import pytest
import json
from services.document_processor import DocumentProcessor


@pytest.fixture(params=["v1", "v2"])
def versioned_processor(request, s3_client, test_bucket):
    """Parametrized fixture: creates a processor with different prefix versions."""
    version = request.param

    s3_client.put_object(
        Bucket=test_bucket,
        Key=f"prompts/summarizer/{version}/system.txt",
        Body=f"Prompt template version {version}".encode("utf-8"),
    )

    processor = DocumentProcessor(s3_client=s3_client, bucket=test_bucket)
    return processor, version


class TestVersionedProcessor:
    def test_get_versioned_prompt(self, versioned_processor):
        processor, version = versioned_processor
        prompt = processor.get_prompt("summarizer", version)
        assert f"version {version}" in prompt

    def test_store_document_with_version(self, versioned_processor, s3_client, test_bucket):
        processor, version = versioned_processor
        doc = {"id": f"doc-{version}", "version": version, "data": "test"}
        key = processor.store_document(f"collection-{version}", f"doc-{version}", doc)

        response = s3_client.get_object(Bucket=test_bucket, Key=key)
        stored = json.loads(response["Body"].read().decode("utf-8"))
        assert stored["version"] == version

Exercise 2: End-to-end migration test

Write a test that simulates a migration: uploads data to "local" (LocalStack), verifies it exists, then verifies that the same data is accessible from the "staging" configuration (if available).

See solution 
import json
import pytest
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory


def test_migration_data_integrity(s3_client, test_bucket):
    """Verifies data integrity: upload, list, and verify content."""
    test_data = [
        {"id": f"migration-doc-{i}", "content": f"Document {i}", "index": i}
        for i in range(5)
    ]

    # 1. Upload the data
    for doc in test_data:
        s3_client.put_object(
            Bucket=test_bucket,
            Key=f"migration-test/{doc['id']}.json",
            Body=json.dumps(doc).encode("utf-8"),
        )

    # 2. List and verify the count
    response = s3_client.list_objects_v2(
        Bucket=test_bucket, Prefix="migration-test/"
    )
    objects = response.get("Contents", [])
    assert len(objects) == 5, f"Expected 5 objects, got {len(objects)}"

    # 3. Verify the content of each document
    for doc in test_data:
        response = s3_client.get_object(
            Bucket=test_bucket,
            Key=f"migration-test/{doc['id']}.json",
        )
        stored = json.loads(response["Body"].read().decode("utf-8"))
        assert stored["id"] == doc["id"]
        assert stored["content"] == doc["content"]
        assert stored["index"] == doc["index"]

    # 4. Clean up
    s3_client.delete_objects(
        Bucket=test_bucket,
        Delete={
            "Objects": [
                {"Key": f"migration-test/{doc['id']}.json"}
                for doc in test_data
            ]
        },
    )

    # 5. Verify cleanup
    response = s3_client.list_objects_v2(
        Bucket=test_bucket, Prefix="migration-test/"
    )
    assert response.get("KeyCount", 0) == 0

Exercise 3: Comparative performance test

Create a test that measures the time of 100 put_object and get_object operations, and records it as the test result to compare between environments.

See solution 
import time
import json
import pytest


class TestPerformance:
    """Performance tests that report per-environment times."""

    def test_put_object_throughput(self, s3_client, clean_bucket, test_settings):
        """Measures put_object throughput."""
        num_operations = 50
        payload = json.dumps({"data": "x" * 1000}).encode("utf-8")

        start = time.time()
        for i in range(num_operations):
            s3_client.put_object(
                Bucket=clean_bucket,
                Key=f"perf/put-{i:04d}.json",
                Body=payload,
            )
        elapsed = time.time() - start

        ops_per_sec = num_operations / elapsed
        print(
            f"\n[{test_settings.environment}] "
            f"put_object: {ops_per_sec:.1f} ops/s "
            f"({elapsed:.2f}s for {num_operations} ops)"
        )

        assert elapsed < 60, f"put_object too slow: {elapsed:.2f}s"

    def test_get_object_throughput(self, s3_client, clean_bucket, test_settings):
        """Measures get_object throughput."""
        num_operations = 50
        payload = json.dumps({"data": "x" * 1000}).encode("utf-8")

        for i in range(num_operations):
            s3_client.put_object(
                Bucket=clean_bucket,
                Key=f"perf/get-{i:04d}.json",
                Body=payload,
            )

        start = time.time()
        for i in range(num_operations):
            response = s3_client.get_object(
                Bucket=clean_bucket,
                Key=f"perf/get-{i:04d}.json",
            )
            response["Body"].read()
        elapsed = time.time() - start

        ops_per_sec = num_operations / elapsed
        print(
            f"\n[{test_settings.environment}] "
            f"get_object: {ops_per_sec:.1f} ops/s "
            f"({elapsed:.2f}s for {num_operations} ops)"
        )

        assert elapsed < 60, f"get_object too slow: {elapsed:.2f}s"

Exercise 4: Feature flags fixture for tests

Create a feature_flags fixture that exposes the current environment's feature flags and a skip_if_feature_disabled helper that tests use to skip if a feature isn't available.

See solution 
import pytest
from config.settings import Settings


@pytest.fixture(scope="session")
def feature_flags(test_settings: Settings) -> dict:
    """Feature flags of the current environment."""
    return {
        "sagemaker": test_settings.feature_sagemaker_enabled,
        "advanced_logging": test_settings.feature_advanced_logging,
        "cost_tracking": test_settings.feature_cost_tracking,
    }


def skip_if_disabled(feature_flags: dict, feature: str):
    """Helper: skips the test if a feature isn't enabled."""
    if not feature_flags.get(feature, False):
        pytest.skip(f"Feature '{feature}' not enabled in this environment")


class TestWithFeatureFlags:
    def test_basic_always_runs(self, feature_flags):
        """This test always runs."""
        assert isinstance(feature_flags, dict)

    def test_sagemaker_integration(self, feature_flags, client_factory):
        """Only runs if SageMaker is enabled."""
        skip_if_disabled(feature_flags, "sagemaker")
        sm = client_factory.get_client("sagemaker")
        response = sm.list_endpoints(MaxResults=1)
        assert "Endpoints" in response

    def test_advanced_logging(self, feature_flags, test_settings):
        """Only runs if advanced logging is enabled."""
        skip_if_disabled(feature_flags, "advanced_logging")
        assert test_settings.feature_advanced_logging is True

    def test_cost_tracking(self, feature_flags, test_settings):
        """Only runs if cost tracking is enabled."""
        skip_if_disabled(feature_flags, "cost_tracking")
        assert test_settings.feature_cost_tracking is True

    def test_feature_flags_report(self, feature_flags, test_settings):
        """Prints a feature flags report."""
        print(f"\nFeature Flags ({test_settings.environment}):")
        for flag, enabled in feature_flags.items():
            status = "✅ enabled" if enabled else "❌ disabled"
            print(f"  {flag}: {status}")

Summary

  • Multi-environment testing is the safety net of the migration. If your tests pass on LocalStack and on AWS, the migration is verified.
  • conftest.py is the control center. It detects the environment, creates fixtures, applies markers, configures cleanup. All in one file.
  • Markers (@pytest.mark.aws_only, @pytest.mark.local_only) enable conditional tests that skip automatically according to the environment.
  • The same test suite runs with ENVIRONMENT=local or ENVIRONMENT=staging. The universal tests always run. The specific tests skip automatically.
  • Fixtures with DI: the processor fixture receives its S3 client from the factory, configured for the current environment. The test doesn't know whether it talks to LocalStack or AWS.
  • In the next capsule, you'll implement feature flags to handle services that only exist in certain environments.

Additional Resources

  1. pytest Documentation — Official pytest documentation
  2. pytest Fixtures — Complete fixtures guide
  3. pytest Markers — Markers to categorize tests
  4. conftest.py — pytest — Sharing fixtures
  5. Testing boto3 Applications — Testing with boto3
  6. moto — Mock AWS Services — Mocking framework for AWS (an alternative to LocalStack in tests)
← Previous4. Dependency Injection for boto3 ClientsNext →6. Cloud Feature Flags