为 Pulumi 程序编写单元测试-未完纪

Pulumi 程序可以使用 TypeScript、Python、Go、.NET 或 Java 等通用语言，因此也能使用这些语言的工具、库和测试框架。

执行更新时，程序会与 Pulumi CLI 通信，由它协调部署。单元测试切断这条通信通道，以 mock 替代引擎。mock 在同一个操作系统进程内响应程序的调用，为每次调用返回占位数据。

mock 不执行真实部署，所以测试速度快；由于不依赖外部系统的行为，也容易得到确定的结果。

准备工作

本教程用 AWS 资源构建测试套件，同样的方法也适用于其他 Pulumi provider。先完成 AWS 入门教程，用所选语言建立 Pulumi 程序。所有通用 Pulumi 语言运行时均支持单元测试。

示例程序

下面测试一个创建 AWS EC2 Web 服务器的程序，并检查三条规则：

实例必须有 Name 标签。
实例不能使用内联 userData 脚本，而应使用虚拟机镜像。
实例不能向整个互联网开放 SSH。

基于 mock 的单元测试需要通用语言运行时。YAML 或 HCL 编写的声明式程序，应参考集成测试。

TypeScript

import * as aws from "@pulumi/aws";

export const group = new aws.ec2.SecurityGroup("web-secgrp", {
    ingress: [
        { protocol: "tcp", fromPort: 22, toPort: 22, cidrBlocks: ["0.0.0.0/0"] },
        { protocol: "tcp", fromPort: 80, toPort: 80, cidrBlocks: ["0.0.0.0/0"] },
    ],
});

const userData = `#!/bin/bash echo "Hello, World!" > index.html nohup python3 -m http.server 80 &`;

// Look up the latest Amazon Linux 2 AMI.
const ami = aws.ec2.getAmiOutput({
    owners: ["amazon"],
    mostRecent: true,
    filters: [{ name: "name", values: ["amzn2-ami-hvm-*-x86_64-gp2"] }],
});

export const server = new aws.ec2.Instance("web-server-www", {
    instanceType: "t2.micro",
    securityGroups: [group.name], // reference the group object above
    ami: ami.id,
    userData: userData, // start a simple web server
});

Python

import pulumi
from pulumi_aws import ec2

group = ec2.SecurityGroup('web-secgrp', ingress=[
    { "protocol": "tcp", "from_port": 22, "to_port": 22, "cidr_blocks": ["0.0.0.0/0"] },
    { "protocol": "tcp", "from_port": 80, "to_port": 80, "cidr_blocks": ["0.0.0.0/0"] },
])

user_data = '#!/bin/bash echo "Hello, World!" > index.html nohup python3 -m http.server 80 &'

# Look up the latest Amazon Linux 2 AMI.
ami = ec2.get_ami_output(
    owners=["amazon"],
    most_recent=True,
    filters=[{"name": "name", "values": ["amzn2-ami-hvm-*-x86_64-gp2"]}])

server = ec2.Instance('web-server-www',
    instance_type="t2.micro",
    security_groups=[ group.name ], # reference the group object above
    ami=ami.id,
    user_data=user_data)            # start a simple web server

Go

package main

import (
	"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ec2"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

type infrastructure struct {
	group  *ec2.SecurityGroup
	server *ec2.Instance
}

func createInfrastructure(ctx *pulumi.Context) (*infrastructure, error) {
	group, err := ec2.NewSecurityGroup(ctx, "web-secgrp", &ec2.SecurityGroupArgs{
		Ingress: ec2.SecurityGroupIngressArray{
			ec2.SecurityGroupIngressArgs{
				Protocol:   pulumi.String("tcp"),
				FromPort:   pulumi.Int(22),
				ToPort:     pulumi.Int(22),
				CidrBlocks: pulumi.StringArray{pulumi.String("0.0.0.0/0")},
			},
			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
	}

	const userData = `#!/bin/bash echo "Hello, World!" > index.html nohup python3 -m http.server 80 &`

	// Look up the latest Amazon Linux 2 AMI.
	ami, err := ec2.LookupAmi(ctx, &ec2.LookupAmiArgs{
		Owners:     []string{"amazon"},
		MostRecent: pulumi.BoolRef(true),
		Filters: []ec2.GetAmiFilter{
			{
				Name:   "name",
				Values: []string{"amzn2-ami-hvm-*-x86_64-gp2"},
			},
		},
	})
	if err != nil {
		return nil, err
	}

	server, err := ec2.NewInstance(ctx, "web-server-www", &ec2.InstanceArgs{
		InstanceType:   pulumi.String("t2.micro"),
		SecurityGroups: pulumi.StringArray{group.Name}, // reference the group object above
		Ami:            pulumi.String(ami.Id),
		UserData:       pulumi.String(userData), // start a simple web server
	})
	if err != nil {
		return nil, err
	}

	return &infrastructure{
		group:  group,
		server: server,
	}, nil
}

func main() {
	pulumi.Run(func(ctx *pulumi.Context) error {
		_, err := createInfrastructure(ctx)
		return err
	})
}

C#

using Pulumi;
using Pulumi.Aws.Ec2;
using Pulumi.Aws.Ec2.Inputs;
using System.Threading.Tasks;

class Program
{
    static Task<int> Main() => Deployment.RunAsync<WebserverStack>();
}

public class WebserverStack : Stack
{
    public WebserverStack()
    {
        var group = new SecurityGroup("web-secgrp", new SecurityGroupArgs
        {
            Ingress =
            {
                new SecurityGroupIngressArgs { Protocol = "tcp", FromPort = 22, ToPort = 22, CidrBlocks = { "0.0.0.0/0" } },
                new SecurityGroupIngressArgs { Protocol = "tcp", FromPort = 80, ToPort = 80, CidrBlocks = { "0.0.0.0/0" } }
            }
        });

        var userData = "#!/bin/bash echo \"Hello, World!\" > index.html nohup python3 -m http.server 80 &";

        // Look up the latest Amazon Linux 2 AMI.
        var ami = GetAmi.Invoke(new GetAmiInvokeArgs
        {
            Owners = { "amazon" },
            MostRecent = true,
            Filters =
            {
                new GetAmiFilterInputArgs { Name = "name", Values = { "amzn2-ami-hvm-*-x86_64-gp2" } }
            }
        });

        var server = new Instance("web-server-www", new InstanceArgs
        {
            InstanceType = "t2.micro",
            SecurityGroups = { group.Name }, // reference the group object above
            Ami = ami.Apply(ami => ami.Id),
            UserData = userData              // start a simple web server
        });
    }
}

Java

package myproject;

import java.util.List;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.aws.ec2.Ec2Functions;
import com.pulumi.aws.ec2.Instance;
import com.pulumi.aws.ec2.InstanceArgs;
import com.pulumi.aws.ec2.SecurityGroup;
import com.pulumi.aws.ec2.SecurityGroupArgs;
import com.pulumi.aws.ec2.inputs.GetAmiArgs;
import com.pulumi.aws.ec2.inputs.GetAmiFilterArgs;
import com.pulumi.aws.ec2.inputs.SecurityGroupIngressArgs;
import com.pulumi.aws.ec2.outputs.GetAmiResult;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        var group = new SecurityGroup("web-secgrp", SecurityGroupArgs.builder()
            .ingress(
                SecurityGroupIngressArgs.builder()
                    .protocol("tcp").fromPort(22).toPort(22).cidrBlocks("0.0.0.0/0")
                    .build(),
                SecurityGroupIngressArgs.builder()
                    .protocol("tcp").fromPort(80).toPort(80).cidrBlocks("0.0.0.0/0")
                    .build())
            .build());

        var userData = "#!/bin/bash echo \"Hello, World!\" > index.html nohup python3 -m http.server 80 &";

        // Look up the latest Amazon Linux 2 AMI.
        var ami = Ec2Functions.getAmi(GetAmiArgs.builder()
            .owners("amazon")
            .mostRecent(true)
            .filters(GetAmiFilterArgs.builder()
                .name("name")
                .values("amzn2-ami-hvm-*-x86_64-gp2")
                .build())
            .build());

        var server = new Instance("web-server-www", InstanceArgs.builder()
            .instanceType("t2.micro")
            .securityGroups(group.name().applyValue(List::of))  // reference the group object above
            .ami(ami.applyValue(GetAmiResult::id))
            .userData(userData)            // start a simple web server
            .build());
    }
}

程序创建一个安全组和一个实例，通过 getAmi 查询 AMI，而不是把镜像 ID 写死。它故意违反上述全部三条规则，随后编写的测试应捕获这些问题。

安装测试框架

可以选择自己熟悉的测试框架及断言库。这里提供五种语言的配置。

TypeScript

使用 Mocha，通过 tsx 直接执行 TypeScript。把它们作为开发依赖安装：

npm install --save-dev mocha @types/mocha tsx

Python

使用 Python 自带的 unittest，无须另行安装。

Go

使用内置的 go test，搭配 testify 断言：

go get github.com/stretchr/testify

C#

使用 NUnit 定义并运行测试，使用 FluentAssertions 编写断言。mock 来自 Pulumi SDK 的 Pulumi.Testing 命名空间，无须额外安装 mocking 库。向项目添加以下 NuGet 包：

dotnet add package NUnit
dotnet add package NUnit3TestAdapter
dotnet add package FluentAssertions
dotnet add package Microsoft.NET.Test.Sdk
dotnet add package Pulumi
dotnet add package Pulumi.Aws

Java

使用 JUnit 5，在 pom.xml 中添加以下依赖：

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter-api</artifactId>
    <version>5.10.0</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter-engine</artifactId>
    <version>5.10.0</version>
    <scope>test</scope>
</dependency>

添加 mock

mock 在同一进程内替代 Pulumi CLI 回答请求。它有两类处理器：newResource 响应资源注册；call 响应 provider 函数调用，包括示例中的 AMI 查询。下面先给出对应文件，再解释各处理器。

TypeScript

文件：ec2tests.ts

import * as pulumi from "@pulumi/pulumi";

pulumi.runtime.setMocks({
    newResource: function(args: pulumi.runtime.MockResourceArgs): {id: string, state: any} {
        switch (args.type) {
            case "aws:ec2/securityGroup:SecurityGroup":
                return {
                    id: "sg-12345678",
                    state: {
                        ...args.inputs,
                        // Mock output properties that may be used in tests
                        arn: "arn:aws:ec2:us-west-2:123456789012:security-group/sg-12345678",
                        name: args.inputs.name || args.name + "-sg",
                    },
                };
            case "aws:ec2/instance:Instance":
                return {
                    id: "i-1234567890abcdef0",
                    state: {
                        ...args.inputs,
                        // Mock output properties that may be used in tests
                        arn: "arn:aws:ec2:us-west-2:123456789012:instance/i-1234567890abcdef0",
                        instanceState: "running",
                        primaryNetworkInterfaceId: "eni-12345678",
                        privateDns: "ip-10-0-1-17.ec2.internal",
                        publicDns: "ec2-203-0-113-12.compute-1.amazonaws.com",
                        publicIp: "203.0.113.12",
                    },
                };
            default:
                return {
                    id: args.inputs.name + "_id",
                    state: {
                        ...args.inputs,
                    },
                };
        }
    },
    call: function(args: pulumi.runtime.MockCallArgs) {
        switch (args.token) {
            case "aws:ec2/getAmi:getAmi":
                return {
                    id: "ami-0eb1f3cdeeb8eed2a",
                    architecture: "x86_64",
                };
            default:
                return args.inputs;
        }
    },
},
  "project", // Project name. Mocked resources get it in their URNs.
  "stack",   // Stack name. Also part of the URN.
  false,     // Sets the flag `dryRun`, which indicates if pulumi is running in preview mode.
);

Python

文件：test_ec2.py

import pulumi

class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        return [args.name + "_id", args.inputs]

    def call(self, args: pulumi.runtime.MockCallArgs):
        if args.token == "aws:ec2/getAmi:getAmi":
            return {
                "id": "ami-0eb1f3cdeeb8eed2a",
                "architecture": "x86_64",
            }
        return {}

注意：new_resource 显式返回输出属性时，名称必须采用 camelCase，例如 publicIp、instanceState，而不是 public_ip。Pulumi 的内部属性序列化使用 camelCase，与程序语言无关。

Go

文件：main_test.go

import (
	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

type mocks int

func (mocks) NewResource(args pulumi.MockResourceArgs) (string, resource.PropertyMap, error) {
	return args.Name + "_id", args.Inputs, nil
}

func (mocks) Call(args pulumi.MockCallArgs) (resource.PropertyMap, error) {
	if args.Token == "aws:ec2/getAmi:getAmi" {
		return resource.NewPropertyMapFromMap(map[string]interface{}{
			"id":           "ami-0eb1f3cdeeb8eed2a",
			"architecture": "x86_64",
		}), nil
	}
	return args.Args, nil
}

C#

文件：Testing.cs

using System.Collections.Generic;
using System.Collections.Immutable;
using System.Threading.Tasks;
using Pulumi;
using Pulumi.Testing;

namespace UnitTesting
{
    class Mocks : IMocks
    {
        public Task<(string? id, object state)> NewResourceAsync(MockResourceArgs args)
        {
            var outputs = ImmutableDictionary.CreateBuilder<string, object>();

            outputs.AddRange(args.Inputs);

            if (args.Type == "aws:ec2/instance:Instance")
            {
                outputs.Add("publicIp", "203.0.113.12");
                outputs.Add("publicDns", "ec2-203-0-113-12.compute-1.amazonaws.com");
            }

            args.Id ??= $"{args.Name}_id";
            return Task.FromResult<(string? id, object state)>((args.Id, (object)outputs));
        }

        public Task<object> CallAsync(MockCallArgs args)
        {
            if (args.Token == "aws:ec2/getAmi:getAmi")
            {
                return Task.FromResult<object>(new Dictionary<string, object>
                {
                    { "id", "ami-0eb1f3cdeeb8eed2a" },
                    { "architecture", "x86_64" },
                });
            }

            return Task.FromResult((object)ImmutableDictionary<string, object>.Empty);
        }
    }

    public static class Testing
    {
        public static Task<ImmutableArray<Resource>> RunAsync<T>() where T : Stack, new()
        {
            return Deployment.TestAsync<T>(new Mocks(), new TestOptions { IsPreview = false });
        }

        public static Task<T> GetValueAsync<T>(this Output<T> output)
        {
            var tcs = new TaskCompletionSource<T>();
            output.Apply(v =>
            {
                tcs.SetResult(v);
                return v;
            });
            return tcs.Task;
        }
    }
}

Java

文件：Ec2Tests.java

package myproject;

import com.pulumi.test.Mocks;
import com.pulumi.test.Mocks.CallArgs;
import com.pulumi.test.Mocks.ResourceArgs;
import com.pulumi.test.Mocks.ResourceResult;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

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) {
        if ("aws:ec2/getAmi:getAmi".equals(args.token)) {
            return CompletableFuture.completedFuture(Map.of(
                "id", "ami-0eb1f3cdeeb8eed2a",
                "architecture", "x86_64"
            ));
        }
        return CompletableFuture.completedFuture(Map.of());
    }
}

完整 mock 接口见 Node.js runtime API 参考。

模拟资源

newResource 一次接收一个资源注册，返回原本由引擎返回的 ID 和状态。经过它的属性分为两类：

输入属性：如 tags、userData、ingress，由程序提供，出现在 mock 参数中。通常可以原样返回。
输出属性：如 arn、publicIp、instanceState，原本由云服务计算。mock 必须显式返回，否则值可能未定义。

测试可能读取这两类属性，因此示例按资源类型分支：每种资源需要不同的输出属性集合。

模拟 provider 函数

aws.ec2.getAmi、aws.getAvailabilityZones 等函数不创建资源，而是查询 provider，所以由 call 处理。固定查询返回值，可以让依赖实时云查询的测试获得确定结果。

每个函数通过 <package>:<module>/<function>:<function> 形式的 token 标识。例如 AWS provider 的 EC2 模块中，getAmi 的 token 是 aws:ec2/getAmi:getAmi。在 Pulumi Registry 对应函数页和 provider schema 中可以找到它，并据此选择返回值。

call 还收到程序传入的参数，即 args.inputs，Go 中为 args.Args。同一个 mock 因而可以按查询条件返回不同结果，例如按过滤器区分 Amazon Linux 和 Ubuntu 镜像。

以下片段来自另外一组精简、独立的项目，便于把 mock 与对应断言放在一起，并非前面 Web 服务器示例的延续。

TypeScript

在 call 中匹配 token；其他函数使用默认分支：

pulumi.runtime.setMocks({
    newResource: function(args: pulumi.runtime.MockResourceArgs): {id: string, state: any} {
        return {
            id: args.name + "_id",
            state: args.inputs,
        };
    },
    call: function(args: pulumi.runtime.MockCallArgs) {
        switch (args.token) {
            case "aws:ec2/getAmi:getAmi":
                return {
                    id: "ami-0eb1f3cdeeb8eed2a",
                    architecture: "x86_64",
                };
            default:
                return args.inputs;
        }
    },
}, "project", "stack", false); // Project and stack names, plus dryRun; the names show up in mocked URNs.

随后断言模拟值确实传到了使用它的资源：

describe("Infrastructure", function() {
    let infra: typeof import("../index");

    before(async function() {
        // It's important to import the program _after_ the mocks are defined.
        infra = await import("../index");
    });

    describe("#server", function() {
        it("must use the AMI returned by the getAmi lookup", function(done) {
            pulumi.all([infra.server.urn, infra.server.ami]).apply(([urn, ami]) => {
                if (ami !== "ami-0eb1f3cdeeb8eed2a") {
                    done(new Error(`Unexpected AMI ${ami} on server ${urn}`));
                } else {
                    done();
                }
            });
        });
    });
});

Python

在 call 中匹配 token；其他函数使用默认分支：

class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        return [args.name + "_id", args.inputs]

    def call(self, args: pulumi.runtime.MockCallArgs):
        if args.token == "aws:ec2/getAmi:getAmi":
            return {
                "id": "ami-0eb1f3cdeeb8eed2a",
                "architecture": "x86_64",
            }
        return {}

随后断言模拟值确实传到了使用它的资源：

class TestingWithMocks(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        pulumi.runtime.set_mocks(
            MyMocks(),
            preview=False,
        )
        # Run the program fresh for each test *after* setting the mocks.
        program = runpy.run_path("__main__.py")
        self.server = program["server"]

    @pulumi.runtime.test
    def test_server_uses_looked_up_ami(self):
        def check_ami(args):
            urn, ami = args
            self.assertEqual(
                ami, "ami-0eb1f3cdeeb8eed2a", f"unexpected AMI on server {urn}"
            )

        return pulumi.Output.all(self.server.urn, self.server.ami).apply(check_ami)

Go

在 Call 中匹配 token；其他函数使用默认分支：

type mocks int

func (mocks) NewResource(args pulumi.MockResourceArgs) (string, resource.PropertyMap, error) {
	return args.Name + "_id", args.Inputs, nil
}

func (mocks) Call(args pulumi.MockCallArgs) (resource.PropertyMap, error) {
	if args.Token == "aws:ec2/getAmi:getAmi" {
		return resource.NewPropertyMapFromMap(map[string]interface{}{
			"id":           "ami-0eb1f3cdeeb8eed2a",
			"architecture": "x86_64",
		}), nil
	}
	return args.Args, nil
}

随后断言模拟值确实传到了使用它的资源：

func TestInfrastructure(t *testing.T) {
	err := pulumi.RunErr(func(ctx *pulumi.Context) error {
		infra, err := createInfrastructure(ctx)
		assert.NoError(t, err)

		var wg sync.WaitGroup
		wg.Add(1)

		// The instance uses the AMI returned by the lookup.
		pulumi.All(infra.server.URN(), infra.server.Ami).ApplyT(func(all []interface{}) error {
			urn := all[0].(pulumi.URN)
			ami := all[1].(string)

			assert.Equalf(t, "ami-0eb1f3cdeeb8eed2a", ami, "unexpected AMI on server %v", urn)
			wg.Done()
			return nil
		})

		wg.Wait()
		return nil
	}, pulumi.WithMocks("project", "stack", mocks(0))) // Project and stack names; they show up in mocked URNs.
	assert.NoError(t, err)
}

C#

在 CallAsync 中匹配 token；其他函数使用默认分支：

class Mocks : IMocks
{
    public Task<(string? id, object state)> NewResourceAsync(MockResourceArgs args)
    {
        var outputs = ImmutableDictionary.CreateBuilder<string, object>();
        outputs.AddRange(args.Inputs);

        args.Id ??= $"{args.Name}_id";
        return Task.FromResult<(string? id, object state)>((args.Id, (object)outputs));
    }

    public Task<object> CallAsync(MockCallArgs args)
    {
        if (args.Token == "aws:ec2/getAmi:getAmi")
        {
            return Task.FromResult<object>(new Dictionary<string, object>
            {
                { "id", "ami-0eb1f3cdeeb8eed2a" },
                { "architecture", "x86_64" },
            });
        }

        return Task.FromResult((object)ImmutableDictionary<string, object>.Empty);
    }
}

随后断言模拟值确实传到了使用它的资源：

[TestFixture]
public class AmiTests
{
    [Test]
    public async Task InstanceUsesLookedUpAmi()
    {
        var resources = await Testing.RunAsync<WebserverStack>();

        var stack = resources.OfType<WebserverStack>().First();
        var ami = await stack.Server.Ami.GetValueAsync();

        Assert.That(ami, Is.EqualTo("ami-0eb1f3cdeeb8eed2a"));
    }
}

Java

在 callAsync 中匹配 token；其他函数使用默认分支：

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) {
        if ("aws:ec2/getAmi:getAmi".equals(args.token)) {
            return CompletableFuture.completedFuture(Map.of(
                "id", "ami-0eb1f3cdeeb8eed2a",
                "architecture", "x86_64"
            ));
        }
        return CompletableFuture.completedFuture(Map.of());
    }
}

随后断言模拟值确实传到了使用它的资源：

class AmiTest {
    @AfterEach
    void cleanup() {
        PulumiTest.cleanup();
    }

    @Test
    void instanceUsesLookedUpAmi() {
        var result = PulumiTest
            .withMocks(new MyMocks())
            .withOptions(TestOptions.builder()
                // Project and stack names; they show up in mocked URNs.
                .projectName("project").stackName("stack").preview(false)
                .build())
            .runTest(App::stack);

        var instance = result.resources().stream()
            .filter(resource -> resource instanceof Instance)
            .map(resource -> (Instance) resource)
            .findFirst()
            .orElseThrow(() -> new AssertionError("the program created no EC2 instance"));

        assertEquals("ami-0eb1f3cdeeb8eed2a", PulumiTest.extractValue(instance.ami()));
    }
}

mock 只需提供测试真正读取的字段。省略的字段会返回空值，类似没有模拟的资源输出。原文说明，这五个独立项目在文档网站 CI 中运行；其源码位于 static/programs，目录名为 unit-testing-function-mock-<language>。这里未执行它们。

模拟 StackReference

程序用 StackReference 读取其他 stack 的输出时，也需让 mock 处理它。newResource 收到的类型为 pulumi:pulumi:StackReference，可返回模拟的 outputs。

TypeScript

pulumi.runtime.setMocks({
    newResource: function(args: pulumi.runtime.MockResourceArgs): {id: string, state: any} {
        // Handle StackReference resources
        if (args.type === "pulumi:pulumi:StackReference") {
            return {
                id: args.inputs.name + "_id",
                state: {
                    ...args.inputs,
                    outputs: {
                        // Mock the outputs from the referenced stack
                        vpcId: "vpc-12345678",
                        subnetIds: ["subnet-11111111", "subnet-22222222"],
                        clusterName: "my-cluster",
                    },
                },
            };
        }
        // Handle all other resources
        return {
            id: args.inputs.name + "_id",
            state: args.inputs,
        };
    },
    call: function(args: pulumi.runtime.MockCallArgs) {
        return args.inputs;
    },
});

程序仍按正常方式使用 StackReference：

// Example: Program that reads from a StackReference
const networkStack = new pulumi.StackReference("organization/network/prod");
const vpcId = networkStack.getOutput("vpcId");

// In tests, vpcId will resolve to "vpc-12345678" based on the mock above

Python

import pulumi

class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        # Handle StackReference resources
        if args.typ == "pulumi:pulumi:StackReference":
            return [
                args.name + "_id",
                {
                    **args.inputs,
                    "outputs": {
                        # Mock the outputs from the referenced stack
                        "vpcId": "vpc-12345678",
                        "subnetIds": ["subnet-11111111", "subnet-22222222"],
                        "clusterName": "my-cluster",
                    },
                },
            ]
        # Handle all other resources
        return [args.name + "_id", args.inputs]

    def call(self, args: pulumi.runtime.MockCallArgs):
        return {}

程序仍按正常方式使用 StackReference：

network_stack = pulumi.StackReference("organization/network/prod")
vpc_id = network_stack.get_output("vpcId")

# In tests, vpc_id will resolve to "vpc-12345678" based on the mock above

Go

type mocks int

func (mocks) NewResource(args pulumi.MockResourceArgs) (string, resource.PropertyMap, error) {
	// Handle StackReference resources
	if args.TypeToken == "pulumi:pulumi:StackReference" {
		outputs := resource.NewPropertyMapFromMap(map[string]interface{}{
			"vpcId":       "vpc-12345678",
			"subnetIds":   []interface{}{"subnet-11111111", "subnet-22222222"},
			"clusterName": "my-cluster",
		})
		// Copy inputs and add outputs
		state := args.Inputs.Copy()
		state["outputs"] = resource.NewObjectProperty(outputs)
		return args.Name + "_id", state, nil
	}
	// Handle all other resources
	return args.Name + "_id", args.Inputs, nil
}

func (mocks) Call(args pulumi.MockCallArgs) (resource.PropertyMap, error) {
	return args.Args, nil
}

程序仍按正常方式使用 StackReference：

networkStack, err := pulumi.NewStackReference(ctx, "organization/network/prod", nil)
if err != nil {
    return err
}
vpcId := networkStack.GetStringOutput(pulumi.String("vpcId"))

// In tests, vpcId will resolve to "vpc-12345678" based on the mock above

C#

class Mocks : IMocks
{
    public Task<(string? id, object state)> NewResourceAsync(MockResourceArgs args)
    {
        var outputs = ImmutableDictionary.CreateBuilder<string, object>();
        outputs.AddRange(args.Inputs);

        // Handle StackReference resources
        if (args.Type == "pulumi:pulumi:StackReference")
        {
            outputs.Add("outputs", new Dictionary<string, object>
            {
                // Mock the outputs from the referenced stack
                { "vpcId", "vpc-12345678" },
                { "subnetIds", new[] { "subnet-11111111", "subnet-22222222" } },
                { "clusterName", "my-cluster" },
            });
        }

        args.Id ??= $"{args.Name}_id";
        return Task.FromResult<(string? id, object state)>((args.Id, (object)outputs));
    }

    public Task<object> CallAsync(MockCallArgs args)
    {
        return Task.FromResult((object)ImmutableDictionary<string, object>.Empty);
    }
}

程序仍按正常方式使用 StackReference：

var networkStack = new StackReference("organization/network/prod");
var vpcId = networkStack.GetOutput("vpcId");

// In tests, vpcId will resolve to "vpc-12345678" based on the mock above

Java

import java.util.List;

class MyMocks implements Mocks {
    @Override
    public CompletableFuture<ResourceResult> newResourceAsync(ResourceArgs args) {
        var state = new HashMap<>(args.inputs);
        // Handle StackReference resources
        if ("pulumi:pulumi:StackReference".equals(args.type)) {
            state.put("outputs", Map.of(
                "vpcId", "vpc-12345678",
                "subnetIds", List.of("subnet-11111111", "subnet-22222222"),
                "clusterName", "my-cluster"
            ));
        }
        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());
    }
}

程序仍按正常方式使用 StackReference：

var networkStack = new StackReference("organization/network/prod",
    StackReferenceArgs.builder().build());
var vpcId = networkStack.getOutput(Output.of("vpcId"));

// In tests, vpcId will resolve to "vpc-12345678" based on the mock above

这样测试不要求被引用的 stack 实际存在。修改测试初始化时的返回输出，便可覆盖不同情境。

编写测试

TypeScript

结构与普通 Mocha 测试相同。文件：ec2tests.ts。

import * as pulumi from "@pulumi/pulumi";
import "mocha";

pulumi.runtime.setMocks({
    // ... mocks as shown above
});

describe("Infrastructure", function() {
    let infra: typeof import("./index");

    before(async function() {
        // It's important to import the program _after_ the mocks are defined.
        infra = await import("./index");
    })

    describe("#server", function() {
        // TODO(check 1): Instances have a Name tag.
        // TODO(check 2): Instances must not use an inline userData script.
    });

    describe("#group", function() {
        // TODO(check 3): Instances must not have SSH open to the internet.
    });
});

Python

Pulumi 的 Python 运行时需要 asyncio 事件循环。继承 unittest.IsolatedAsyncioTestCase，为每个测试创建并关闭独立循环。在 setUp 中初始化 mock，然后重新运行 Pulumi 程序。文件：test_ec2.py。

import runpy
import unittest
import pulumi

# ... MyMocks as shown above

class TestingWithMocks(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        pulumi.runtime.set_mocks(
            MyMocks(),
            preview=False, # Sets the flag `dry_run`, which is true at runtime during a preview.
        )
        # Run the program fresh for each test *after* setting the mocks.
        program = runpy.run_path("__main__.py")
        self.group = program["group"]
        self.server = program["server"]

    # TODO(check 1): Instances have a Name tag.
    # TODO(check 2): Instances must not use an inline userData script.
    # TODO(check 3): Instances must not have SSH open to the internet.

Go

结构与普通 Go 测试相同。文件：main_test.go。

package main

import (
	"sync"
	"testing"

	"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ec2"
	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
	"github.com/stretchr/testify/assert"
)

// ... mocks as shown above

func TestInfrastructure(t *testing.T) {
	err := pulumi.RunErr(func(ctx *pulumi.Context) error {
		infra, err := createInfrastructure(ctx)
		assert.NoError(t, err)

		var wg sync.WaitGroup
		wg.Add(3)

		// TODO(check 1): Instances have a Name tag.
		// TODO(check 2): Instances must not use an inline userData script.
		// TODO(check 3): Instances must not have SSH open to the internet.

		wg.Wait()
		return nil
	}, pulumi.WithMocks("project", "stack", mocks(0))) // Project and stack names; they end up in the mocked resources' URNs.
	assert.NoError(t, err)
}

C#

结构与普通 NUnit 测试相同。文件：WebserverStackTests.cs。

using System.Linq;
using System.Threading.Tasks;
using FluentAssertions;
using NUnit.Framework;
using Pulumi.Aws.Ec2;

namespace UnitTesting
{
    [TestFixture]
    public class WebserverStackTests
    {
        // TODO(check 1): Instances have a Name tag.
        // TODO(check 2): Instances must not use an inline userData script.
        // TODO(check 3): Instances must not have SSH open to the internet.
    }
}

Java

结构与普通 JUnit 5 测试类相同。每次测试后调用 PulumiTest.cleanup()，重置 Pulumi 运行时状态。文件：Ec2Tests.java。

package myproject;

import com.pulumi.test.PulumiTest;
import com.pulumi.test.TestOptions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;

class Ec2Tests {
    @AfterEach
    void cleanup() {
        PulumiTest.cleanup();
    }

    // TODO(check 1): Instances have a Name tag.
    // TODO(check 2): Instances must not use an inline userData script.
    // TODO(check 3): Instances must not have SSH open to the internet.
}

检查 Name 标签

先实现第一条规则：取得 EC2 实例对象并检查它的标签属性。

TypeScript

// check 1: Instances have a Name tag.
it("must have a name tag", function(done) {
    pulumi.all([infra.server.urn, infra.server.tags]).apply(([urn, tags]) => {
        if (!tags || !tags["Name"]) {
            done(new Error(`Missing a name tag on server ${urn}`));
        } else {
            done();
        }
    });
});

Python

class TestingWithMocks(unittest.IsolatedAsyncioTestCase):
    # ... setUp as shown above

    # check 1: Instances have a Name tag.
    @pulumi.runtime.test
    def test_server_tags(self):
        def check_tags(args):
            urn, tags = args
            self.assertIsNotNone(tags, f"server {urn} must have tags")
            self.assertIn("Name", tags, f"server {urn} must have a name tag")

        return pulumi.Output.all(self.server.urn, self.server.tags).apply(check_tags)

Go

// check 1: Instances have a Name tag.
pulumi.All(infra.server.URN(), infra.server.Tags).ApplyT(func(all []interface{}) error {
	urn := all[0].(pulumi.URN)
	tags := all[1].(map[string]string)

	assert.Containsf(t, tags, "Name", "missing a Name tag on server %v", urn)
	wg.Done()
	return nil
})

C#

// check 1: Instances have a Name tag.
[Test]
public async Task InstanceHasNameTag()
{
    var resources = await Testing.RunAsync<WebserverStack>();

    var instance = resources.OfType<Instance>().FirstOrDefault();
    instance.Should().NotBeNull("EC2 Instance not found");

    var tags = await instance.Tags.GetValueAsync();
    tags.Should().NotBeNull("Tags are not defined");
    tags.Should().ContainKey("Name");
}

Java

// check 1: Instances have a Name tag.
@Test
void instanceMustHaveNameTag() {
    var result = PulumiTest
        .withMocks(new MyMocks())
        .withOptions(TestOptions.builder()
            // Project and stack names; they end up in the mocked resources' URNs.
            .projectName("project").stackName("stack").preview(false)
            .build())
        .runTest(App::stack);

    var instances = result.resources().stream()
        .filter(r -> r instanceof Instance)
        .map(r -> (Instance) r)
        .toList();

    assertFalse(instances.isEmpty(), "EC2 Instance not found");
    for (var instance : instances) {
        var urn = PulumiTest.extractValue(instance.urn());
        var tags = PulumiTest.extractValue(instance.tags());
        assertNotNull(tags, "Server " + urn + " must have tags");
        assertTrue(tags.containsKey("Name"), "Server " + urn + " must have a Name tag");
    }
}

这与普通测试相似，但需留意三个特点：

没有真实部署，许多云计算出的输出属性是未定义的，除非 mock 显式提供。这一组规则只读取输入属性，因此不受这些输出缺失影响。
Pulumi 资源属性是 Output，许多值异步解析。用 apply 访问底层值，C# 可参看 Testing.cs 中的 GetValueAsync。
测试依赖框架对异步测试的支持，以等待输出解析完成。

解析后，原始输入就是普通值。标签是映射，测试检查其存在且含有 Name 键。同样的方法可检查其他资源属性。

检查 userData

第二条规则要求 userData 属性为空：

TypeScript

// check 2: Instances must not use an inline userData script.
it("must not use userData (use an AMI instead)", function(done) {
    pulumi.all([infra.server.urn, infra.server.userData]).apply(([urn, userData]) => {
        if (userData) {
            done(new Error(`Illegal use of userData on server ${urn}`));
        } else {
            done();
        }
    });
});

Python

class TestingWithMocks(unittest.IsolatedAsyncioTestCase):
    # ... setUp as shown above

    # check 2: Instances must not use an inline userData script.
    @pulumi.runtime.test
    def test_server_userdata(self):
        def check_user_data(args):
            urn, user_data = args
            self.assertFalse(user_data, f"illegal use of user_data on server {urn}")

        return pulumi.Output.all(self.server.urn, self.server.user_data).apply(check_user_data)

Go

// check 2: Instances must not use an inline userData script.
pulumi.All(infra.server.URN(), infra.server.UserData).ApplyT(func(all []interface{}) error {
	urn := all[0].(pulumi.URN)
	userData := all[1].(string)

	assert.Emptyf(t, userData, "illegal use of userData on server %v", urn)
	wg.Done()
	return nil
})

C#

// check 2: Instances must not use an inline userData script.
[Test]
public async Task InstanceMustNotUseInlineUserData()
{
    var resources = await Testing.RunAsync<WebserverStack>();

    var instance = resources.OfType<Instance>().FirstOrDefault();
    instance.Should().NotBeNull("EC2 Instance not found");

    var tags = await instance.UserData.GetValueAsync();
    tags.Should().BeNull();
}

Java

// check 2: Instances must not use an inline userData script.
@Test
void instanceMustNotUseInlineUserData() {
    var result = PulumiTest
        .withMocks(new MyMocks())
        .withOptions(TestOptions.builder()
            // Project and stack names; they end up in the mocked resources' URNs.
            .projectName("project").stackName("stack").preview(false)
            .build())
        .runTest(App::stack);

    var instance = result.resources().stream()
        .filter(r -> r instanceof Instance)
        .map(r -> (Instance) r)
        .findFirst().orElse(null);

    assertNotNull(instance, "EC2 Instance not found");
    var urn = PulumiTest.extractValue(instance.urn());
    var userData = PulumiTest.extractValue(instance.userData());
    assertNull(userData, "Illegal use of userData on server " + urn);
}

检查 SSH 入口

安全组可能有多条入口规则，每条规则也可能含多个 CIDR。第三条检查遍历这些规则及 CIDR：

TypeScript

// check 3: Instances must not have SSH open to the internet.
it("must not open port 22 (SSH) to the internet", function(done) {
    pulumi.all([infra.group.urn, infra.group.ingress]).apply(([ urn, ingress ]) => {
        if (ingress.find(rule =>
            rule.fromPort === 22 && (rule.cidrBlocks || []).find(block => block === "0.0.0.0/0"))) {
                done(new Error(`Illegal SSH port 22 open to the internet (CIDR 0.0.0.0/0) on group ${urn}`));
        } else {
            done();
        }
    });
});

Python

class TestingWithMocks(unittest.IsolatedAsyncioTestCase):
    # ... setUp as shown above

    # check 3: Test if port 22 for ssh is exposed.
    @pulumi.runtime.test
    def test_security_group_rules(self):
        def check_security_group_rules(args):
            urn, ingress = args
            ssh_open = any(
                rule["from_port"] == 22
                and "0.0.0.0/0" in rule["cidr_blocks"]
                for rule in ingress
            )
            self.assertFalse(
                ssh_open,
                f"security group {urn} exposes port 22 to the internet (CIDR 0.0.0.0/0)",
            )

        return pulumi.Output.all(self.group.urn, self.group.ingress).apply(check_security_group_rules)

Go

// check 3: Test if port 22 for ssh is exposed.
pulumi.All(infra.group.URN(), infra.group.Ingress).ApplyT(func(all []interface{}) error {
	urn := all[0].(pulumi.URN)
	ingress := all[1].([]ec2.SecurityGroupIngress)

	for _, i := range ingress {
		openToInternet := false
		for _, b := range i.CidrBlocks {
			if b == "0.0.0.0/0" {
				openToInternet = true
				break
			}
		}

		assert.Falsef(t, i.FromPort == 22 && openToInternet, "illegal SSH port 22 open to the internet (CIDR 0.0.0.0/0) on group %v", urn)
	}

	wg.Done()
	return nil
})

C#

// check 3: Test if port 22 for ssh is exposed.
[Test]
public async Task SecurityGroupMustNotHaveSshPortsOpenToInternet()
{
    var resources = await Testing.RunAsync<WebserverStack>();

    foreach (var securityGroup in resources.OfType<SecurityGroup>())
    {
        var urn = await securityGroup.Urn.GetValueAsync();
        var ingress = await securityGroup.Ingress.GetValueAsync();
        foreach (var rule in ingress)
        {
            (rule.FromPort == 22 && rule.CidrBlocks.Any(b => b == "0.0.0.0/0"))
                .Should().BeFalse($"Illegal SSH port 22 open to the internet (CIDR 0.0.0.0/0) on group {urn}");
        }
    }
}

Java

// check 3: Instances must not have SSH open to the internet.
@Test
void securityGroupMustNotHaveSshOpenToInternet() {
    var result = PulumiTest
        .withMocks(new MyMocks())
        .withOptions(TestOptions.builder()
            // Project and stack names; they end up in the mocked resources' URNs.
            .projectName("project").stackName("stack").preview(false)
            .build())
        .runTest(App::stack);

    for (var resource : result.resources()) {
        if (resource instanceof SecurityGroup group) {
            var urn = PulumiTest.extractValue(group.urn());
            var ingress = PulumiTest.extractValue(group.ingress());
            if (ingress != null) {
                for (var rule : ingress) {
                    var fromPort = PulumiTest.extractValue(rule.fromPort());
                    var cidrBlocks = PulumiTest.extractValue(rule.cidrBlocks());
                    boolean sshOpen = fromPort != null && fromPort == 22
                        && cidrBlocks != null && cidrBlocks.contains("0.0.0.0/0");
                    assertFalse(sshOpen, "Illegal SSH port 22 open to the internet "
                        + "(CIDR 0.0.0.0/0) on group " + urn);
                }
            }
        }
    }
}

三条检查已齐备。注意这些 SSH 示例仅匹配起始端口恰为 22 且 IPv4 CIDR 为 0.0.0.0/0 的情形，未全面覆盖包含 22 的端口区间、所有协议或 IPv6 等开放方式。实际安全策略需要按资源模型扩展断言。

运行测试

在完成项目配置后，可以使用以下命令。本文未运行这些命令，也未部署 AWS 资源。

TypeScript

npx mocha --require tsx ec2tests.ts

Python

python -m unittest

Go

go test

C#

dotnet test

Java

mvn test

原文示范中，初始程序故意违反三条规则，因此三项测试失败。以下是原文的输出示例：

TypeScript

  Infrastructure
    #server
      1) must have a name tag
      2) must not use userData (use an AMI instead)
    #group
      3) must not open port 22 (SSH) to the internet

  0 passing (454ms)
  3 failing

Python

======================================================================
FAIL: test_security_group_rules (test_ec2.TestingWithMocks)
----------------------------------------------------------------------
...
======================================================================
FAIL: test_server_tags (test_ec2.TestingWithMocks)
----------------------------------------------------------------------
...
======================================================================
FAIL: test_server_userdata (test_ec2.TestingWithMocks)
----------------------------------------------------------------------
...
----------------------------------------------------------------------
Ran 3 tests in 0.034s

FAILED (failures=3)

Go

--- FAIL: TestInfrastructure (0.00s)
...
        	Error:      	Should be false
        	Test:       	TestInfrastructure
        	Messages:   	illegal SSH port 22 open to the internet (CIDR 0.0.0.0/0) on group urn:pulumi:stack::project::aws:ec2/securityGroup:SecurityGroup::web-secgrp
...
        	Error:      	Expected nil, but got: (*string)(0xc000217390)
        	Test:       	TestInfrastructure
        	Messages:   	illegal use of userData on server urn:pulumi:stack::project::aws:ec2/instance:Instance::web-server-www
...
        	Error:      	"map[]" does not contain "Name"
        	Test:       	TestInfrastructure
        	Messages:   	missing a Name tag on server urn:pulumi:stack::project::aws:ec2/instance:Instance::web-server-www
FAIL	testing-unit-go	0.501s

C#

X InstanceHasNameTag [387ms]
  Error Message:
   Expected tags not to be <null> because Tags are not defined.

X InstanceMustNotUseInlineUserData [17ms]
  Error Message:
   Expected tags to be <null>, but found "#!/bin/bash echo "Hello, World!" > index.html nohup python3 -m http.server 80 &".

X SecurityGroupMustNotHaveSshPortsOpenToInternet [11ms]
  Error Message:
   Expected boolean to be false because Illegal SSH port 22 open to the internet (CIDR 0.0.0.0/0) on group urn:pulumi:stack::project::pulumi:pulumi:Stack$aws:ec2/securityGroup:SecurityGroup::web-secgrp, but found True.

Test Run Failed.
Total tests: 3
     Failed: 3

Java

[ERROR] Tests run: 3, Failures: 3, Errors: 0, Skipped: 0
[ERROR] Ec2Tests.instanceMustHaveNameTag -- AssertionFailedError: Server ... must have a Name tag
[ERROR] Ec2Tests.instanceMustNotUseInlineUserData -- AssertionFailedError: Illegal use of userData on server ...
[ERROR] Ec2Tests.securityGroupMustNotHaveSshOpenToInternet -- AssertionFailedError: Illegal SSH port 22 open to the internet (CIDR 0.0.0.0/0) on group ...
[ERROR] BUILD FAILURE

现在把程序修改为符合规则的版本：

TypeScript

文件：index.ts。

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"] },
    ],
});

// Look up the latest Amazon Linux 2 AMI.
const ami = aws.ec2.getAmiOutput({
    owners: ["amazon"],
    mostRecent: true,
    filters: [{ name: "name", values: ["amzn2-ami-hvm-*-x86_64-gp2"] }],
});

export const server = new aws.ec2.Instance("web-server-www", {
    instanceType: "t2.micro",
    securityGroups: [ group.name ], // reference the group object above
    ami: ami.id,
    tags: { Name: "webserver" },    // name tag
});

Python

文件：__main__.py。

import pulumi
from pulumi_aws import ec2

group = ec2.SecurityGroup('web-secgrp', ingress=[
    { "protocol": "tcp", "from_port": 80, "to_port": 80, "cidr_blocks": ["0.0.0.0/0"] },
])

# Look up the latest Amazon Linux 2 AMI.
ami = ec2.get_ami_output(
    owners=["amazon"],
    most_recent=True,
    filters=[{"name": "name", "values": ["amzn2-ami-hvm-*-x86_64-gp2"]}])

server = ec2.Instance("web-server-www",
    instance_type="t2.micro",
    security_groups=[ group.name ], # reference the group object above
    tags={'Name': 'webserver'},     # name tag
    ami=ami.id)

Go

文件：main.go。

package main

import (
	"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ec2"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

type infrastructure struct {
	group  *ec2.SecurityGroup
	server *ec2.Instance
}

func createInfrastructure(ctx *pulumi.Context) (*infrastructure, 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
	}

	// Look up the latest Amazon Linux 2 AMI.
	ami, err := ec2.LookupAmi(ctx, &ec2.LookupAmiArgs{
		Owners:     []string{"amazon"},
		MostRecent: pulumi.BoolRef(true),
		Filters: []ec2.GetAmiFilter{
			{
				Name:   "name",
				Values: []string{"amzn2-ami-hvm-*-x86_64-gp2"},
			},
		},
	})
	if err != nil {
		return nil, err
	}

	server, err := ec2.NewInstance(ctx, "web-server-www", &ec2.InstanceArgs{
		InstanceType:   pulumi.String("t2.micro"),
		SecurityGroups: pulumi.StringArray{group.Name}, // reference the group object above
		Ami:            pulumi.String(ami.Id),
		Tags:           pulumi.StringMap{"Name": pulumi.String("webserver")},
	})
	if err != nil {
		return nil, err
	}

	return &infrastructure{
		group:  group,
		server: server,
	}, nil
}

C#

文件：WebserverStack.cs。

using Pulumi;
using Pulumi.Aws.Ec2;
using Pulumi.Aws.Ec2.Inputs;

public class WebserverStack : Stack
{
    public WebserverStack()
    {
        var group = new SecurityGroup("web-secgrp", new SecurityGroupArgs
        {
            Ingress =
            {
                new SecurityGroupIngressArgs { Protocol = "tcp", FromPort = 80, ToPort = 80, CidrBlocks = { "0.0.0.0/0" } }
            }
        });

        // Look up the latest Amazon Linux 2 AMI.
        var ami = GetAmi.Invoke(new GetAmiInvokeArgs
        {
            Owners = { "amazon" },
            MostRecent = true,
            Filters =
            {
                new GetAmiFilterInputArgs { Name = "name", Values = { "amzn2-ami-hvm-*-x86_64-gp2" } }
            }
        });

        var server = new Instance("web-server-www", new InstanceArgs
        {
            InstanceType = "t2.micro",
            SecurityGroups = { group.Name }, // reference the group object above
            Ami = ami.Apply(ami => ami.Id),
            Tags = { { "Name", "webserver" } }  // name tag
        });
    }
}

Java

文件：App.java。

package myproject;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.aws.ec2.Ec2Functions;
import com.pulumi.aws.ec2.Instance;
import com.pulumi.aws.ec2.InstanceArgs;
import com.pulumi.aws.ec2.SecurityGroup;
import com.pulumi.aws.ec2.SecurityGroupArgs;
import com.pulumi.aws.ec2.inputs.GetAmiArgs;
import com.pulumi.aws.ec2.inputs.GetAmiFilterArgs;
import com.pulumi.aws.ec2.inputs.SecurityGroupIngressArgs;
import com.pulumi.aws.ec2.outputs.GetAmiResult;
import java.util.Map;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        var group = new SecurityGroup("web-secgrp", SecurityGroupArgs.builder()
            .ingress(
                SecurityGroupIngressArgs.builder()
                    .protocol("tcp").fromPort(80).toPort(80).cidrBlocks("0.0.0.0/0")
                    .build())
            .build());

        // Look up the latest Amazon Linux 2 AMI.
        var ami = Ec2Functions.getAmi(GetAmiArgs.builder()
            .owners("amazon")
            .mostRecent(true)
            .filters(GetAmiFilterArgs.builder()
                .name("name")
                .values("amzn2-ami-hvm-*-x86_64-gp2")
                .build())
            .build());

        var server = new Instance("web-server-www", InstanceArgs.builder()
            .instanceType("t2.micro")
            .securityGroups(group.name())  // reference the group object above
            .ami(ami.applyValue(GetAmiResult::id))
            .tags(Map.of("Name", "webserver")) // name tag
            .build());
    }
}

再次运行测试，原文展示以下通过结果：

TypeScript

Infrastructure
    #server
      ✓ must have a name tag
      ✓ must not use userData (use an AMI instead)
    #group
      ✓ must not open port 22 (SSH) to the internet

  3 passing (454ms)

Python

----------------------------------------------------------------------
Ran 3 tests in 0.022s

OK

Go

PASS
ok  	testing-unit-go	0.704s

C#

Test Run Successful.
Total tests: 3
     Passed: 3

Java

[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0
[INFO] BUILD SUCCESS

这组输出说明的是官方示例的预期结果，不构成本机运行验证。

限制

用于单元测试的 mock server 没有实现完整 Pulumi 引擎。因此，依赖引擎部署协调的功能可能不会执行。

生命周期 hook 与资源 transform

程序可以向 mock server 注册生命周期 hook 和资源 transform，但 mock 测试并不实际执行它们。若要完全支持这些功能，各语言 SDK 都需要重新实现引擎的相当一部分逻辑；为了保持测试快速、确定且不依赖外部系统，mock 选择了这个边界。

程序依赖 hook 或 transform 时，可以：

把内部业务逻辑提取为独立函数，单独进行单元测试。
让 mock 返回预期执行后的资源状态，测试消费该状态的逻辑。
通过在测试环境中部署真实资源的集成测试，验证完整 hook 或 transform 行为。

例如，某个 transform 给全部资源增加默认标签，mock 的 newResource 可以直接返回已包含这些标签的状态，以模拟结果，但这并没有验证 transform 本身已经执行。

完整示例

原文指出，examples 仓库尚无这篇指南对应的 Java 完整示例；欢迎向 pulumi/examples贡献代码。前述函数 mock 的 Java 独立项目是另一组示例。

继续阅读

集成测试会部署真实资源并作端到端检查，适合 hook、transform 及 mock 未实现的功能。
输入与输出解释资源属性为何异步解析，以及 apply 如何取得值。
Pulumi Policies可把类似本教程的规则应用到组织的各个 stack，而非仅检查单个程序。

静态核对范围：保留五种语言的示例、命令及示例输出，解析并核对官方引用的代码片段。含省略号、TODO 或缺少导入的片段需与项目上下文结合；未执行测试套件、安装依赖或访问云账户。

文章版权归作者所有，未经允许请勿转载。

THE END

技术译文