pulumi new aws-python
  

  
pulumi config set aws:profile 
  

  
"""An AWS Python Pulumi program"""

import pulumi
from pulumi_aws import s3,lambda_,iam,dynamodb

# Create bucket to store transcribed audio file
bucket_tts = s3.Bucket(
    resource_name='slsguru-tts-bucket',
    bucket="",
    )

# Create dynamoDB table containing text entries
ddb = dynamodb.Table(
    resource_name="ttsslsguru",
    name="ttsslsguru", # this disables auto-naming
    billing_mode="PAY_PER_REQUEST",
    attributes=[
                {
                    "name": "Id",
                    "type": "N",
                },
            ],
    hash_key="Id",
    stream_enabled=True,
    stream_view_type='NEW_IMAGE',
)

# Create Lambda IAM lambda_role
lambda_role = iam.Role(
    resource_name='lambda-tts-iam-role',
    assume_role_policy="""{
        "Version": "2012-10-17",
        "Statement": [
            {
                "Action": "sts:AssumeRole",
                "Principal": {
                    "Service": "lambda.amazonaws.com"
                },
                "Effect": "Allow",
                "Sid": ""
            }
        ]
    }"""
)

# Attach policy to the role
lambda_role_policy = iam.RolePolicy(
    resource_name='lambda-tts-iam-policy',
    role=lambda_role.id,
    policy="""{
        "Version": "2012-10-17",
        "Statement": [
            {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "s3:PutObject",
                "polly:SynthesizeSpeech",
                "dynamodb:*"
            ],
            "Resource": "*"
        }
        ]
    }"""
)

# Create Lambda function and attach the created role 
lambda_fn = lambda_.Function(
    resource_name='slsguru-tts',
    role=lambda_role.arn,
    runtime="python3.6",
    handler="lambdatts.lambda_handler",
    code=pulumi.AssetArchive({
        '.': pulumi.FileArchive('.')
    }),
    timeout=30,
    memory_size=128,
)

# Allow lambda to get events
lambda_event = lambda_.EventSourceMapping(
    resource_name="ddb_stream_event",
    event_source_arn=ddb.stream_arn,
    function_name=lambda_fn.arn,
    starting_position="LATEST",
    maximum_batching_window_in_seconds=2
)

# Export created assets (buckets, lambda function)
pulumi.export('Text_bucket', bucket_tts.id)
pulumi.export('table', ddb.arn)
pulumi.export('lambda function', lambda_fn.arn)
  

  
import boto3
import os
from contextlib import closing

def lambda_handler(event, context):

    for record in event['Records']:

        dynamodb_record = record["dynamodb"]
        textID = dynamodb_record["Keys"]["Id"]["N"]
        text_content = dynamodb_record["NewImage"]["text"]
        
        
        print("Text to Speech function. Text ID in DynamoDB: {}".format(textID))
            

        #For each block, invoke Polly API, which will transform text into audio
        polly = boto3.client('polly')
        response = polly.synthesize_speech(
            OutputFormat='mp3',
            Text = text_content["S"],
            VoiceId = "Joey")
        
        #Save the audio stream returned by Amazon Polly on Lambda's temp directory.
        if "AudioStream" in response:
            with closing(response["AudioStream"]) as stream:
                output = os.path.join("/tmp/", textID)
                with open(output, "wb") as file:
                    file.write(stream.read())


        s3 = boto3.client('s3')

        s3.upload_file('/tmp/' + textID, 
        '', 
        textID + ".mp3")

        return
  

  
pulumi up
  

  
import unittest
import pulumi


class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, type_, name, inputs, provider, id_):
        return [name + '_id', inputs]
    def call(self, token, args, provider):
        return {}

pulumi.runtime.set_mocks(MyMocks())


# It's important to import the program after the mocks are defined.
import  as infra

class TestingWithMocks(unittest.TestCase):
    # check 1: Lambdas should have a minimum timeout of 30sec.
    @pulumi.runtime.test
    def test_lambda_timeout(self):
        def check_timeout(args):
            urn, timeout = args
            self.assertIsNotNone(timeout, f'lambda {urn} must have timeout')
            self.assertGreaterEqual(timeout,30,f'lambda {urn} must have a timeout greater or equal to 30')
        return pulumi.Output.all(infra.lambda_fn.urn, infra.lambda_fn.timeout).apply(check_timeout)

    # check 2: Bucket to have 'bucket' parameter.
    @pulumi.runtime.test
    def test_bucket_name(self):
        def check_name(args):
            urn, bucket = args
            self.assertIsNotNone(bucket, f'Bucket {urn} must have a bucket parameter with a name')
        return pulumi.Output.all(infra.bucket_tts.urn, infra.bucket_tts.bucket).apply(check_name)    


    # check 3: DynamoDB table should have stream_view_type as NEW_IMAGE
    @pulumi.runtime.test
    def test_dynamodb_stream_type(self):
        def check_stream_view(args):
            urn, stream_view_type = args
            self.assertIsNotNone(stream_view_type, f'Table {urn} should have stream_view_type parameter')
            self.assertEqual('NEW_IMAGE',stream_view_type,f'Table should have NEW_IMAGE as stream view type value')
        return pulumi.Output.all(infra.ddb.urn, infra.ddb.stream_view_type).apply(check_stream_view)
  

  
python -m unittest 
  

  
pip install pulumi-policy
  

  
description: A minimal Policy Pack for AWS using Python.
runtime: python
version: 0.0.1
  

  
from pulumi_policy import (
    EnforcementLevel,
    PolicyPack,
    ReportViolation,
    ResourceValidationArgs,
    ResourceValidationPolicy,
)
import json


# Check for DynamoDB stream enabled value
def dynamodb_stream_validation(args: ResourceValidationArgs, report_violation: ReportViolation):
    if args.resource_type == "aws:dynamodb/table:Table" and "streamEnabled" in args.props:
        stream_value = args.props["streamEnabled"]
        if stream_value != True:
            report_violation("For this program to work DynamoDB streams need to be enabled")

stream_check = ResourceValidationPolicy(
    name="dynamodb_stream_check",
    description="Check if streams are enabled.",
    validate=dynamodb_stream_validation,
)

# Check if policies are included
def iam_policy_validation(args: ResourceValidationArgs, report_violation: ReportViolation):
    if args.resource_type == "aws:iam/rolePolicy:RolePolicy" and "policy" in args.props:
        policy = json.loads(args.props["policy"])
        policies = policy["Statement"][0]["Action"]
        if "s3:PutObject" not in policies or "polly:SynthesizeSpeech" not in policies:
            report_violation("For this program to work S3 and Polly policies needed")

policy_check = ResourceValidationPolicy(
    name="s3-polly-policy",
    description="Check if permissions are present.",
    validate=iam_policy_validation,
)




PolicyPack(
    name="aws-python",
    enforcement_level=EnforcementLevel.MANDATORY,
    policies=[
        stream_check,
        policy_check
    ],
)
  

  
pulumi up --policy-pack tests