Unit Testing Infrastructure
Posted on
We’re pleased to announce that unit testing with Node.js, Python, .NET, and Go is supported in recent releases. You can test resources before deploying your infrastructure using familiar tools and test frameworks. Check your resource configuration and responses without the wait of deploying them and speed up infrastructure development and production deployments.
Pulumi brings testing to Infrastructure as Code. Testing your infrastructure code will provide early bug notification, cleaner and more extensible code, and simplify refactoring. We implemented testing with these goals in mind.
- Run code without the engine
- Write tests using existing test tools and frameworks
- Mock any dependencies
Here are examples in Python and Go to get you started.
Python
First we’ll set up an AWS EC2 instance as a web server to test.
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 python -m SimpleHTTPServer 80 &'
server = ec2.Instance('web-server-www;',
instance_type="t2.micro",
security_groups=[ group.name ], # reference the group object above
user_data=user_data, # start a simple web server
ami="ami-c55673a0") # AMI for us-east-2 (Ohio)
pulumi.export('group', group)
pulumi.export('server', server)
pulumi.export('publicIp', server.public_ip)
pulumi.export('publicHostName', server.public_dns)
Using Python’s unittest framework we can test if the test service conforms to a configuration that restricts an exposed port 22 and use of user_data and enforces the use of tags and membership in a security group.
import unittest
import pulumi
# Create the mock.
class MyMocks(pulumi.runtime.Mocks):
def call(self, token, args, provider):
return {}
def new_resource(self, type_, name, inputs, provider, id_):
if type_ == 'aws:ec2/securityGroup:SecurityGroup':
state = {
'arn': 'arn:aws:ec2:us-west-2:123456789012:security-group/sg-12345678',
'name': inputs['name'] if 'name' in inputs else name + '-sg',
}
return ['sg-12345678', dict(inputs, **state)]
elif type_ == 'aws:ec2/instance:Instance':
state = {
'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',
}
return ['i-1234567890abcdef0', dict(inputs, **state)]
else:
return ['', {}]
pulumi.runtime.set_mocks(MyMocks())
# Import the EC2 web server.
import infra
class InfraTests(unittest.TestCase):
# Test if the service has tags and 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, 'server {urn} must have a name tag')
return pulumi.Output.all(infra.server.urn, infra.server.tags).apply(check_tags)
# Test if the instance is configured with user_data.
@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(infra.server.urn, infra.server.user_data).apply(check_user_data)
# Test if service is a member of a security group.
@pulumi.runtime.test
def test_server_security_groups(self):
def check_security_groups(args):
urn, security_groups = args
self.assertIsNotNone(security_groups, f'server {urn} does not specify security_groups')
self.assertGreater(len(security_groups), 0, f'server {urn} does not specify security_groups')
return pulumi.Output.all(infra.server.urn, infra.server.security_groups).apply(check_security_groups)
# 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 any([block == "0.0.0.0/0" for block 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 the results of the unit tests.
return pulumi.Output.all(infra.group.urn, infra.group.ingress).apply(check_security_group_rules)
Go
This is the same example, but written in Go using the stretchr framework for unit tests.
package main
import (
"github.com/pulumi/pulumi-aws/sdk/v2/go/aws/ec2"
"github.com/pulumi/pulumi/sdk/v2/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 python -m SimpleHTTPServer 80 &`
server, err := ec2.NewInstance(ctx, "web-server-www", &ec2.InstanceArgs{
InstanceType: pulumi.String("t2-micro"),
SecurityGroups: pulumi.StringArray{group.ID()},
Ami: pulumi.String("ami-c55673a0"),
UserData: pulumi.String(userData),
})
if err != nil {
return nil, err
}
return &infrastructure{
group: group,
server: server,
}, nil
}
func main() {
pulumi.Run(func(ctx *pulumi.Context) error {
infra, err := createInfrastructure(ctx)
if err != nil {
return err
}
ctx.Export("group", infra.group.ID())
ctx.Export("server", infra.server.ID())
ctx.Export("publicIp", infra.server.PublicIp)
ctx.Export("publicHostName", infra.server.PublicDns)
return nil
})
}
We implement the same tests used in the Python example.
package main
import (
"fmt"
"sync"
"testing"
"github.com/pulumi/pulumi-aws/sdk/v2/go/aws/ec2"
"github.com/pulumi/pulumi/sdk/v2/go/common/resource"
"github.com/pulumi/pulumi/sdk/v2/go/pulumi"
"github.com/stretchr/testify/assert"
)
type mocks int
// Create the mock.
func (mocks) NewResource(typeToken, name string, inputs resource.PropertyMap, provider, id string) (string, resource.PropertyMap, error) {
switch typeToken {
case "aws:ec2/securityGroup:SecurityGroup":
if _, ok := inputs["name"]; !ok {
inputs["name"] = resource.NewStringProperty(name + "-sg")
}
inputs["arn"] = resource.NewStringProperty("arn:aws:ec2:us-west-2:123456789012:security-group/sg-12345678")
return "sg-12345678", inputs, nil
case "aws:ec2/instance:Instance":
inputs["arn"] = resource.NewStringProperty("arn:aws:ec2:us-west-2:123456789012:instance/i-1234567890abcdef0")
inputs["instanceState"] = resource.NewStringProperty("running")
inputs["primaryNetworkInterfaceId"] = resource.NewStringProperty("eni-12345678")
inputs["privateDns"] = resource.NewStringProperty("ip-10-0-1-17.ec2.internal")
inputs["publicDns"] = resource.NewStringProperty("ec2-203-0-113-12.compute-1.amazonaws.com")
inputs["publicIp"] = resource.NewStringProperty("203.0.113.12")
return "i-1234567890abcdef0", inputs, nil
default:
return "", inputs, fmt.Errorf("unexpected resource type %v", typeToken)
}
}
func (mocks) Call(token string, args resource.PropertyMap, provider string) (resource.PropertyMap, error) {
return args, fmt.Errorf("unexpected function call %v", token)
}
// Applying unit tests.
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(4)
// Test if the service has tags and 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]interface{})
assert.Containsf(t, tags, "name", "missing a name tag on server %v", urn)
wg.Done()
return nil
})
// Test if the instance is configured with user_data.
pulumi.All(infra.server.URN(), infra.server.UserData).ApplyT(func(all []interface{}) error {
urn := all[0].(pulumi.URN)
userData := all[1].(*string)
assert.Nilf(t, userData, "illegal use of userData on server %v", urn)
wg.Done()
return nil
})
// Test if service is a member of a security group.
pulumi.All(infra.server.URN(), infra.server.SecurityGroups).ApplyT(func(all []interface{}) error {
urn := all[0].(pulumi.URN)
securityGroups := all[1].([]string)
assert.Greaterf(t, len(securityGroups), 0, "illegal security group spec on server %v", urn)
wg.Done()
return nil
})
// 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
})
wg.Wait()
return nil
}, pulumi.WithMocks("project", "stack", mocks(0)))
assert.NoError(t, err)
}
Try it out today
We’ve released these features early so that you can try them out before the 2.0 release. To get you started, we’ve added full working examples on Github. With the 2.0 release we’ll have more documentation, examples, and blog posts. In the meantime, we would appreciate feedback on our Community Slack channel.
