def testLocation()

in datasync_log_prep/lambda_function.py [0:0]

• 16 lines of code
• 5 McCabe index (conditional complexity)

def testLocation(testLoc):
    listLocations = ds.list_locations()
    allLocations = listLocations['Locations']
    myLoc = next((item for item in allLocations if item['LocationArn'] == testLoc), False)
    if 'myLoc' in locals():
        if not myLoc['LocationUri'].startswith('s3'):
            raise ValueError("The destination location is not an S3 bucket. This solution can only be deployed for AWS DataSync tasks that have a destination location that is an S3 bucket.")
        else:
            print("Destination Location is S3 Bucket.  Updating lambda access role...")
            bucketParts = urlparse(myLoc['LocationUri'], allow_fragments=False)
            bucketArn = "arn:aws:s3:::"+bucketParts.netloc
            bucketPath = "arn:aws:s3:::"+bucketParts.netloc+"/*"
            bucketPolicy = '{"Version":"2012-10-17","Statement":{"Effect":"Allow","Action":["s3:ListBucket","s3:GetBucketLocation","s3:ListBucketMultipartUploads"],"Resource":"'+bucketArn+'"}}'
            objectPolicy = '{"Version":"2012-10-17","Statement":{"Effect":"Allow","Action":["s3:GetObject","s3:PutObject","s3:GetObjectAcl","s3:PutObjectAcl"],"Resource":"'+bucketPath+'"}}'
            iam.put_role_policy(RoleName=lambda_role,PolicyName="BucketActions",PolicyDocument=bucketPolicy)
            iam.put_role_policy(RoleName=lambda_role,PolicyName="ObjectActions",PolicyDocument=objectPolicy)