Add Integration test in python example with docs
ADD
testing-integration-py/.gitignore
+*.pyc
+venv/

ADD
testing-integration-py/Pulumi.yaml
+name: s3-py-it
+runtime:
+ name: python
+ options:
+ virtualenv: venv
+description: A Integration Testing of s3 use case

ADD
testing-integration-py/README.md
+
+# Integration Testing of Pulumi programs in Python
+
+This integration test is using Pulumi [Automation API](https://www.pulumi.com/blog/automation-api/) and [Python Unittest](https://docs.python.org/3/library/unittest.html) to simulate integration test in Pulumi without native python integration library.
+
+| UnitTest | Integration Test |Target|
+|--|--|--|
+| YES | YES|Input|
+| No | YES|Output|
+
+## Test Case
+To create a s3 bucket and upload a file
+
+ 1. Create the stack with a bucket
+ 3. Verify S3 bucket name and region in Output
+ 4. Using client to check the S3 name again
+ 5. Using client to upload a file
+ 6. Deleting that file
+ 7. Delete the stack with the bucket
+
+
+## Precondition
+
+You need to create the credential file yourself. By default, its location is at ~/.aws/credentials. Your access key needs the correct permissions for S3 bucket and object creation/deletion.
+```
+[default]
+aws_access_key_id = YOUR_ACCESS_KEY
+aws_secret_access_key = YOUR_SECRET_KEY
+```
+
+## Running the tests
+```
+pip install -r requirements.txt
+python -m unittest test_s3_it.py
+```
+
+## Test Life Cycle
+ - Create a stack and export the desired outputs.
+ - Validate any output values you defined in advance.
+ - In the end, don't forget to destroy the stack.
+```
+from pulumi.x import automation as auto
+
+class TestS3(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ ....
+ cls.stack = auto.create_or_select_stack(stack_name=cls.STACK_NAME, work_dir=cls.WORK_DIR)
+ cls.stack.up(output=print)
+ cls.outputs = cls.stack.outputs()
+ ...
+
+ @classmethod
+ def tearDownClass(cls) -> None:
+ cls.stack.destroy(on_output=print)
+ cls.stack.workspace.remove_stack(cls.STACK_NAME)
+
+ def test_s3_output_case(self):
+ ...
+ bucket_region = self.outputs.get(OUTPUT_KEY_REGION)
+ self.assertEqual(self.REGION_NAME, bucket_region.value)
+ ...
+
+```
+## Further steps
+
+Learn more about testing Pulumi programs and Automation API:
+
+ - [Automation API](https://www.pulumi.com/blog/automation-api/)
+ - [Pulumi Test](https://www.pulumi.com/docs/guides/testing/)

ADD
testing-integration-py/__init__.py
testing-integration-py/__init__.py
ADD
testing-integration-py/__main__.py
+"""An AWS Python Pulumi program"""
+
+from resource_s3 import create_s3_bucket
+
+bucket = create_s3_bucket()

ADD
testing-integration-py/requirements.txt
+pulumi>=2.0.0,<3.0.0
+pulumi-aws>=3.2.0,<4.0.0
+boto3==1.16.58
\ No newline at end of file

ADD
testing-integration-py/resource_s3.py
+import pulumi
+from pulumi_aws import s3
+
+BUCKET_NAME = 'pulumi-it-bucket'
+OUTPUT_KEY_BUCKET_NAME = 'bucket_name'
+OUTPUT_KEY_REGION = 'region'
+
+
+def create_s3_bucket():
+ # Create an AWS resource (S3 Bucket)
+ bucket = s3.Bucket(BUCKET_NAME)
+
+ # Export the value of the bucket
+ pulumi.export(OUTPUT_KEY_BUCKET_NAME, bucket.bucket)
+ pulumi.export(OUTPUT_KEY_REGION, bucket.region)
+
+ return bucket

ADD
testing-integration-py/test_s3_it.py
+import os
+import unittest
+
+import boto3
+from pulumi.x import automation as auto
+
+from resource_s3 import BUCKET_NAME, OUTPUT_KEY_BUCKET_NAME, OUTPUT_KEY_REGION
+
+
+class TestS3(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.STACK_NAME = 'staging'
+ cls.REGION_NAME = 'eu-north-1'
+ cls.WORK_DIR = os.path.join(os.path.dirname(__file__))
+ cls.FILE_NAME = 'bucket.txt'
+
+ cls.stack = auto.create_or_select_stack(stack_name=cls.STACK_NAME, work_dir=cls.WORK_DIR)
+ cls.stack.set_config("aws:region", auto.ConfigValue(value=cls.REGION_NAME))
+ cls.stack.up(on_output=print)
+ cls.outputs = cls.stack.outputs()
+ cls.s3 = boto3.resource('s3')
+
+ @classmethod
+ def tearDownClass(cls) -> None:
+ cls.stack.destroy(on_output=print)
+ cls.stack.workspace.remove_stack(cls.STACK_NAME)
+
+ def test_s3_output_region(self):
+ bucket_region = self.outputs.get(OUTPUT_KEY_REGION)
+ self.assertEqual(self.REGION_NAME, bucket_region.value)
+
+ def test_s3_output_name(self):
+ bucket_name_output = self.outputs.get(OUTPUT_KEY_BUCKET_NAME)
+ self.assertIn(BUCKET_NAME, bucket_name_output.value)
+
+ def test_s3_exist(self):
+ buckets = self.s3.buckets.all()
+ bucket_names = []
+ for bucket in buckets:
+ bucket_names.append(bucket.name)
+ output_bucket_name = self.outputs.get(OUTPUT_KEY_BUCKET_NAME).value
+ self.assertIn(output_bucket_name, bucket_names)
+
+ def test_s3_create_permission(self):
+ output_bucket_name = self.outputs.get(OUTPUT_KEY_BUCKET_NAME).value
+ created_response = self.s3.Bucket(output_bucket_name).put_object(Key=self.FILE_NAME, Body='Hi')
+ self.assertEqual(created_response.key, self.FILE_NAME)
+
+ def test_s3_delete_permission(self):
+ output_bucket_name = self.outputs.get(OUTPUT_KEY_BUCKET_NAME).value
+ delete_response = self.s3.meta.client.delete_object(Bucket=output_bucket_name, Key=self.FILE_NAME)
+ self.assertIsNotNone(delete_response)
+
+
+if __name__ == '__main__':
+ unittest.main()

Add Integration test in python example with docs (#881)

Co-authored-by: Komal <komalsali@gmail.com>
Co-authored-by: Qunfei Wu <qunfei.wu@outlook.com>