Cómo probar la infraestructura como código

Las pruebas de IaC significan validar tu código de infraestructura de la misma manera que pruebas el software de aplicación: pruebas unitarias con proveedores de nube simulados que se ejecutan en milisegundos, pruebas de integración que despliegan e inspeccionan recursos reales, y verificaciones de políticas que imponen reglas de cumplimiento en cada vista previa y despliegue. Juntas, estas capas capturan configuraciones incorrectas antes de que lleguen a producción. El código de infraestructura no probado es una responsabilidad. Una etiqueta que falta en una regla de grupo de seguridad, un bucket de S3 con acceso de lectura pública, una política IAM mal configurada: cualquiera de estas puede pasar desapercibida en la revisión de código y llegar a producción. Los errores de infraestructura suelen ser más difíciles de depurar que los errores de aplicación, porque el ciclo de retroalimentación es lento (desplegar, observar, destruir) y el radio de impacto es grande (una interrupción, un incidente de seguridad, una factura sorpresa de AWS). La buena noticia: Pulumi está construido sobre lenguajes de programación de propósito general, lo que significa que puedes probar el código de infraestructura con las mismas herramientas y marcos que ya utilizas para probar el código de aplicación. No hay un nuevo lenguaje que aprender. No hay una cadena de herramientas separada. Solo pytest, Mocha o go test, apuntando a tu programa de infraestructura. Esta guía recorre las tres capas de prueba con ejemplos completos y ejecutables. ¿Por qué deberías probar infraestructura como código? El argumento a favor de las pruebas de código de aplicación está bien establecido. El argumento a favor de las pruebas de IaC es igualmente fuerte, quizás más fuerte, porque las consecuencias de un error no detectado son más severas. Detecta configuraciones incorrectas antes de que se desplieguen. Una prueba unitaria que verifica si un grupo de seguridad permite SSH desde 0.0.0.0/0 capturará esa configuración incorrecta en milisegundos, antes de que se realice cualquier llamada a la API de la nube. La misma prueba en CI la detecta antes de que se fusione el código. Previene desviaciones y fallos. Pruebas de integración que se despliegan en un entorno de pruebas y validan el comportamiento en tiempo de ejecución: ¿el endpoint HTTP devuelve 200? ¿la base de datos acepta conexiones? te dan la confianza de que un cambio en la infraestructura no interrumpe silenciosamente un servicio dependiente. Refactoriza de manera segura. Un código de infraestructura bien probado puede reestructurarse, modularizarse y mejorarse sin miedo. Cuando todas tus propiedades, etiquetas y reglas de seguridad están afirmadas en pruebas, sabes de inmediato si un refactor rompe una restricción. Imponer seguridad y cumplimiento automáticamente. La política como código te permite codificar las reglas de seguridad de tu organización como pruebas ejecutables. Las políticas obligatorias bloquean por completo los despliegues no conformes; las políticas de asesoría presentan advertencias para revisión humana. Estas reglas se ejecutan en cada vista previa de Pulumi y Pulumi up, no solo en el momento de la auditoría. Despliega infraestructura a la velocidad del software. Los equipos que prueban IaC envían cambios más rápido, con menos miedo. El pipeline de CI se convierte en un motor de confianza en lugar de una puerta de despliegue. La investigación anual DORA State of DevOps se asocia constantemente con un patrón entre los equipos de ingeniería de alto rendimiento: aquellos que despliegan con frecuencia y se recuperan rápidamente de fallos: pruebas automatizadas completas en toda la pila. La infraestructura ya no está exenta de ese estándar. ¿Qué tipos de pruebas de IaC existen? Las pruebas de IaC reflejan la clásica pirámide de pruebas de software: una base amplia de pruebas unitarias rápidas y económicas, una capa más pequeña de pruebas de integración contra infraestructura real, y verificaciones de políticas que imponen reglas a lo largo de todo. Capa Qué prueba Velocidad Costo ¿Credenciales de nube? Pruebas unitarias Propiedades de recursos, etiquetas, reglas de seguridad, convenciones de nomenclatura Milisegundos Gratis No Pruebas de integración Ciclo de vida de la pila completa, comportamiento en tiempo de ejecución, dependencias entre recursos Minutos Gasto en la nube Sí Política como código Reglas de cumplimiento, estándares de seguridad, estándares organizacionales Milisegundos Gratis No (se ejecuta en el momento de la vista previa) Comienza con pruebas unitarias: son gratis, rápidas y capturan los errores más comunes. Agrega pruebas de integración para rutas críticas (el pipeline de despliegue, infraestructura de datos, recursos sensibles a la seguridad). Incorpora política como código para requisitos de cumplimiento que nunca deben retroceder. ¿Cómo escribes pruebas unitarias para Pulumi? Las pruebas unitarias de Pulumi reemplazan el canal de comunicación entre el programa de Pulumi y el proveedor de nube con simulaciones. Cuando el programa registra un recurso (por ejemplo, aws.ec2.SecurityGroup(...)), el simulador intercepta la llamada, devuelve un estado ficticio y nunca realiza una llamada real a la API de la nube. Las pruebas se ejecutan en el mismo proceso, en el mismo lenguaje, utilizando el mismo corredor de pruebas que ya utilizas. Configuración de simulaciones La regla de orden de importación: debes configurar simulaciones antes de importar tu programa de Pulumi. Si importas el programa primero, el tiempo de ejecución de Pulumi se inicializa sin simulaciones y intentará realizar llamadas a la nube reales. # test_infra.py import unittest import pulumi class MyMocks(pulumi.runtime.Mocks): def new_resource(self, args: pulumi.runtime.MockResourceArgs): # Retorna [id, estado]. Las claves de estado deben estar en camelCase. return [args.name + "_id", args.inputs] def call(self, args: pulumi.runtime.MockCallArgs): return {} # Establece simulaciones ANTES de importar el programa pulumi.runtime.set_mocks(MyMocks()) import infra # importa el programa DESPUÉS de establecer simulaciones class TestInfra(unittest.TestCase): @pulumi.runtime.test def test_no_public_ssh(self): """El grupo de seguridad no debe exponer SSH a internet público.""" def check_security_group(args): ingress_rules = args[0] for rule in (ingress_rules or []): for cidr in (rule.get("cidrBlocks") or []): self.assertNotEqual( cidr, "0.0.0.0/0", "El grupo de seguridad no debe exponer el puerto 22 a Internet." ) return pulumi.Output.all(infra.group.ingress).apply(check_security_group) @pulumi.runtime.test def test_server_has_tags(self): """La instancia EC2 debe tener las etiquetas requeridas.""" def check_tags(args): tags = args[0] self.assertIn("Environment", tags, "Falta la etiqueta 'Environment'") self.assertIn("Name", tags, "Falta la etiqueta 'Name'") return pulumi.Output.all(infra.server.tags).apply(check_tags) if __name__ == "__main__": unittest.main() Ejecuta con: python -m unittest discover -v // infra.test.ts import * as pulumi from "@pulumi/pulumi"; import { expect } from "chai"; // Establece simulaciones ANTES de la importación dinámica del programa pulumi.runtime.setMocks( { newResource: (args: pulumi.runtime.MockResourceArgs): { id: string; state: any } => { // Las claves de estado deben estar en camelCase return { id: `${args.name}-id`, state: args.inputs }; }, call: (args: pulumi.runtime.MockCallArgs) => ({ ...args.inputs }), }, "my-project", "test", false // vista previa = false ); describe("Infraestructura", function () { let infra: typeof import("./index"); before(async function () { // Importación dinámica DESPUÉS de establecer simulaciones infra = await import("./index"); }); it("no debe exponer SSH a internet", function (done) { pulumi.all([infra.group.ingress]).apply(([ingress]) => { for (const rule of ingress ?? []) { for (const cidr of rule.cidrBlocks ?? []) { expect(cidr).to.not.equal( "0.0.0.0/0", "El grupo de seguridad no debe exponer el puerto 22 a Internet" ); } } done(); }); }); it("debe tener etiquetas requeridas", function (done) { pulumi.all([infra.server.tags]).apply(([tags]) => { expect(tags).to.include.keys("Environment", "Name"); done(); }); }); }); Ejecuta con: mocha -r ts-node/register infra.test.ts // infra_test.go package infra_test import ( "testing" "github.com/pulumi/pulumi-aws/sdk/v6/go/aws/ec2" "github.com/pulumi/pulumi/sdk/v3/go/pulumi" "github.com/stretchr/testify/assert" ) // MockResourceMonitor implementa pulumi.ResourceMonitor para pruebas unitarias. func TestNoPublicSSH(t *testing.T) { err := pulumi.RunErr(func(ctx *pulumi.Context) error { infra, err := NewInfra(ctx) assert.NoError(t, err) infra.Group.Ingress.ApplyT(func(rules []ec2.SecurityGroupIngress) error { for _, rule := range rules { for _, cidr := range rule.CidrBlocks { assert.NotEqual(t, "0.0.0.0/0", cidr, "El grupo de seguridad no debe exponer el puerto 22 a Internet") } } return nil }) return nil }, pulumi.WithMocks("my-project", "test", mocks{})) assert.NoError(t, err) } type mocks struct{} func (mocks) NewResource(args pulumi.MockResourceArgs) (string, map[string]interface{}, error) { return args.Name + "_id", args.Inputs, nil } func (mocks) Call(args pulumi.MockCallArgs) (map[string]interface{}, error) { return args.Args, nil } Ejecuta con: go test ./... -v // InfraTests.cs usando System.Collections.Generic; usando System.Linq; usando System.Threading.Tasks; usando Pulumi; usando Pulumi.Testing; usando Xunit; public class InfraTests { [Fact] public async Task NoPublicSSH() { var resources = await Testing.RunAsync<MyStack>(); var sg = resources.OfType<Pulumi.Aws.Ec2.SecurityGroup>().First(); var ingress = await sg.Ingress.GetValueAsync(); foreach (var rule in ingress) { foreach (var cidr in rule.CidrBlocks) { Assert.NotEqual("0.0.0.0/0", cidr); } } } [Fact] public async Task ServerHasRequiredTags() { var resources = await Testing.RunAsync<MyStack>(); var server = resources.OfType<Pulumi.Aws.Ec2.Instance>().First(); var tags = await server.Tags.GetValueAsync(); Assert.True(tags.ContainsKey("Environment"), "Falta la etiqueta 'Environment'"); Assert.True(tags.ContainsKey("Name"), "Falta la etiqueta 'Name'"); } } Ejecuta con: dotnet test // InfraTest.java package myproject; import com.pulumi.aws.ec2.Instance; import com.pulumi.aws.ec2.SecurityGroup; import com.pulumi.test.Mocks; import com.pulumi.test.Mocks.CallArgs; import com.pulumi.test.Mocks.ResourceArgs; import com.pulumi.test.Mocks.ResourceResult; import com.pulumi.test.PulumiTest; import com.pulumi.test.TestOptions; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; import static org.junit.jupiter.api.Assertions.*; class InfraTest { // Implementa Mocks para interceptar la creación de recursos. Las claves del estado deben estar en camelCase. static class MyMocks implements Mocks { @Override public CompletableFuture<ResourceResult> newResourceAsync(ResourceArgs args) { var state = new HashMap<>(args.inputs); return CompletableFuture.completedFuture( ResourceResult.of(Optional.of(args.name + "_id"), state)); } @Override public CompletableFuture<Map<String, Object>> callAsync(CallArgs args) { return CompletableFuture.completedFuture(Map.of()); } } // PulumiTest.cleanup() debe ejecutarse después de cada prueba para restablecer el estado de tiempo de ejecución. @AfterEach void cleanup() { PulumiTest.cleanup(); } @Test void serverHasRequiredTags() { var result = PulumiTest .withMocks(new MyMocks()) .withOptions(TestOptions.builder() .projectName("project").stackName("stack").preview(false) .build()) .runTest(ctx -> new Infra()); var instance = result.resources().stream() .filter(r -> r instanceof Instance) .map(r -> (Instance) r) .findFirst() .orElseThrow(() -> new AssertionError("Instancia EC2 no encontrada")); var tags = PulumiTest.extractValue(instance.tags()) .orElseThrow(() -> new AssertionError("Las etiquetas no deben ser nulas")); assertTrue(tags.containsKey("Environment"), "Falta la etiqueta 'Environment'"); assertTrue(tags.containsKey("Name"), "Falta la etiqueta 'Name'"); } @Test void noPublicSshExposed() { var result = PulumiTest .withMocks(new MyMocks()) .withOptions(TestOptions.builder() .projectName("project").stackName("stack").preview(false) .build()) .runTest(ctx -> new Infra()); var group = result.resources().stream() .filter(r -> r instanceof SecurityGroup) .map(r -> (SecurityGroup) r) .findFirst() .orElseThrow(() -> new AssertionError("Grupo de seguridad no encontrado")); var rules = PulumiTest.extractValue(group.ingress()); for (var rule : rules) { for (var cidr : rule.cidrBlocks()) { assertNotEquals("0.0.0.0/0", cidr, "El grupo de seguridad no debe exponer SSH a internet"); } } } } Ejecuta con: mvn test (JUnit 5) El programa bajo prueba Ambos ejemplos anteriores prueban este programa: # infra.py import pulumi from pulumi_aws import ec2 group = ec2.SecurityGroup("web-secgrp", ingress=[{ "protocol": "tcp", "from_port": 80, "to_port": 80, "cidrBlocks": ["0.0.0.0/0"], }], ) server = ec2.Instance("web-server", instance_type="t2.micro", ami="ami-0b0ea68c435eb488d", vpc_security_group_ids=[group.id], tags={"Name": "web-server", "Environment": "dev"}, ) // index.ts import * as pulumi from "@pulumi/pulumi"; import * as aws from "@pulumi/aws"; export const group = new aws.ec2.SecurityGroup("web-secgrp", { ingress: [{ protocol: "tcp", fromPort: 80, toPort: 80, cidrBlocks: ["0.0.0.0/0"], }], }); export const server = new aws.ec2.Instance("web-server", { instanceType: "t2.micro", ami: "ami-0b0ea68c435eb488d", vpcSecurityGroupIds: [group.id], tags: { Name: "web-server", Environment: "dev" }, }); // main.go package infra_test import ( "github.com/pulumi/pulumi-aws/sdk/v6/go/aws/ec2" "github.com/pulumi/pulumi/sdk/v3/go/pulumi" ) type Infra struct { Group *ec2.SecurityGroup Server *ec2.Instance } func NewInfra(ctx *pulumi.Context) (*Infra, error) { group, err := ec2.NewSecurityGroup(ctx, "web-secgrp", &ec2.SecurityGroupArgs{ Ingress: ec2.SecurityGroupIngressArray{ ec2.SecurityGroupIngressArgs{ Protocol: pulumi.String("tcp"), FromPort: pulumi.Int(80), ToPort: pulumi.Int(80), CidrBlocks: pulumi.StringArray{pulumi.String("0.0.0.0/0")}, }, }, }) if err != nil { return nil, err } server, err := ec2.NewInstance(ctx, "web-server", &ec2.InstanceArgs{ InstanceType: pulumi.String("t2.micro"), Ami: pulumi.String("ami-0b0ea68c435eb488d"), VpcSecurityGroupIds: pulumi.StringArray{group.ID()}, Tags: pulumi.StringMap{ "Name": pulumi.String("web-server"), "Environment": pulumi.String("dev"), }, }) if err != nil { return nil, err } return &Infra{Group: group, Server: server}, nil } // MyStack.cs usando Pulumi; usando Pulumi.Aws.Ec2; usando Pulumi.Aws.Ec2.Inputs; public class MyStack : Stack { public MyStack() { var group = new SecurityGroup("web-secgrp", new SecurityGroupArgs { Ingress = new[] { new SecurityGroupIngressArgs { Protocol = "tcp", FromPort = 80, ToPort = 80, CidrBlocks = new[] { "0.0.0.0/0" }, }, }, }); var server = new Instance("web-server", new InstanceArgs { InstanceType = "t2.micro", Ami = "ami-0b0ea68c435eb488d", VpcSecurityGroupIds = new[] { group.Id }, Tags = new InputMap<string> { { "Name", "web-server" }, { "Environment", "dev" }, }, }); } } // Infra.java package myproject; import com.pulumi.Context; import com.pulumi.Pulumi; import com.pulumi.aws.ec2.SecurityGroup; import com.pulumi.aws.ec2.SecurityGroupArgs; import com.pulumi.aws.ec2.Instance; import com.pulumi.aws.ec2.InstanceArgs; import com.pulumi.aws.ec2.inputs.SecurityGroupIngressArgs; import java.util.List; import java.util.Map; public class Infra { public final SecurityGroup group; public final Instance server; public Infra() { this.group = new SecurityGroup("web-secgrp", SecurityGroupArgs.builder() .ingress(SecurityGroupIngressArgs.builder() .protocol("tcp") .fromPort(80) .toPort(80) .cidrBlocks("0.0.0.0/0") .build()) .build()); this.server = new Instance("web-server", InstanceArgs.builder() .instanceType("t2.micro") .ami("ami-0b0ea68c435eb488d") .vpcSecurityGroupIds(group.id().applyValue(List::of)) .tags(Map.of("Name", "web-server", "Environment", "dev")) .build()); } } Detalle clave: Las claves de propiedades de estado en el valor de retorno de new_resource/newResource del simulador deben estar en camelCase (cidrBlocks, no cidr_blocks), independientemente del lenguaje que estés utilizando. Así es como Pulumi serializa internamente las propiedades. ¿Cómo ejecutas pruebas de integración contra recursos de nube reales? Las pruebas unitarias te dicen si tus definiciones de recursos son correctas. Las pruebas de integración te dicen si tu infraestructura realmente funciona. Las pruebas de integración despliegan recursos de nube reales, ejecutan afirmaciones contra ellos y luego destruyen todo, dándote confianza de extremo a extremo a costa de un gasto real en la nube y tiempo. Usando la API de Automatización La API de Automatización de Pulumi te permite ejecutar pulumi up, pulumi destroy y cada otra operación CLI de Pulumi programáticamente, desde dentro de una prueba. defines la pila en línea (usando una función que ejecuta tu programa de Pulumi) o la apuntas a un directorio de proyecto existente. # test_integration.py import pytest import pulumi import pulumi_aws como aws from pulumi import automation as auto def create_bucket_program(): """El programa de Pulumi a desplegar durante la prueba.""" bucket = aws.s3.Bucket("test-bucket", tags={"Environment": "test", "ManagedBy": "pulumi"}, ) pulumi.export("bucket_name", bucket.id) pulumi.export("bucket_arn", bucket.arn) @pytest.fixture(scope="module") def deployed_stack(): """Desplegar la pila para el módulo de prueba y destruirla al finalizar.""" stack = auto.create_or_select_stack( stack_name="integration-test", project_name="bucket-test", program=create_bucket_program, ) stack.set_config("aws:region", auto.ConfigValue(value="us-west-2")) # Instalar plugins del proveedor stack.workspace.install_plugin("aws", "v7.34.0") # Desplegar up_result = stack.up(on_output=print) yield up_result.outputs # Teardown: destruir todo después de que el módulo de prueba esté completo stack.destroy(on_output=print) stack.workspace.remove_stack("integration-test") def test_bucket_was_created(deployed_stack): """El bucket de S3 debería haberse creado con el prefijo de nombre esperado.""" bucket_name = deployed_stack["bucket_name"].value assert bucket_name.startswith("test-bucket"), f"Nombre de bucket inesperado: {bucket_name}" def test_bucket_has_tags(deployed_stack): """Verifica que el bucket fue creado con las etiquetas requeridas (a través del SDK de AWS).""" import boto3 s3 = boto3.client("s3", region_name="us-west-2") bucket_name = deployed_stack["bucket_name"].value tagging = s3.get_bucket_tagging(Bucket=bucket_name) tag_map = {t["Key"]: t["Value"] for t in tagging["TagSet"]} assert tag_map.get("Environment") == "test" assert tag_map.get("ManagedBy") == "pulumi" Ejecuta con: pytest test_integration.py -v // integration.test.ts import { LocalWorkspace, ConfigValue, UpResult } from "@pulumi/pulumi/automation"; import * as aws from "@pulumi/aws"; import * as pulumi from "@pulumi/pulumi"; import { expect } from "chai"; describe("Prueba de Integración del Bucket S3", function () { this.timeout(300_000); // 5 min — los despliegues reales llevan tiempo let outputs: UpResult["outputs"]; let stackName = "integration-test-ts"; before(async function () { const stack = await LocalWorkspace.createOrSelectStack({ stackName, projectName: "bucket-test", program: async () => { const bucket = new aws.s3.Bucket("test-bucket", { tags: { Environment: "test", ManagedBy: "pulumi" }, }); return { bucketName: bucket.id, bucketArn: bucket.arn, }; }, }); await stack.setConfig("aws:region", { value: "us-west-2" }); await stack.workspace.installPlugin("aws", "v7.34.0"); const upResult = await stack.up({ onOutput: console.log }); outputs = upResult.outputs; }); after(async function () { const stack = await LocalWorkspace.selectStack({ stackName, projectName: "bucket-test", program: async () => {} }); await stack.destroy({ onOutput: console.log }); await stack.workspace.removeStack(stackName); }); it("crea un bucket con el prefijo de nombre correcto", function () { const name = outputs["bucketName"].value as string; expect(name).to.match(/^test-bucket/); }); }); Ejecuta con: mocha -r ts-node/register integration.test.ts // integration_test.go package integration_test import ( "context" "fmt" "os" "testing" "github.com/pulumi/pulumi-aws/sdk/v6/go/aws/s3" "github.com/pulumi/pulumi/sdk/v3/go/auto" "github.com/pulumi/pulumi/sdk/v3/go/auto/optup" "github.com/pulumi/pulumi/sdk/v3/go/pulumi" "github.com/stretchr/testify/assert" ) func TestBucketIntegration(t *testing.T) { ctx := context.Background() stack, err := auto.NewStackInlineSource(ctx, "integration-test", "bucket-test", func(ctx *pulumi.Context) error { bucket, err := s3.NewBucket(ctx, "test-bucket", &s3.BucketArgs{ Tags: pulumi.StringMap{ "Environment": pulumi.String("test"), "ManagedBy": pulumi.String("pulumi"), }, }) if err != nil { return err } ctx.Export("bucketName", bucket.ID()) return nil }, ) assert.NoError(t, err) _ = stack.SetConfig(ctx, "aws:region", auto.ConfigValue{Value: "us-west-2"}) upResult, err := stack.Up(ctx, optup.ProgressStreams(os.Stdout)) assert.NoError(t, err) t.Cleanup(func() { _, _ = stack.Destroy(ctx) _ = stack.Workspace().RemoveStack(ctx, "integration-test") }) name := fmt.Sprintf("%v", upResult.Outputs["bucketName"].Value) assert.Contains(t, name, "test-bucket") } Ejecuta con: go test ./... -v -timeout 10m // IntegrationTests.cs usando System; usando System.Threading.Tasks; usando Pulumi.Automation; usando Xunit; public class IntegrationTests : IAsyncLifetime { private WorkspaceStack _stack = null!; private UpResult _upResult = null!; public async Task InitializeAsync() { var program = PulumiFn.Create(() => { var bucket = new Pulumi.Aws.S3.Bucket("test-bucket", new() { Tags = new() { ["Environment"] = "test", ["ManagedBy"] = "pulumi" }, }); return new System.Collections.Generic.Dictionary<string, object?> { ["bucketName"] = bucket.Id, }; }); _stack = await LocalWorkspace.CreateOrSelectStackAsync(new InlineProgramArgs( projectName: "bucket-test", stackName: "integration-test", program: program )); await _stack.SetConfigAsync("aws:region", new ConfigValue("us-west-2")); _upResult = await _stack.UpAsync(new UpOptions { OnStandardOutput = Console.WriteLine }); } public async Task DisposeAsync() { await _stack.DestroyAsync(new DestroyOptions { OnStandardOutput = Console.WriteLine }); await _stack.Workspace.RemoveStackAsync("integration-test"); } [Fact] public void BucketNameHasExpectedPrefix() { var name = _upResult.Outputs["bucketName"].Value.ToString(); Assert.StartsWith("test-bucket", name); } } Ejecuta con: dotnet test // IntegrationTest.java package myproject; import com.pulumi.Context; import com.pulumi.automation.LocalWorkspace; import com.pulumi.automation.UpOptions; import com.pulumi.automation.DestroyOptions; import com.pulumi.automation.ConfigValue; import com.pulumi.automation.WorkspaceStack; import org.junit.jupiter.api.*; import java.util.Map; import static org.junit.jupiter.api.Assertions.*; @TestInstance(TestInstance.Lifecycle.PER_CLASS) class IntegrationTest { private static final String PROJECT_NAME = "bucket_test_project"; private static final String STACK_NAME = "integration-test"; private WorkspaceStack stack; private Map<String, com.pulumi.automation.OutputValue> outputs; @BeforeAll void setUp() throws Exception { stack = LocalWorkspace.createOrSelectStack(PROJECT_NAME, STACK_NAME, (Context ctx) -> { var bucket = new com.pulumi.aws.s3.Bucket("test-bucket", com.pulumi.aws.s3.BucketArgs.builder() .tags(Map.of("Environment", "test", "ManagedBy", "pulumi")) .build()); ctx.export("bucketName", bucket.id()); }); stack.workspace().installPlugin("aws", "v7.34.0"); stack.setConfig("aws:region", new ConfigValue("us-west-2")); var upResult = stack.up(UpOptions.builder().onStandardOutput(System.out::println).build()); outputs = upResult.outputs(); } @AfterAll void tearDown() throws Exception { if (stack != null) { stack.destroy(DestroyOptions.builder().onStandardOutput(System.out::println).build()); stack.workspace().removeStack(STACK_NAME); } } @Test void bucketNameHasExpectedPrefix() { var name = (String) outputs.get("bucketName").value(); assertTrue(name.startsWith("test-bucket"), "Nombre de bucket inesperado: " + name); } } Ejecuta con: mvn test Usando el marco de integración de Go Si prefieres pruebas de integración basadas en Go, Pulumi ofrece un marco de prueba basado en Go en github.com/pulumi/pulumi/pkg/v3/testing/integration. Maneja el ciclo completo de creación → validación → destrucción para cualquier programa de Pulumi (escrito en cualquier lenguaje), con un gancho de validación de tiempo de ejecución adicional para afirmaciones sobre el estado desplegado. func TestS3Website(t *testing.T) { integration.ProgramTest(t, &integration.ProgramTestOptions{ Dir: path.Join("..", "my-s3-website"), Config: map[string]string{"aws:region": "us-west-2"}, ExtraRuntimeValidation: func(t *testing.T, stack integration.RuntimeValidationStackInfo) { url := stack.Outputs["websiteUrl"].(string) resp, _ := http.Get("http://" + url) assert.Equal(t, 200, resp.StatusCode) }, }) } Mejor práctica: Siempre destruye pilas de prueba en un bloque finally o una función de limpieza de prueba. Las pilas de prueba de integración filtradas son una causa común de gastos inesperados en la nube. ¿Cómo complementa la política como código las pruebas de IaC? Las pruebas unitarias e de integración validan lo que hace tu infraestructura. La política como código impone a lo que debe conformarse tu infraestructura: estándares de seguridad, convenciones de etiquetado, requisitos de cumplimiento, y se ejecuta automáticamente en cada vista previa de Pulumi y Pulumi up. Las Políticas de Pulumi son actualmente compatibles en Python y TypeScript. Escribes políticas en el mismo lenguaje que el resto de tu pila, sin necesidad de aprender un DSL. Una política obligatoria bloquea completamente un despliegue cuando se infringe; una política de asesoría presenta una advertencia pero permite que el despliegue continúe. # __main__.py en tu directorio de paquete de políticas from pulumi_policy import ( EnforcementLevel, PolicyPack, ReportViolation, ResourceValidationArgs, ResourceValidationPolicy, ) REQUIRED_TAGS = {"Name", "Environment", "Owner"} def require_tags(args: ResourceValidationArgs, report_violation: ReportViolation): if args.resource_type.startswith("aws:"): tags = args.props.get("tags") or {} missing = REQUIRED_TAGS - set(tags.keys()) if missing: report_violation( f"El recurso carece de las etiquetas requeridas: {', '.join(sorted(missing))}" ) def no_public_s3(args: ResourceValidationArgs, report_violation: ReportViolation): if args.resource_type == "aws:s3/bucket:Bucket": if args.props.get("acl") in ("public-read", "public-read-write"): report_violation("Los buckets de S3 no deben tener un ACL público.") PolicyPack( name="aws-security-baseline", policies=[ ResourceValidationPolicy( name="require-tags", description="Todos los recursos de AWS deben tener etiquetas de Nombre, Entorno y Propietario.", enforcement_level=EnforcementLevel.MANDATORY, validate=require_tags, ), ResourceValidationPolicy( name="no-public-s3", description="Los buckets de S3 no deben ser accesibles públicamente a través de ACL.", enforcement_level=EnforcementLevel.MANDATORY, validate=no_public_s3, ), ], ) // index.ts en tu directorio de paquete de políticas import * as aws from "@pulumi/aws"; import { PolicyPack, validateResourceOfType } from "@pulumi/policy"; new PolicyPack("aws-security-baseline", { policies: [ { name: "require-tags", description: "Todas las instancias de EC2 deben tener etiquetas de Nombre, Entorno y Propietario.", enforcementLevel: "mandatory", validateResource: validateResourceOfType(aws.ec2.Instance, (instance, args, reportViolation) => { const required = ["Name", "Environment", "Owner"]; const tags = instance.tags || {}; for (const tag of required) { if (!tags[tag]) { reportViolation(`Falta etiqueta requerida: ${tag}`); } } }), }, { name: "no-public-s3", description: "Los buckets de S3 no deben tener un ACL público.", enforcementLevel: "mandatory", validateResource: validateResourceOfType(aws.s3.Bucket, (bucket, args, reportViolation) => { if (bucket.acl === "public-read" || bucket.acl === "public-read-write") { reportViolation("El bucket de S3 no debe tener un ACL público."); } }), }, ], }); Ejecutando un paquete de políticas # Crea un nuevo paquete de políticas desde una plantilla mkdir my-policies && cd my-policies pulumi policy new aws-typescript # Ejecuta verificaciones de políticas contra una pila pulumi preview --policy-pack ./my-policies pulumi up --policy-pack ./my-policies # O publica en Pulumi Cloud y aplica a nivel organizacional pulumi policy publish Como dice Joe Duffy, CEO de Pulumi, “el agente más inteligente del mundo aún necesita barreras, registros de auditoría y aplicación de políticas para ser confiable con sistemas de producción a gran escala.” Las verificaciones de políticas complementan las pruebas unitarias: las pruebas verifican tu intención (“escribí etiquetas en este recurso”), mientras que las políticas imponen tus estándares (“todos los recursos deben tener estas etiquetas, independientemente de lo que un desarrollador haya pretendido”). Para entornos de cumplimiento complejos, publica políticas en Pulumi Cloud y asígnalas a través de grupos de políticas para aplicarlas en las pilas o cuentas que elijas, incluyendo a nivel organizacional. ¿Cómo es diferente las pruebas de IaC en Pulumi a Terraform? La principal diferencia es la cohesión del lenguaje. Con Pulumi, escribes infraestructura, pruebas y políticas todo en el mismo lenguaje de propósito general. Con Terraform, tu infraestructura está en HCL, pero tus pruebas deben ser algo distinto: ya sea Go (Terratest) o un DSL HCL restringido (terraform test). Pulumi Terratest terraform test (≥ 1.6) Lenguaje para pruebas Mismo lenguaje que la infra (Python, TypeScript, Go, C#, Java); programas YAML probados a través de API de Automatización de cualquiera de esos lenguajes Siempre Go (sin importar el lenguaje de infra) HCL DSL (.tftest.hcl) Pruebas unitarias con simulaciones Sí — pulumi.runtime.set_mocks(), no se necesitan credenciales de nube No — sin capa simulada para proveedores de nube Limitado — mock_provider (v1.7+) es HCL-declarativo, sin lógica programática Pruebas de integración API de Automatización (cualquier lenguaje) o marco de Go Completamente funcional pero solo para Go Sí, a través de bloques command = apply run Políticas / barreras Políticas de Pulumi: Python o TypeScript, se ejecuta en el momento de la vista previa Herramientas externas (checkov, tfsec, OPA/Rego — cadena de herramientas separada) Sentinel (enterprise) o externo Velocidad de ejecución Unidad: milisegundos; integración: depende de la nube Minutos (despliegues reales por defecto) Planificar solo: rápido; aplicar: minutos Curva de aprendizaje Usa habilidades y marcos de lenguaje existentes Requiere conocimientos de Go y comprensión de go test Baja si conoces HCL; expresividad limitada Terratest es ampliamente adoptado y probado en batalla, particularmente para pruebas de módulos de Terraform. Ofrece profundas integraciones con AWS, GCP, Azure y Kubernetes. Su limitación es que no tiene capa simulada para proveedores de nube, por lo que la mayoría de los conjuntos de pruebas despliegan infraestructura real, lo que hace que sean más lentos y costosos de ejecutar en cada commit. terraform test ha mejorado significativamente en Terraform 1.6 y 1.7. El modo command = plan evita despliegues reales, y mock_provider habilita pruebas básicas similares a las de unidad. Pero la expresividad está limitada por el modelo declarativo de HCL—sin programación de propósito general, sin bibliotecas de afirmación de terceros y un conjunto fijo de constructos de prueba run/assert. Con Pulumi, no hay cambio de contexto. El mismo desarrollador que escribe la infraestructura escribe las pruebas en el mismo lenguaje, usando el mismo IDE, el mismo depurador y el mismo pipeline CI. Esa cohesión es un multiplicador de productividad. Cómo configurar pruebas de IaC con Pulumi Sigue estos pasos para agregar un conjunto de pruebas completo a un proyecto de Pulumi. Los ejemplos utilizan Python, pero el patrón es idéntico en TypeScript, Go, C#, o Java. Paso 1: Instalar Pulumi y elegir un SDK de lenguaje # Instalar la CLI de Pulumi curl -fsSL https://get.pulumi.com | sh # Crear un nuevo proyecto (o usar uno existente) pulumi new aws-python --name my-project --stack dev Paso 2: Agregar un marco de pruebas # Python pip install pytest pytest-asyncio # TypeScript npm install --save-dev mocha ts-node @types/mocha chai @types/chai # Go usa el paquete de pruebas integrado — sin dependencias adicionales Paso 3: Implementar la clase Mocks Crear mocks.py (o equivalente) con tu subclase de Mocks. Mantenlo simple para pruebas unitarias: retorna [args.name + "_id", args.inputs] desde new_resource. Paso 4: Escribir pruebas unitarias Crear test_infra.py en la raíz de tu proyecto. Importa primero los mocks, luego tu programa. Escribe funciones de prueba que usen @pulumi.runtime.test y pulumi.Output.all(...).apply(...) para afirmar propiedades de recursos. Paso 5: Agregar pruebas de integración Crear test_integration.py con una función de pytest que crea una pila, llama a stack.up(), entrega salidas para afirmaciones, y llama a stack.destroy() en la limpieza. Apunta esto a una cuenta de nube de prueba dedicada. Paso 6: Crear un paquete de políticas mkdir policies && cd policies pulumi policy new aws-python # Edita __main__.py para agregar tus reglas Paso 7: Conectar todo en CI # .github/workflows/test.yml (fragmento de GitHub Actions) jobs: unit-tests: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: pulumi/actions@v6 - run: pip install -r requirements.txt - run: python -m pytest tests/unit/ -v integration-tests: runs-on: ubuntu-latest needs: unit-tests environment: staging steps: - uses: actions/checkout@v4 - uses: pulumi/actions@v6 - run: pip install -r requirements.txt - env: PULUMI_ACCESS_TOKEN: ${{ secrets.PULUMI_ACCESS_TOKEN }} AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }} run: python -m pytest tests/integration/ -v --timeout=300 Preguntas frecuentes ¿Qué es la prueba de infraestructura como código? La prueba de infraestructura como código es la práctica de escribir verificaciones automatizadas que validan tus programas de IaC antes de que se desplieguen en producción. Al igual que las pruebas de aplicación, incluye pruebas unitarias (rápidas, sin credenciales de nube, simula el proveedor), pruebas de integración (despliegues de nube reales en entornos efímeros) y verificaciones de políticas (reglas que imponen estándares de cumplimiento en cada vista previa y despliegue). ¿Puedes probar unitariamente IaC sin desplegar en la nube? Sí. El marco de simulación de Pulumi intercepta todas las llamadas a la API de nube y las reemplaza con simulaciones en proceso. Las pruebas unitarias se ejecutan completamente en memoria: sin credenciales de nube, sin llamadas de red, sin costo. Un conjunto de 50 pruebas unitarias se ejecuta en menos de un segundo. Esta es la razón principal para escribir pruebas unitarias: comentarios rápidos, gratuitos y deterministas en cada commit. ¿Necesitas credenciales de nube reales para probar Pulumi? Solo para pruebas de integración. Las pruebas unitarias usan el tiempo de ejecución simulado de Pulumi y no requieren credenciales de nube en absoluto. Las verificaciones de políticas también se ejecutan sin credenciales en el momento de la vista previa de Pulumi. Las pruebas de integración que utilizan la API de Automatización sí necesitan credenciales, ya que despliegan recursos reales, por lo que normalmente se ejecutan en una etapa CI con acceso a una cuenta de prueba dedicada, no en cada PR. ¿Cuál es la diferencia entre pruebas unitarias y de integración para IaC? Las pruebas unitarias validan propiedades de recursos, configuraciones y relaciones usando proveedores de nube simulados: se ejecutan en milisegundos sin credenciales de nube. Las pruebas de integración despliegan recursos reales en un entorno de nube real, validan el comportamiento en tiempo de ejecución (¿responde el endpoint? ¿acepta la base de datos conexiones?), y luego destruyen todo. Las pruebas unitarias son económicas y frecuentes; las pruebas de integración son más lentas, implican un gasto real en la nube y se ejecutan con menos frecuencia. ¿Cómo simulas recursos de nube en Pulumi? Implementa pulumi.runtime.Mocks (Python) o llama a pulumi.runtime.setMocks() (TypeScript/JavaScript) antes de importar tu programa de Pulumi. El método new_resource recibe cada registro de recurso y retorna un par falso [id, estado]. El método call maneja las búsquedas de datos. Establece simulaciones con pulumi.runtime.set_mocks(MyMocks()) y luego importa tu programa; el tiempo de ejecución dirigirá todos los registros de recursos a través de tu simulación en lugar de realizar llamadas API reales. ¿Qué marcos de prueba funcionan con Pulumi? Cualquier marco de prueba nativo del lenguaje funciona. Para Python: pytest (recomendado) o unittest. Para TypeScript/JavaScript: Mocha, Jest o Vitest. Para Go: el paquete de pruebas estándar. Para C#: NUnit o xUnit. Para Java: JUnit. El SDK de @pulumi/pulumi/automation y el módulo pulumi.automation se integran de manera natural con todos ellos. YAML es el lenguaje de configuración declarativa de Pulumi; debido a que no tiene flujo de control o declaraciones ejecutables, no puedes escribir lógica de prueba en YAML mismo. Los programas YAML se prueban apuntando la API de Automatización desde Python, TypeScript, Go, C#, o Java al directorio del proyecto y afirmando sobre las salidas de la pila. ¿Cómo pruebas Pulumi en CI/CD? Divide tu suite de pruebas en dos etapas de CI. Etapa 1 (pruebas unitarias): se ejecuta en cada commit de PR, sin requerir credenciales de nube, rápida. Etapa 2 (pruebas de integración): se ejecuta al fusionarse en main o en un horario, requiere credenciales de nube para una cuenta de prueba dedicada, usa la API de Automatización para desplegar pilas efímeras. La integración de acciones de GitHub de Pulumi simplifica la inyección de credenciales y la gestión de pilas en ambas etapas. ¿Es la política como código lo mismo que probar? La política como código es complementaria, pero distinta de las pruebas. Las pruebas validan lo que produce tu programa (el recurso A tiene la propiedad B); las políticas imponen a qué deben conformarse tus recursos (todos los recursos deben tener la etiqueta X). Las políticas se ejecutan automáticamente en el momento de la vista previa y bloquean despliegues no conformes en el momento de despliegue. Juntos, crean una defensa en profundidad: las pruebas verifican la intención, las políticas imponen estándares sin importar la intención. ¿Cómo es diferente la prueba de Pulumi a Terratest? Terratest típicamente despliega infraestructura real: no tiene capa simulada para proveedores de nube, por lo que la mayoría de las ejecuciones de prueba provisionan recursos reales en la nube, incurriendo en un costo real y tomando de 5 a 30 minutos. Las pruebas deben escribirse en Go independientemente del lenguaje que utilice tu infraestructura. El marco de pruebas unitarias de Pulumi se ejecuta en milisegundos, sin credenciales de nube, utiliza tu lenguaje existente y corredor de pruebas, y proporciona una verdadera aislación simulada. Para pruebas de integración, la API de Automatización de Pulumi ofrece un modelo similar a Terratest pero en cualquier lenguaje. ¿Cómo desmantelas recursos después de las pruebas de integración? En Python pytest, usa un fixture de módulo que entrega después de stack.up() y llama a stack.destroy() en la limpieza. En TypeScript Mocha, llama a stack.destroy() en un gancho after(). En Go, usa defer stack.Destroy(ctx, ...). Siempre usa un patrón try/finally (o equivalente en tu lenguaje) para que la limpieza se ejecute incluso cuando falla una afirmación de prueba. Pulumi Cloud también tiene características de TTL de pilas incorporadas y auto-destrucción para mayor seguridad. ¿Dónde ir a continuación? Probar IaC no es una inversión total o nada. Comienza agregando unas pocas pruebas unitarias a tu pila más crítica: la que gestiona tu VPC de producción, tus roles de IAM, o tu infraestructura de datos. Elige entre tres y cinco propiedades que, si son incorrectas, causarían una interrupción o un incidente de seguridad. Escribe pruebas para esas. Ejecútalas en CI. Expande a partir de ahí. Cuando estés listo para profundizar: Visión general de pruebas — La documentación completa de pruebas de Pulumi Guía de pruebas unitarias — referencia detallada de API de simulación con ejemplos en Go y C# Pruebas de integración de API de Automatización — patrones avanzados (pruebas de múltiples pilas, inyección de configuración, ejecución de pruebas en paralelo) Redacción de políticas como código — escritura y publicación de Políticas de Pulumi Pulumi vs. Terraform — una comparación completa de las dos plataformas Si estás migrando desde Terraform, la herramienta de conversión de Pulumi traduce tu HCL existente a Python, TypeScript, Go, C#, y más, incluyendo tu infraestructura de prueba. Tus suites de prueba de Terratest o terraform test pueden ser portadas al corredor nativo de pruebas de Pulumi, usando tu lenguaje existente, como parte de la migración. Comienza a usar Pulumi gratis →