Channels
announcements
automation-api
aws
azure
blog-posts
built-with-pulumi
cloudengineering
cloudengineering-support
content-share
contribex
contribute
docs
dotnet
finops
general
getting-started
gitlab
golang
google-cloud
hackathon-03-19-2020
hacktoberfest
install
java
jobs
kubernetes
learn-pulumi-events
linen
localstack
multi-language-hackathon
office-hours
oracle-cloud-infrastructure
plugin-framework
pulumi-ai
pulumi-cdk
pulumi-cloud
pulumi-crosscode
pulumi-deployments
pulumi-kubernetes-operator
pulumiverse
python
registry
status
testingtesting123
testingtesting321
typescript
welcome
workshops
yaml
Title
n
numerous-train-50906
03/08/2023, 4:58 PMHi!
I'm new to Pulumi. I'm trying to unit test some Python code to provision an EKS cluster using
pulumi/pulumi-eksimport 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)self.assertEqual(eks_c, expected_node_type, 'Cluster must provision `t3.medium` instance types')
AssertionError: None != 't3.medium' : Cluster must provision `t3.medium` instance typesinfra.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.