KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_invoke_authorizer method in localstack

Best Python code snippet using localstack_python

test_parser.py

Source:test_parser.py

1import os2from bs4 import BeautifulSoup3import requests4from pyboto3 import boto3_interface_generator5from pyboto3.boto3_interface_generator import iter_params, iter_methods, get_method_description, get_param_description, \6    iter_code_lines, generate_all_services_code, iter_all_services, generate_service_code7def test_iter_methods_api_gateway():8    response = requests.get('https://boto3.readthedocs.io/en/latest/reference/services/apigateway.html')9    soup = BeautifulSoup(response._content, 'html.parser')10    methods = {'can_paginate', 'create_api_key', 'create_authorizer', 'create_base_path_mapping', 'create_deployment',11               'create_domain_name', 'create_model', 'create_resource', 'create_rest_api', 'create_stage',12               'delete_api_key', 'delete_authorizer', 'delete_base_path_mapping', 'delete_client_certificate',13               'delete_deployment', 'delete_domain_name', 'delete_integration', 'delete_integration_response',14               'delete_method', 'delete_method_response', 'delete_model', 'delete_resource', 'delete_rest_api',15               'delete_stage', 'flush_stage_authorizers_cache', 'flush_stage_cache', 'generate_client_certificate',16               'generate_presigned_url', 'get_account', 'get_api_key', 'get_api_keys', 'get_authorizer',17               'get_authorizers', 'get_base_path_mapping', 'get_base_path_mappings', 'get_client_certificate',18               'get_client_certificates', 'get_deployment', 'get_deployments', 'get_domain_name', 'get_domain_names',19               'get_export', 'get_integration', 'get_integration_response', 'get_method', 'get_method_response',20               'get_model', 'get_model_template', 'get_models', 'get_paginator', 'get_resource', 'get_resources',21               'get_rest_api', 'get_rest_apis', 'get_sdk', 'get_stage', 'get_stages', 'get_waiter', 'import_rest_api',22               'put_integration', 'put_integration_response', 'put_method', 'put_method_response', 'put_rest_api',23               'test_invoke_authorizer', 'test_invoke_method', 'update_account', 'update_api_key', 'update_authorizer',24               'update_base_path_mapping', 'update_client_certificate', 'update_deployment', 'update_domain_name',25               'update_integration', 'update_integration_response', 'update_method', 'update_method_response',26               'update_model', 'update_resource', 'update_rest_api', 'update_stage'}27    method_soup_dict = {name: html for name, html in iter_methods(soup.find(id='client'))}28    assert methods == set([str(x) for x in method_soup_dict.keys()])29    can_paginate_params = [('operation_name', 'string')]30    assert can_paginate_params == list(31        (name, type) for name, type, desc in iter_params(method_soup_dict['can_paginate']))32    create_api_key_params = [('name', 'string', 'The name of the ApiKey .'),33                             ('description', 'string', 'The description of the ApiKey .'),34                             ('enabled', 'boolean', 'Specifies whether the ApiKey can be used by callers.'),35                             ('stageKeys', 'list', 'Specifies whether the ApiKey can be used by callers.'36                                                   '\n(dict) --A reference to a unique stage identified in the format {restApiId}/{stage} .'37                                                   '\nrestApiId (string) --A list of Stage resources that are associated with the ApiKey resource.'38                                                   '\nstageName (string) --The stage name in the RestApi that the stage key references.\n\n\n')]39    assert create_api_key_params == list(40        (str(name), str(type), str(desc)) for name, type, desc in iter_params(method_soup_dict['create_api_key']))41    assert [('restApiId', 'string'), ('resourceId', 'string')] == list(42        (name, type) for name, type, desc in iter_params(method_soup_dict['get_resource']))43    for method, html in method_soup_dict.iteritems():44        param_dict = {param: type for param, type, description in iter_params(html)}45        print method, param_dict.keys()46def test_description():47    response = requests.get('https://boto3.readthedocs.io/en/latest/reference/services/apigateway.html')48    soup = BeautifulSoup(response._content, 'html.parser')49    method_soup_dict = {name: html for name, html in iter_methods(soup.find(id='client'))}50    can_paginate_description = 'Check if an operation can be paginated.'51    assert can_paginate_description == get_method_description(method_soup_dict.get('can_paginate'))52def test_signle_param_description():53    soup = BeautifulSoup(54        """<dd class="field-body"><strong>operation_name</strong> (<em>string</em>) -- The operation name.  This is the same name as the method name on the client.  For example, if the method name is <tt class="docutils literal"><span class="pre">create_foo</span></tt>, and you'd normally invoke the operation as <tt class="docutils literal"><span class="pre">client.create_foo(**kwargs)</span></tt>, if the <tt class="docutils literal"><span class="pre">create_foo</span></tt> operation can be paginated, you can use the call <tt class="docutils literal"><span class="pre">client.get_paginator("create_foo")</span></tt>.</dd>""",55        'html.parser')56    result = get_param_description(soup)57    print result58    assert result == """The operation name. This is the same name as the method name on the client. For example, if the method name is create_foo, and you'd normally invoke the operation as client.create_foo(**kwargs), if the create_foo operation can be paginated, you can use the call client.get_paginator("create_foo")."""59def test_multiple_param_description():60    soup = BeautifulSoup(61        """<dd class="field-body"><ul class="first simple"> <li><strong>restApiId</strong> (<em>string</em>) -- <p><strong>[REQUIRED]</strong></p> <p>The RestApi identifier under which the Authorizer will be created.</p> </li> <li><strong>name</strong> (<em>string</em>) -- <p><strong>[REQUIRED]</strong></p> <p>[Required] The name of the authorizer.</p> </li> <li><strong>type</strong> (<em>string</em>) -- <p><strong>[REQUIRED]</strong></p> <p>[Required] The type of the authorizer.</p> </li> <li><strong>authType</strong> (<em>string</em>) -- Optional customer-defined field, used in Swagger imports/exports. Has no functional impact.</li> <li><strong>authorizerUri</strong> (<em>string</em>) -- <p><strong>[REQUIRED]</strong></p> <p>[Required] Specifies the authorizer's Uniform Resource Identifier (URI).</p> </li> <li><strong>authorizerCredentials</strong> (<em>string</em>) -- Specifies the credentials required for the authorizer, if any.</li> <li><strong>identitySource</strong> (<em>string</em>) -- <p><strong>[REQUIRED]</strong></p> <p>[Required] The source of the identity in an incoming request.</p> </li> <li><strong>identityValidationExpression</strong> (<em>string</em>) -- A validation expression for the incoming identity.</li> <li><strong>authorizerResultTtlInSeconds</strong> (<em>integer</em>) -- The TTL of cached authorizer results.</li> </ul> </dd>""")62    result = get_param_description(soup)63def test_cloudformation():64    response = requests.get('https://boto3.readthedocs.io/en/latest/reference/services/cloudformation.html')65    soup = BeautifulSoup(response._content, 'html.parser')66    method_soup_dict = {name: html for name, html in iter_methods(soup.find(id='client'))}67    get_template_params = [('StackName', 'string')]68    assert get_template_params == list(69        (name, type) for name, type, desc in iter_params(method_soup_dict['get_template']))70def test_iter_methods_s3():71    response = requests.get('https://boto3.readthedocs.io/en/latest/reference/services/s3.html')72    soup = BeautifulSoup(response._content, 'html.parser')73    s3_methods = {'abort_multipart_upload', 'can_paginate', 'complete_multipart_upload', 'copy_object', 'create_bucket',74                  'create_multipart_upload', 'delete_bucket', 'delete_bucket_cors', 'delete_bucket_lifecycle',75                  'delete_bucket_policy', 'delete_bucket_replication', 'delete_bucket_tagging', 'delete_bucket_website',76                  'delete_object', 'delete_objects', 'download_file', 'generate_presigned_post',77                  'generate_presigned_url', 'get_bucket_accelerate_configuration', 'get_bucket_acl', 'get_bucket_cors',78                  'get_bucket_lifecycle', 'get_bucket_lifecycle_configuration', 'get_bucket_location',79                  'get_bucket_logging', 'get_bucket_notification', 'get_bucket_notification_configuration',80                  'get_bucket_policy', 'get_bucket_replication', 'get_bucket_request_payment', 'get_bucket_tagging',81                  'get_bucket_versioning', 'get_bucket_website', 'get_object', 'get_object_acl', 'get_object_torrent',82                  'get_paginator', 'get_waiter', 'head_bucket', 'head_object', 'list_buckets', 'list_multipart_uploads',83                  'list_object_versions', 'list_objects', 'list_objects_v2', 'list_parts',84                  'put_bucket_accelerate_configuration', 'put_bucket_acl', 'put_bucket_cors', 'put_bucket_lifecycle',85                  'put_bucket_lifecycle_configuration', 'put_bucket_logging', 'put_bucket_notification',86                  'put_bucket_notification_configuration', 'put_bucket_policy', 'put_bucket_replication',87                  'put_bucket_request_payment', 'put_bucket_tagging', 'put_bucket_versioning', 'put_bucket_website',88                  'put_object', 'put_object_acl', 'restore_object', 'upload_file', 'upload_part', 'upload_part_copy'}89    method_soup_dict = {name: html for name, html in iter_methods(soup.find(id='client'))}90    assert s3_methods == set([str(x) for x in method_soup_dict.keys()])91    abort_multipart_upload_params = [('Bucket', 'string'),92                                     ('Key', 'string'),93                                     ('UploadId', 'string'),94                                     ('RequestPayer', 'string')]95    assert abort_multipart_upload_params == list(96        (name, type) for name, type, desc in iter_params(method_soup_dict['abort_multipart_upload']))97    for method, html in method_soup_dict.iteritems():98        param_dict = {param: type for param, type, description in iter_params(html)}99        # print method, param_dict.keys()100    for line in iter_code_lines(soup):101        print line102def test_generate_code():103    generate_all_services_code(os.path.join(os.path.dirname(__file__), '..', 'pyboto3'))104def test_generate_rds_code():105    generate_service_code('https://boto3.readthedocs.io/en/latest/reference/services/rds.html',106                          os.path.join(os.path.dirname(__file__), '..', 'pyboto3', 'rds.py'))107def test_generate_s3_code():108    generate_service_code('https://boto3.readthedocs.io/en/latest/reference/services/s3.html',109                          os.path.join(os.path.dirname(__file__), '..', 'pyboto3', 's3.py'))110def test_get_services():111    services = list(iter_all_services())112    print services...

test_generator.py

Source:test_generator.py

1from pyboto3 import interface_generator2import json3def test_get_services_dir():4    assert interface_generator.get_services_dir()5    service_json_path_dict = dict(interface_generator.iter_services_json_paths())6    for service_name, json_path in service_json_path_dict.iteritems():7        print service_name, json_path8    apigateway_methods = {'can_paginate', 'create_api_key', 'create_authorizer', 'create_base_path_mapping',9                          'create_deployment',10                          'create_domain_name', 'create_model', 'create_resource', 'create_rest_api', 'create_stage',11                          'delete_api_key', 'delete_authorizer', 'delete_base_path_mapping',12                          'delete_client_certificate',13                          'delete_deployment', 'delete_domain_name', 'delete_integration',14                          'delete_integration_response',15                          'delete_method', 'delete_method_response', 'delete_model', 'delete_resource',16                          'delete_rest_api',17                          'delete_stage', 'flush_stage_authorizers_cache', 'flush_stage_cache',18                          'generate_client_certificate',19                          'generate_presigned_url', 'get_account', 'get_api_key', 'get_api_keys', 'get_authorizer',20                          'get_authorizers', 'get_base_path_mapping', 'get_base_path_mappings',21                          'get_client_certificate',22                          'get_client_certificates', 'get_deployment', 'get_deployments', 'get_domain_name',23                          'get_domain_names',24                          'get_export', 'get_integration', 'get_integration_response', 'get_method',25                          'get_method_response',26                          'get_model', 'get_model_template', 'get_models', 'get_paginator', 'get_resource',27                          'get_resources',28                          'get_rest_api', 'get_rest_apis', 'get_sdk', 'get_stage', 'get_stages', 'get_waiter',29                          'import_rest_api',30                          'put_integration', 'put_integration_response', 'put_method', 'put_method_response',31                          'put_rest_api',32                          'test_invoke_authorizer', 'test_invoke_method', 'update_account', 'update_api_key',33                          'update_authorizer',34                          'update_base_path_mapping', 'update_client_certificate', 'update_deployment',35                          'update_domain_name',36                          'update_integration', 'update_integration_response', 'update_method',37                          'update_method_response',38                          'update_model', 'update_resource', 'update_rest_api', 'update_stage'}39    apigateway_dict = json.load(open(service_json_path_dict.get('apigateway')))40    assert set(interface_generator.iter_method_names(apigateway_dict)) == apigateway_methods41    s3_methods = {'abort_multipart_upload', 'can_paginate', 'complete_multipart_upload', 'copy_object',42                  'create_bucket',43                  'create_multipart_upload', 'delete_bucket', 'delete_bucket_cors', 'delete_bucket_lifecycle',44                  'delete_bucket_policy', 'delete_bucket_replication', 'delete_bucket_tagging',45                  'delete_bucket_website',46                  'delete_object', 'delete_objects', 'download_file', 'generate_presigned_post',47                  'generate_presigned_url', 'get_bucket_accelerate_configuration', 'get_bucket_acl',48                  'get_bucket_cors',49                  'get_bucket_lifecycle', 'get_bucket_lifecycle_configuration', 'get_bucket_location',50                  'get_bucket_logging', 'get_bucket_notification', 'get_bucket_notification_configuration',51                  'get_bucket_policy', 'get_bucket_replication', 'get_bucket_request_payment', 'get_bucket_tagging',52                  'get_bucket_versioning', 'get_bucket_website', 'get_object', 'get_object_acl',53                  'get_object_torrent',54                  'get_paginator', 'get_waiter', 'head_bucket', 'head_object', 'list_buckets',55                  'list_multipart_uploads',56                  'list_object_versions', 'list_objects', 'list_objects_v2', 'list_parts',57                  'put_bucket_accelerate_configuration', 'put_bucket_acl', 'put_bucket_cors',58                  'put_bucket_lifecycle',59                  'put_bucket_lifecycle_configuration', 'put_bucket_logging', 'put_bucket_notification',60                  'put_bucket_notification_configuration', 'put_bucket_policy', 'put_bucket_replication',61                  'put_bucket_request_payment', 'put_bucket_tagging', 'put_bucket_versioning', 'put_bucket_website',62                  'put_object', 'put_object_acl', 'restore_object', 'upload_file', 'upload_part',63                  'upload_part_copy'}64    s3_dict = json.load(open(service_json_path_dict.get('s3')))...

test_invoke.py

Source:test_invoke.py

...12)13def test_unsigned_invalid(stack_params: stack_params):14    # This is a sanity check to make sure that the authorizer actually works, we don't care about the values15    if stack_params["UnsignedAuthorizerStatus"]:16        response = iot_client.test_invoke_authorizer(17            authorizerName=stack_params["UnsignedAuthorizerName"],18            mqttContext=MqttContextTypeDef(19                username=str(uuid4()), password=str(uuid4()), clientId=str(uuid4())20            ),21        )22        assert response["isAuthenticated"] is False23@pytest.mark.dependency()24def test_unsigned_valid(stack_params: stack_params, generate_table: generate_table):25    if stack_params["UnsignedAuthorizerStatus"]:26        response = iot_client.test_invoke_authorizer(27            authorizerName=stack_params["UnsignedAuthorizerName"],28            mqttContext=MqttContextTypeDef(29                username=test_user, password=test_pass, clientId=test_client30            ),31        )32        assert response["ResponseMetadata"]["HTTPStatusCode"] == 20033        assert response["isAuthenticated"] is True34        policy_doc = loads(response["policyDocuments"][0])...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.