Módulo 6: Cloud Migration Patterns

5. Testing Multi-Entorno

Siguiente →

Descripción

En esta cápsula vas a construir un test suite que corre contra LocalStack Y contra AWS con el mismo código. Es el patrón más valioso de la migración cloud: si tus tests pasan en ambos entornos, la migración es un proceso verificable — no un salto de fe. Vas a usar pytest fixtures que detectan el entorno, conftest.py con configuración automática, markers para tests que solo aplican a ciertos entornos, y feature flags en tests para AWS-only features como SageMaker.

Contexto: Tienes environment abstraction (C02), config management (C03), y dependency injection (C04). Tu código funciona en LocalStack y AWS sin cambios. Pero ¿cómo verificas que funciona igual en ambos? Sin tests multi-entorno, tu migración es "confío en que funciona." Con tests multi-entorno, tu migración es "los 47 tests pasan en LocalStack y los 47 tests pasan en AWS — migración verificada." Esa es la diferencia entre un developer y un ingeniero de producción.

La Estrategia: Mismo Suite, Múltiples Backends

Qué significa testing multi-entorno

Test Suite (47 tests)
├── 35 tests universales → corren en LocalStack Y AWS
│   ├── test_s3_upload_prompt
│   ├── test_s3_get_document
│   ├── test_lambda_invoke
│   └── ... (35 tests)
│
├── 8 tests AWS-only → corren solo cuando ENVIRONMENT=aws
│   ├── test_sagemaker_endpoint
│   ├── test_iam_permissions
│   ├── test_cloudwatch_logs
│   └── ... (8 tests)
│
└── 4 tests local-only → corren solo cuando ENVIRONMENT=local
    ├── test_localstack_health
    ├── test_local_performance
    └── ... (4 tests)

Ejecución:
$ 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)

La pieza clave: conftest.py

conftest.py es el archivo donde pytest busca fixtures compartidas. Aquí vive toda la lógica de detección de entorno, creación de clients, y configuración de tests:

tests/
├── conftest.py          ← Detección de entorno, fixtures, markers
├── test_s3_operations.py
├── test_document_processor.py
├── test_lambda_integration.py
├── test_migration.py
└── test_aws_only.py

conftest.py: El Centro de Control

Implementación completa

"""tests/conftest.py — Fixtures multi-entorno para 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


# ---------------------------------------------------------------------------
# Detección de entorno
# ---------------------------------------------------------------------------

def get_test_environment() -> str:
    """Determina el entorno de testing."""
    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"


# ---------------------------------------------------------------------------
# Markers custom
# ---------------------------------------------------------------------------

def pytest_configure(config):
    """Registra markers custom para tests por entorno."""
    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):
    """Salta tests marcados según el entorno actual."""
    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"Requiere AWS (entorno actual: {env})"
            ))

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

        if "sagemaker" in item.keywords:
            settings = Settings()
            if not settings.feature_sagemaker_enabled:
                item.add_marker(pytest.mark.skip(
                    reason="SageMaker no habilitado en este entorno"
                ))


# ---------------------------------------------------------------------------
# Fixtures de configuración
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session")
def test_settings() -> Settings:
    """Settings para el entorno de testing actual."""
    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:
    """Factory de clients para el entorno de testing."""
    return ClientFactory(test_settings)


# ---------------------------------------------------------------------------
# Fixtures de S3
# ---------------------------------------------------------------------------

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


@pytest.fixture(scope="session")
def test_bucket(test_settings, s3_client) -> str:
    """Bucket para tests — se crea si no existe."""
    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):
    """Limpia el bucket de test antes de cada 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]},
            )


# ---------------------------------------------------------------------------
# Fixtures de servicios
# ---------------------------------------------------------------------------

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


# ---------------------------------------------------------------------------
# Fixture de información del entorno (para debugging)
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session", autouse=True)
def print_environment(test_settings):
    """Imprime info del entorno al inicio del 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")

Tests Universales: Corren en Todos los Entornos

test_s3_operations.py

"""tests/test_s3_operations.py — Tests de S3 que corren en cualquier entorno."""

import json
import pytest


class TestS3PutAndGet:
    """Tests de operaciones básicas S3."""

    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 — Tests del processor en cualquier entorno."""

import json
import pytest


class TestDocumentProcessor:
    """Tests del DocumentProcessor contra el entorno actual."""

    @pytest.fixture(autouse=True)
    def setup_prompts(self, s3_client, test_bucket):
        """Crea prompt templates necesarios para los tests."""
        s3_client.put_object(
            Bucket=test_bucket,
            Key="prompts/summarizer/v1/system.txt",
            Body="Resume el documento en 3 puntos clave.".encode("utf-8"),
        )
        s3_client.put_object(
            Bucket=test_bucket,
            Key="prompts/classifier/v1/system.txt",
            Body="Clasifica el documento en una categoría.".encode("utf-8"),
        )

    def test_get_prompt_summarizer(self, processor):
        prompt = processor.get_prompt("summarizer", "v1")
        assert "Resume" in prompt
        assert "3 puntos" in prompt

    def test_get_prompt_classifier(self, processor):
        prompt = processor.get_prompt("classifier", "v1")
        assert "Clasifica" 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": "Contenido de prueba para testing multi-entorno.",
        }
        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 con Markers de Entorno

Tests AWS-only

"""tests/test_aws_only.py — Tests que solo corren en AWS."""

import pytest


@pytest.mark.aws_only
class TestAWSSpecific:
    """Tests que requieren AWS real (no LocalStack)."""

    def test_iam_role_exists(self, client_factory, test_settings):
        """Verifica que el IAM role de Lambda existe en 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 no configurado en este entorno")

    def test_s3_bucket_encryption(self, s3_client, test_settings):
        """Verifica que el bucket tiene encryption habilitado."""
        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 no configurado")

    def test_cloudwatch_log_group_exists(self, client_factory, test_settings):
        """Verifica que el log group de Lambda existe."""
        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} no encontrado"


@pytest.mark.sagemaker
class TestSageMaker:
    """Tests que requieren SageMaker (solo AWS con feature flag)."""

    def test_sagemaker_endpoint_exists(self, client_factory):
        """Verifica que el endpoint de SageMaker existe."""
        sm = client_factory.get_client("sagemaker")
        response = sm.list_endpoints(MaxResults=10)
        endpoints = response.get("Endpoints", [])
        assert len(endpoints) >= 0  # Verificar que la API responde

Tests LocalStack-only

"""tests/test_local_only.py — Tests específicos de LocalStack."""

import pytest
import requests


@pytest.mark.local_only
class TestLocalStackSpecific:
    """Tests que solo tienen sentido en LocalStack."""

    def test_localstack_health(self):
        """Verifica que LocalStack está saludable."""
        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):
        """Verifica que el servicio S3 está activo."""
        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):
        """Verifica que las operaciones locales son rápidas."""
        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 tardaron {elapsed:.2f}s — "
            f"LocalStack debería ser más rápido"
        )

pytest.ini: Configuración del Test Runner

# 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

Ejecución: Mismo Suite, Diferentes Entornos

Contra LocalStack

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

# Ejecutar tests
ENVIRONMENT=local pytest tests/ -v

# Output esperado:
# ============================================
# 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 (Requiere AWS)
# tests/test_aws_only.py::TestSageMaker::test_sagemaker_endpoint SKIPPED (SageMaker no habilitado)
# tests/test_local_only.py::TestLocalStackSpecific::test_localstack_health PASSED
#
# 39 passed, 8 skipped

Contra AWS Staging

# Asegurar credenciales AWS configuradas
aws sts get-caller-identity

# Ejecutar tests
ENVIRONMENT=staging pytest tests/ -v

# Output esperado:
# ============================================
# 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 (Requiere LocalStack)
#
# 43 passed, 4 skipped

Comparación automatizada

"""scripts/run_migration_tests.py — Corre tests en ambos entornos y compara."""

import subprocess
import json
import sys


def run_tests(environment: str) -> dict:
    """Ejecuta pytest contra un entorno y captura resultados."""
    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():
    """Corre tests en local y staging, compara resultados."""
    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

Problema 1: Tests pasan en local pero fallan en AWS

El comportamiento de S3 difiere ligeramente entre LocalStack y AWS.

# Diferencia común: LocalStack es más permisivo con bucket names
# AWS rechaza nombres con mayúsculas o underscores

# ❌ Funciona en LocalStack, falla en AWS
bucket = "My_Test_Bucket"

# ✅ Funciona en ambos
bucket = "my-test-bucket"

# Otra diferencia: consistency model
# LocalStack es always consistent, AWS S3 es strongly consistent
# (desde 2020), pero list_objects puede tener retraso mínimo

Problema 2: "Bucket already exists" al correr tests

El fixture test_bucket usa scope="session" pero si otro developer tiene el bucket:

# Solución: usar nombres de bucket con sufijo único
import hashlib
import os

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

Problema 3: Tests AWS-only se ejecutan en local

El marker no se está aplicando correctamente. Verificar conftest.py:

# Asegurar que pytest_collection_modifyitems está en conftest.py
# (no en otro archivo) y que usa el marker correcto:

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

# Debug: verificar el entorno
print(f"ENVIRONMENT = {os.environ.get('ENVIRONMENT')}")

Problema 4: Fixtures de session no se limpian entre runs

El bucket de test acumula objetos entre ejecuciones.

# Agregar un fixture de cleanup al final del session
@pytest.fixture(scope="session", autouse=True)
def cleanup_test_bucket(s3_client, test_bucket):
    yield
    # Cleanup después de todos los 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]},
            )

Ejercicios Prácticos

Ejercicio 1: Fixture parametrizada por entorno

Crea un fixture que parametrize los tests para que corran con dos configuraciones de bucket: una con prefijo v1/ y otra con prefijo v2/, verificando que el DocumentProcessor funciona con ambas estructuras.

Ver solución 
import pytest
import json
from services.document_processor import DocumentProcessor


@pytest.fixture(params=["v1", "v2"])
def versioned_processor(request, s3_client, test_bucket):
    """Fixture parametrizada: crea processor con diferentes versiones de prefix."""
    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

Ejercicio 2: Test de migración end-to-end

Escribe un test que simule una migración: sube datos a "local" (LocalStack), verifica que existen, luego verifica que los mismos datos son accesibles desde la configuración de "staging" (si disponible).

Ver solución 
import json
import pytest
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory


def test_migration_data_integrity(s3_client, test_bucket):
    """Verifica integridad de datos: sube, lista, y verifica contenido."""
    test_data = [
        {"id": f"migration-doc-{i}", "content": f"Document {i}", "index": i}
        for i in range(5)
    ]

    # 1. Subir datos
    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. Listar y verificar cantidad
    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. Verificar contenido de cada documento
    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. Limpiar
    s3_client.delete_objects(
        Bucket=test_bucket,
        Delete={
            "Objects": [
                {"Key": f"migration-test/{doc['id']}.json"}
                for doc in test_data
            ]
        },
    )

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

Ejercicio 3: Test de performance comparativo

Crea un test que mida el tiempo de 100 operaciones put_object y get_object, y lo guarde como resultado del test para comparar entre entornos.

Ver solución 
import time
import json
import pytest


class TestPerformance:
    """Tests de performance que reportan tiempos por entorno."""

    def test_put_object_throughput(self, s3_client, clean_bucket, test_settings):
        """Mide throughput de put_object."""
        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):
        """Mide throughput de get_object."""
        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"

Ejercicio 4: Fixture de feature flags para tests

Crea una fixture feature_flags que exponga las feature flags del entorno actual y un helper skip_if_feature_disabled que los tests usen para saltarse si una feature no está disponible.

Ver solución 
import pytest
from config.settings import Settings


@pytest.fixture(scope="session")
def feature_flags(test_settings: Settings) -> dict:
    """Feature flags del entorno actual."""
    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: salta el test si una feature no está habilitada."""
    if not feature_flags.get(feature, False):
        pytest.skip(f"Feature '{feature}' no habilitada en este entorno")


class TestWithFeatureFlags:
    def test_basic_always_runs(self, feature_flags):
        """Este test siempre corre."""
        assert isinstance(feature_flags, dict)

    def test_sagemaker_integration(self, feature_flags, client_factory):
        """Solo corre si SageMaker está habilitado."""
        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):
        """Solo corre si advanced logging está habilitado."""
        skip_if_disabled(feature_flags, "advanced_logging")
        assert test_settings.feature_advanced_logging is True

    def test_cost_tracking(self, feature_flags, test_settings):
        """Solo corre si cost tracking está habilitado."""
        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):
        """Imprime reporte de feature flags."""
        print(f"\nFeature Flags ({test_settings.environment}):")
        for flag, enabled in feature_flags.items():
            status = "✅ enabled" if enabled else "❌ disabled"
            print(f"  {flag}: {status}")

Resumen

  • Testing multi-entorno es la red de seguridad de la migración. Si tus tests pasan en LocalStack y en AWS, la migración está verificada.
  • conftest.py es el centro de control. Detecta el entorno, crea fixtures, aplica markers, configura cleanup. Todo en un archivo.
  • Markers (@pytest.mark.aws_only, @pytest.mark.local_only) permiten tests condicionales que se saltan automáticamente según el entorno.
  • El mismo test suite se ejecuta con ENVIRONMENT=local o ENVIRONMENT=staging. Los tests universales corren siempre. Los tests específicos se saltan automáticamente.
  • Fixtures con DI: el processor fixture recibe su S3 client del factory, configurado para el entorno actual. El test no sabe si habla con LocalStack o AWS.
  • En la siguiente cápsula, implementarás feature flags para manejar servicios que solo existen en ciertos entornos.

Recursos Adicionales

  1. pytest Documentation — Documentación oficial de pytest
  2. pytest Fixtures — Guía completa de fixtures
  3. pytest Markers — Markers para categorizar tests
  4. conftest.py — pytest — Compartir fixtures
  5. Testing boto3 Applications — Testing con boto3
  6. moto — Mock AWS Services — Framework de mocking para AWS (alternativa a LocalStack en tests)
← Anterior4. Dependency Injection para boto3 ClientsSiguiente →6. Feature Flags para Cloud