Anyone have experience writing python tests for pu...
# python
w
Anyone have experience writing python tests for pulumi? Read @big-piano-35669ā€™s blog post and trying to kick the tires on this. https://blog.pulumi.com/testing-your-infrastructure-as-code-with-pulumi Is there an example of the tests from that blog but written in Python?
I am looking for the equivalent of this in the python lib: https://github.com/pulumi/pulumi/blob/3ddaf69e185be84b1f1cd28b0fdaefbf700557bd/sdk/nodejs/runtime/settings.ts#L212-L219 but I'm not seeing it. Is testing not supported in Python? cc @gentle-diamond-70147
l
i was asking the same several weeks ago (how to test python pulumi stuff). no answer
but in my mind its quite important
g
I'm paraphrasing from an email where Joe answered this separately, but... This is doable, but takes a few more settings to work within Python. We'll open an issue to make this more "first class
From Joe:
Copy code
I did a slight refactoring of the AWS WebServer example at <https://github.com/pulumi/examples/tree/master/aws-py-webserver>, and wrote a little test below that checks to ensure we don't have unrestricted CIDR blocks on security groups. A few things to note: (1) the project structure needs to change a little rather than declaring everything in __main__.py to facilitate importing, (2) we call a few pulumi.runtime.settings._set_x(..) helpers in our test before allocating anything, and (3) slightly awkwardly, I had to do the import inside of the test case itself, to avoid promises being allocated on a separate event loop than the test is running on.
šŸ‘ 1
Copy code
test.py
import asyncio
import unittest

# Import the Pulumi SDK and turn on test mode *before* allocating anything.
import pulumi
pulumi.runtime.settings._set_test_mode_enabled(True)
pulumi.runtime.settings._set_project('webserver')
pulumi.runtime.settings._set_stack('test')

## Tests come next:

def async_test(coro):
    def wrapper(*args, **kwargs):
        loop = asyncio.new_event_loop()
        loop.run_until_complete(coro(*args, **kwargs))
        loop.close()
    return wrapper

class InstanceTests(unittest.TestCase):
    @async_test
    async def test_no_internet_access(self):
        # Now import the infrastructure module which will allocate our resource objects.
        from infra import group, server

        # Rendezvous with the resource's resulting ingress rules.
        ingress_fut = asyncio.Future()
        group.ingress.apply(lambda ig: ingress_fut.set_result(ig))
        ingress = await ingress_fut
        # Now loop through them and reject any that are open to the Internet (0.0.0.0/0).
        for rule in ingress:
            for cidr in rule['cidr_blocks']:
                self.assertNotEqual('0.0.0.0/0', cidr)
šŸ‘ 1
b
Apologies for missing this folks, we will be sure to blog this in the coming week or two.
w
Hi Joe/Cameron, thanks for this! I'm able to use the pulumi.runtime.settings helpers to run an integration test. When I try to unpack the
id
output for a resource, e.g.
bucket.id
using the same async + apply strategy you've shown here, it comes back as
None
. Is that expected? In my test, I am checking that an S3 bucket name is what I set the name to. So in
Copy code
bucket = s3.Bucket('my-bucket')
I am trying to assert that
await id_future == 'my-bucket'
but the id is
None
. Am I missing something? cc @white-balloon-205 @bitter-oil-46081
b
I believe that the issue is that when we are running in testing mode, we only reflect back the set of input properties (i.e. what you would pass in the args dictionary to
s3.Bucket
. Since we can not actually run the resource providers (since there's no
pulumi
invocation in the mix) we can't know what
id
it is going to set in general. Since there's no
id
input to the resource (normally the provider computes it based on the other inputs) it doesn't get a real output.