RM Mock examples

__main__.py

import infra

ec2tests.js

let assert = require("assert");
let mocha = require("mocha");
let pulumi = require("@pulumi/pulumi");

pulumi.runtime.setMocks({
    newResource: function(type, name, inputs) {
        switch (type) {
            case "aws:ec2/securityGroup:SecurityGroup":
                return {
                    id: "sg-12345678",
                    state: {
                        ...inputs,

                        arn: "arn:aws:ec2:us-west-2:123456789012:security-group/sg-12345678",
                        name: inputs.name || name + "-sg",
                    },
                };
            case "aws:ec2/instance:Instance":
                return {
                    id: "i-1234567890abcdef0",
                    state: {
                        ...inputs,

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

let infra = require("./index");

describe("Infrastructure", function() {
    let server = infra.server;
    describe("#server", function() {
        // check 1: Instances have a Name tag.
        it("must have a name tag", function(done) {
            pulumi.all([server.urn, server.tags]).apply(([urn, tags]) => {
                if (!tags || !tags["Name"]) {
                    done(new Error(`Missing a name tag on server ${urn}`));
                } else {
                    done();
                }
            });
        });
        
        // check 2: Instances must not use an inline userData script.
        it("must not use userData (use an AMI instead)", function(done) {
            pulumi.all([server.urn, server.userData]).apply(([urn, userData]) => {
                if (userData) {
                    done(new Error(`Illegal use of userData on server ${urn}`));
                } else {
                    done();
                }
            });
        });

        // check 3: Instances must name at least one security group.
        it("must name a security group", function(done) {
            pulumi.all([server.urn, server.securityGroups]).apply(([urn, securityGroups]) => {
                if (!securityGroups || securityGroups.length === 0 || typeof securityGroups[0] !== "string") {
                    done(new Error(`illegal security group spec on server ${urn}`));
                } else {
                    done();
                }
            });
        });
    });

    let group = infra.group;
    describe("#group", function() {
        // check 4: Instances must not have SSH open to the Internet.
        it("must not open port 22 (SSH) to the Internet", function(done) {
            pulumi.all([ group.urn, 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();
                }
            });
        });
    });
});

ec2tests.py

import unittest
import pulumi

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 infra

class InfraTests(unittest.TestCase):
    @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)

    @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)

    @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)

    @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 pulumi.Output.all(infra.group.urn, infra.group.ingress).apply(check_security_group_rules)

index.js

let aws = require("@pulumi/aws");

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

let userData = `#!/bin/bash echo "Hello, World!" > index.html nohup python -m SimpleHTTPServer 80 &`;

let server = new aws.ec2.Instance("web-server-www", {
    instanceType: "t2.micro",
    securityGroups: [ group.name ], // reference the group object above
    ami: "ami-c55673a0",            // AMI for us-east-2 (Ohio),
    userData: userData,             // start a simple web server
});

exports.group = group;
exports.server = server;
exports.publicIp = server.publicIp;
exports.publicHostName = server.publicDns;

infra.py

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)