https://pulumi.com logo
Title
n

numerous-train-50906

03/08/2023, 4:58 PM
Hi! I'm new to Pulumi. I'm trying to unit test some Python code to provision an EKS cluster using
pulumi/pulumi-eks
. I'd like to be able to test the input values to the Cluster object. However, apart from version, all other input properties are None in the test below- _test_infra.py_
import unittest
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):
        return {}

pulumi.runtime.set_mocks(MyMocks())

import infra

class TestingWithMocks(unittest.TestCase):
    @pulumi.runtime.test
    def test_kubernetes_version(self):
        def check_version(args):
            eks_version, version = args
            self.assertEqual(eks_version, version, 'Cluster must be on Kubernetes 1.24')
        return pulumi.Output.all(infra.eks_cluster.version, "1.24").apply(check_version) 

    @pulumi.runtime.test
    def test_node_instance_type(self):
        def check_node_type(args):
            eks_c, expected_node_type = args
            self.assertEqual(eks_c, expected_node_type, 'Cluster must provision `t3.medium` instance types')

        return pulumi.Output.all(infra.eks_cluster.instance_type, "t3.medium").apply(check_node_type)
However the second test fails when run-
self.assertEqual(eks_c, expected_node_type, 'Cluster must provision `t3.medium` instance types')
AssertionError: None != 't3.medium' : Cluster must provision `t3.medium` instance types
infra.py
import pulumi
import pulumi_awsx as awsx
import pulumi_eks as eks

# Get some values from the Pulumi configuration (or use defaults)
config = pulumi.Config()
min_cluster_size = config.get_float("minClusterSize", 3)
max_cluster_size = config.get_float("maxClusterSize", 6)
desired_cluster_size = config.get_float("desiredClusterSize", 3)
eks_node_instance_type = config.get("eksNodeInstanceType", "t3.medium")
vpc_network_cidr = config.get("vpcNetworkCidr", "10.0.0.0/16")
eks_version = config.get("eksVersion", "1.24")
# Create a VPC for the EKS cluster
eks_vpc = awsx.ec2.Vpc("pulumi-eks-vpc",
    enable_dns_hostnames=True,
    cidr_block=vpc_network_cidr)

# Create the EKS cluster
eks_cluster = eks.Cluster("pulumi-eks-cluster",
    # Put the cluster in the new VPC created earlier
    vpc_id=eks_vpc.vpc_id,
    # Public subnets will be used for load balancers
    public_subnet_ids=eks_vpc.public_subnet_ids,
    # Private subnets will be used for cluster nodes
    private_subnet_ids=eks_vpc.private_subnet_ids,
    # Change configuration values to change any of the following settings
    instance_type="t3.medium",
    desired_capacity=desired_cluster_size,
    min_size=min_cluster_size,
    max_size=max_cluster_size,
    # Do not give worker nodes a public IP address
    node_associate_public_ip_address=False,
    version=eks_version,
    # Uncomment the next two lines for private cluster (VPN access required)
    # endpoint_private_access=true,
    # endpoint_public_access=false
    )

# Export values to use elsewhere
pulumi.export("kubeconfig", eks_cluster.kubeconfig)
pulumi.export("vpcId", eks_vpc.vpc_id)
I've read the docs around Input/Output, but frankly have not fully understood when the outputs are available for access in unit tests.