<s> import base64 <EOL> import collections <EOL> import json <EOL> import mock <EOL> import os <EOL> import random <EOL> import string <EOL> import unittest <EOL> from lambda_packages import lambda_packages <EOL> from . utils import placebo_session <EOL> from zappa . cli import ZappaCLI <EOL> from zappa . handler import LambdaHandler , lambda_handler <EOL> from zappa . wsgi import create_wsgi_request , common_log <EOL> from zappa . zappa import Zappa , ASSUME_POLICY , ATTACH_POLICY <EOL> def random_string ( length ) : <EOL> return '<STR_LIT>' . join ( random . choice ( string . printable ) for _ in range ( length ) ) <EOL> class TestZappa ( unittest . TestCase ) : <EOL> def test_test ( self ) : <EOL> self . assertTrue ( True ) <EOL> def test_zappa ( self ) : <EOL> self . assertTrue ( True ) <EOL> Zappa ( ) <EOL> def test_create_lambda_package ( self ) : <EOL> mock_named_tuple = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> mock_return_val = [ mock_named_tuple ( lambda_packages . keys ( ) [ <NUM_LIT:0> ] ) ] <EOL> with mock . patch ( '<STR_LIT>' , return_value = mock_return_val ) : <EOL> z = Zappa ( ) <EOL> path = z . create_lambda_zip ( handler_file = os . path . realpath ( __file__ ) ) <EOL> self . assertTrue ( os . path . isfile ( path ) ) <EOL> os . remove ( path ) <EOL> def test_load_credentials ( self ) : <EOL> z = Zappa ( ) <EOL> z . aws_region = '<STR_LIT>' <EOL> z . load_credentials ( ) <EOL> self . assertEqual ( z . boto_session . region_name , '<STR_LIT>' ) <EOL> self . assertEqual ( z . aws_region , '<STR_LIT>' ) <EOL> z . aws_region = '<STR_LIT>' <EOL> z . profile_name = '<STR_LIT:default>' <EOL> z . load_credentials ( ) <EOL> self . assertEqual ( z . boto_session . region_name , '<STR_LIT>' ) <EOL> self . assertEqual ( z . aws_region , '<STR_LIT>' ) <EOL> creds = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> with mock . patch . dict ( '<STR_LIT>' , creds ) : <EOL> z . aws_region = None <EOL> z . load_credentials ( ) <EOL> loaded_creds = z . boto_session . _session . get_credentials ( ) <EOL> self . assertEqual ( loaded_creds . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( loaded_creds . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( z . boto_session . region_name , '<STR_LIT>' ) <EOL> @ placebo_session <EOL> def test_upload_remove_s3 ( self , session ) : <EOL> bucket_name = '<STR_LIT>' <EOL> z = Zappa ( session ) <EOL> zip_path = z . create_lambda_zip ( minify = False ) <EOL> res = z . upload_to_s3 ( zip_path , bucket_name ) <EOL> os . remove ( zip_path ) <EOL> self . assertTrue ( res ) <EOL> s3 = session . resource ( '<STR_LIT>' ) <EOL> s3 . meta . client . head_bucket ( Bucket = bucket_name ) <EOL> s3 . meta . client . head_object ( <EOL> Bucket = bucket_name , <EOL> Key = zip_path , <EOL> ) <EOL> res = z . remove_from_s3 ( zip_path , bucket_name ) <EOL> self . assertTrue ( res ) <EOL> fail = z . upload_to_s3 ( '<STR_LIT>' , bucket_name ) <EOL> self . assertFalse ( fail ) <EOL> @ placebo_session <EOL> def test_create_lambda_function ( self , session ) : <EOL> bucket_name = '<STR_LIT>' <EOL> zip_path = '<STR_LIT>' <EOL> z = Zappa ( session ) <EOL> z . aws_region = '<STR_LIT>' <EOL> z . load_credentials ( session ) <EOL> z . credentials_arn = '<STR_LIT>' <EOL> arn = z . create_lambda_function ( <EOL> bucket = bucket_name , <EOL> s3_key = zip_path , <EOL> function_name = '<STR_LIT>' , <EOL> handler = '<STR_LIT>' <EOL> ) <EOL> arn = z . update_lambda_function ( <EOL> bucket = bucket_name , <EOL> s3_key = zip_path , <EOL> function_name = '<STR_LIT>' , <EOL> ) <EOL> @ placebo_session <EOL> def test_rollback_lambda_function_version ( self , session ) : <EOL> z = Zappa ( session ) <EOL> z . credentials_arn = '<STR_LIT>' <EOL> function_name = '<STR_LIT>' <EOL> too_many_versions = z . rollback_lambda_function_version ( function_name , <NUM_LIT> ) <EOL> self . assertFalse ( too_many_versions ) <EOL> function_arn = z . rollback_lambda_function_version ( function_name , <NUM_LIT:1> ) <EOL> @ placebo_session <EOL> def test_invoke_lambda_function ( self , session ) : <EOL> z = Zappa ( session ) <EOL> z . credentials_arn = '<STR_LIT>' <EOL> function_name = '<STR_LIT>' <EOL> payload = '<STR_LIT>' <EOL> response = z . invoke_lambda_function ( function_name , payload ) <EOL> @ placebo_session <EOL> def test_create_iam_roles ( self , session ) : <EOL> z = Zappa ( session ) <EOL> arn = z . create_iam_roles ( ) <EOL> self . assertEqual ( arn , "<STR_LIT>" . format ( z . role_name ) ) <EOL> @ placebo_session <EOL> def test_create_api_gateway_routes ( self , session ) : <EOL> z = Zappa ( session ) <EOL> z . parameter_depth = <NUM_LIT:1> <EOL> z . integration_response_codes = [ <NUM_LIT:200> ] <EOL> z . method_response_codes = [ <NUM_LIT:200> ] <EOL> z . http_methods = [ '<STR_LIT:GET>' ] <EOL> z . credentials_arn = '<STR_LIT>' <EOL> lambda_arn = '<STR_LIT>' <EOL> with mock . patch ( '<STR_LIT>' , return_value = <NUM_LIT> ) : <EOL> api_id = z . create_api_gateway_routes ( lambda_arn ) <EOL> self . assertEqual ( api_id , '<STR_LIT>' ) <EOL> @ placebo_session <EOL> def test_deploy_api_gateway ( self , session ) : <EOL> z = Zappa ( session ) <EOL> z . credentials_arn = '<STR_LIT>' <EOL> z . parameter_depth = <NUM_LIT:1> <EOL> z . integration_response_codes = [ <NUM_LIT:200> ] <EOL> z . method_response_codes = [ <NUM_LIT:200> ] <EOL> z . http_methods = [ '<STR_LIT:GET>' ] <EOL> lambda_arn = '<STR_LIT>' <EOL> api_id = z . create_api_gateway_routes ( lambda_arn ) <EOL> endpoint_url = z . deploy_api_gateway ( api_id , "<STR_LIT>" ) <EOL> @ placebo_session <EOL> def test_get_api_url ( self , session ) : <EOL> z = Zappa ( session ) <EOL> z . credentials_arn = '<STR_LIT>' <EOL> url = z . get_api_url ( '<STR_LIT>' ) <EOL> @ placebo_session <EOL> def test_fetch_logs ( self , session ) : <EOL> z = Zappa ( session ) <EOL> z . credentials_arn = '<STR_LIT>' <EOL> events = z . fetch_logs ( '<STR_LIT>' ) <EOL> self . assertTrue ( events != None ) <EOL> def test_policy_json ( self ) : <EOL> json . loads ( ASSUME_POLICY ) <EOL> json . loads ( ATTACH_POLICY ) <EOL> def test_logging ( self ) : <EOL> """<STR_LIT>""" <EOL> Zappa ( ) <EOL> def test_redirect_pattern ( self ) : <EOL> test_urls = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> for code in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> pattern = Zappa . selection_pattern ( code ) <EOL> for url in test_urls : <EOL> self . assertRegexpMatches ( url , pattern ) <EOL> def test_b64_pattern ( self ) : <EOL> head = '<STR_LIT>' <EOL> for code in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> pattern = Zappa . selection_pattern ( code ) <EOL> document = base64 . b64encode ( head + code + random_string ( <NUM_LIT:50> ) ) <EOL> self . assertRegexpMatches ( document , pattern ) <EOL> for bad_code in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> document = base64 . b64encode ( head + bad_code + random_string ( <NUM_LIT:50> ) ) <EOL> self . assertNotRegexpMatches ( document , pattern ) <EOL> def test_200_pattern ( self ) : <EOL> pattern = Zappa . selection_pattern ( '<STR_LIT>' ) <EOL> self . assertEqual ( pattern , '<STR_LIT>' ) <EOL> def test_wsgi_event ( self ) : <EOL> event = { <EOL> "<STR_LIT:body>" : { } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:false>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:false>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:false>" , <EOL> "<STR_LIT>" : "<STR_LIT:true>" , <EOL> "<STR_LIT:Content-Type>" : "<STR_LIT:application/json>" <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> request = create_wsgi_request ( event ) <EOL> def test_wsgi_path_info ( self ) : <EOL> event = { <EOL> "<STR_LIT:body>" : { } , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : { } <EOL> } <EOL> request = create_wsgi_request ( event , trailing_slash = True ) <EOL> self . assertEqual ( "<STR_LIT:/>" , request [ '<STR_LIT>' ] ) <EOL> request = create_wsgi_request ( event , trailing_slash = False ) <EOL> self . assertEqual ( "<STR_LIT:/>" , request [ '<STR_LIT>' ] ) <EOL> event = { <EOL> "<STR_LIT:body>" : { } , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : { } <EOL> } <EOL> request = create_wsgi_request ( event , trailing_slash = True ) <EOL> self . assertEqual ( "<STR_LIT>" , request [ '<STR_LIT>' ] ) <EOL> request = create_wsgi_request ( event , trailing_slash = False ) <EOL> self . assertEqual ( "<STR_LIT>" , request [ '<STR_LIT>' ] ) <EOL> request = create_wsgi_request ( event , trailing_slash = False , script_name = '<STR_LIT>' ) <EOL> self . assertEqual ( "<STR_LIT>" , request [ '<STR_LIT>' ] ) <EOL> def test_wsgi_logging ( self ) : <EOL> event = { <EOL> "<STR_LIT:body>" : { } , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : { } <EOL> } <EOL> environ = create_wsgi_request ( event , trailing_slash = False ) <EOL> response_tuple = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:content>' ] ) <EOL> response = response_tuple ( <NUM_LIT:200> , '<STR_LIT:hello>' ) <EOL> le = common_log ( environ , response , response_time = True ) <EOL> le = common_log ( environ , response , response_time = False ) <EOL> def test_wsgi_multipart ( self ) : <EOL> event = { u'<STR_LIT:body>' : u'<STR_LIT>' , u'<STR_LIT>' : { u'<STR_LIT:Content-Type>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT:false>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT:false>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT:false>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT:true>' } , u'<STR_LIT>' : { u'<STR_LIT>' : u'<STR_LIT>' } , u'<STR_LIT>' : u'<STR_LIT:POST>' , u'<STR_LIT>' : { } } <EOL> environ = create_wsgi_request ( event , trailing_slash = False ) <EOL> response_tuple = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:content>' ] ) <EOL> response = response_tuple ( <NUM_LIT:200> , '<STR_LIT:hello>' ) <EOL> def test_handler ( self ) : <EOL> lh = LambdaHandler ( '<STR_LIT>' ) <EOL> event = { <EOL> "<STR_LIT:body>" : { } , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : { } <EOL> } <EOL> lh . handler ( event , None ) <EOL> def test_cli_sanity ( self ) : <EOL> zappa_cli = ZappaCLI ( ) <EOL> return <EOL> def test_cli_utility ( self ) : <EOL> zappa_cli = ZappaCLI ( ) <EOL> zappa_cli . api_stage = '<STR_LIT>' <EOL> zappa_cli . load_settings ( '<STR_LIT>' ) <EOL> zappa_cli . create_package ( ) <EOL> zappa_cli . remove_local_zip ( ) <EOL> logs = [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:message>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:message>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:message>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:message>' : '<STR_LIT:test>' <EOL> } <EOL> ] <EOL> zappa_cli . print_logs ( logs ) <EOL> def test_cli_args ( self ) : <EOL> zappa_cli = ZappaCLI ( ) <EOL> argv = '<STR_LIT>' . split ( ) <EOL> zappa_cli . handle ( argv ) <EOL> @ placebo_session <EOL> def test_cli_aws ( self , session ) : <EOL> zappa_cli = ZappaCLI ( ) <EOL> zappa_cli . api_stage = '<STR_LIT>' <EOL> zappa_cli . load_settings ( '<STR_LIT>' , session ) <EOL> zappa_cli . zappa . credentials_arn = '<STR_LIT>' <EOL> zappa_cli . deploy ( ) <EOL> zappa_cli . update ( ) <EOL> zappa_cli . rollback ( <NUM_LIT:1> ) <EOL> zappa_cli . tail ( False ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
