Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use update_authorizer method in localstack
Best Python code snippet using localstack_python
test_apigw_authorizer.py
Source:test_apigw_authorizer.py 
1#!/usr/bin/python2# TODO: License goes here3import library.apigw_authorizer as apigw_authorizer4from library.apigw_authorizer import ApiGwAuthorizer5import mock6from mock import patch7from mock import create_autospec8from mock import ANY9import unittest10import boto11from botocore.exceptions import BotoCoreError12import copy13class TestApiGwAuthorizer(unittest.TestCase):14 def setUp(self):15 self.module = mock.MagicMock()16 self.module.check_mode = False17 self.module.exit_json = mock.MagicMock()18 self.module.fail_json = mock.MagicMock()19 self.authorizer = ApiGwAuthorizer(self.module)20 self.authorizer.client = mock.MagicMock()21 self.authorizer.module.params = {22 'rest_api_id': 'rest_id',23 'name': 'testify',24 'type': 'TOKEN',25 'uri': 'my uri',26 'identity_source': 'source-arn',27 'auth_type': 'yolo',28 'state': 'present',29 }30 reload(apigw_authorizer)31 def test_boto_module_not_found(self):32 # Setup Mock Import Function33 import __builtin__ as builtins34 real_import = builtins.__import__35 def mock_import(name, *args):36 if name == 'boto': raise ImportError37 return real_import(name, *args)38 with mock.patch('__builtin__.__import__', side_effect=mock_import):39 reload(apigw_authorizer)40 ApiGwAuthorizer(self.module)41 self.module.fail_json.assert_called_with(msg='boto and boto3 are required for this module')42 def test_boto3_module_not_found(self):43 # Setup Mock Import Function44 import __builtin__ as builtins45 real_import = builtins.__import__46 def mock_import(name, *args):47 if name == 'boto3': raise ImportError48 return real_import(name, *args)49 with mock.patch('__builtin__.__import__', side_effect=mock_import):50 reload(apigw_authorizer)51 ApiGwAuthorizer(self.module)52 self.module.fail_json.assert_called_with(msg='boto and boto3 are required for this module')53 @patch.object(apigw_authorizer, 'boto3')54 def test_boto3_client_properly_instantiated(self, mock_boto):55 ApiGwAuthorizer(self.module)56 mock_boto.client.assert_called_once_with('apigateway')57 @patch.object(ApiGwAuthorizer, '_update_authorizer', return_value=(None, None))58 def test_process_request_calls_get_authorizers_and_stores_result_when_invoked(self, m):59 resp = {60 'items': [61 {'id': 'nope', 'name': 'nope'},62 {'id': 'match', 'name': 'testify'}63 ],64 }65 self.authorizer.client.get_authorizers = mock.MagicMock(return_value=resp)66 self.authorizer.process_request()67 self.assertEqual(resp['items'][1], self.authorizer.me)68 self.authorizer.client.get_authorizers.assert_called_once_with(restApiId='rest_id')69 def test_process_request_stores_None_result_when_not_found_in_get_authorizers_result(self):70 resp = {71 'items': [72 {'id': 'nope', 'name': 'nope'},73 {'id': 'not match', 'name': 'also nope'}74 ],75 }76 self.authorizer.client.get_authorizers = mock.MagicMock(return_value=resp)77 self.authorizer.process_request()78 self.assertEqual(None, self.authorizer.me)79 self.authorizer.client.get_authorizers.assert_called_once_with(restApiId='rest_id')80 def test_process_request_calls_fail_json_when_get_authorizers_raises_exception(self):81 self.authorizer.client.get_authorizers = mock.MagicMock(side_effect=BotoCoreError())82 self.authorizer.process_request()83 self.authorizer.client.get_authorizers.assert_called_once_with(restApiId='rest_id')84 self.authorizer.module.fail_json.assert_called_once_with(85 msg='Error when getting authorizers from boto3: An unspecified error occurred'86 )87 @patch.object(ApiGwAuthorizer, '_delete_authorizer', return_value='sprinkles')88 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'id': 'found'})89 def test_process_request_calls_exit_json_with_expected_value_after_successful_delete(self, mra, mda):90 self.authorizer.module.params = {91 'rest_api_id': 'rest_id',92 'name': 'testify',93 'state': 'absent',94 }95 self.authorizer.process_request()96 self.authorizer.module.exit_json.assert_called_once_with(changed='sprinkles', authorizer=None)97 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'id': 'found'})98 def test_process_request_calls_delete_authorizer_when_state_absent_and_authorizer_found(self, m):99 self.authorizer.module.params = {100 'rest_api_id': 'rest_id',101 'name': 'testify',102 'state': 'absent',103 }104 self.authorizer.process_request()105 self.authorizer.client.delete_authorizer.assert_called_once_with(106 restApiId='rest_id',107 authorizerId='found'108 )109 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'id': 'found'})110 def test_process_request_skips_delete_and_calls_exit_json_with_true_when_check_mode_set_and_auth_found(self, m):111 self.authorizer.module.params = {112 'rest_api_id': 'rest_id',113 'name': 'testify',114 'state': 'absent',115 }116 self.authorizer.module.check_mode = True117 self.authorizer.process_request()118 self.assertEqual(0, self.authorizer.client.delete_authorizer.call_count)119 self.authorizer.module.exit_json.assert_called_once_with(changed=True, authorizer=None)120 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'id': 'found'})121 def test_process_request_calls_fail_json_when_delete_authorizer_raises_error(self, m):122 self.authorizer.module.params = {123 'rest_api_id': 'rest_id',124 'name': 'testify',125 'state': 'absent',126 }127 self.authorizer.client.delete_authorizer = mock.MagicMock(side_effect=BotoCoreError)128 self.authorizer.process_request()129 self.authorizer.client.delete_authorizer.assert_called_once_with(130 restApiId='rest_id',131 authorizerId='found'132 )133 self.authorizer.module.fail_json.assert_called_once_with(134 msg='Error when deleting authorizer via boto3: An unspecified error occurred'135 )136 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value=None)137 def test_process_request_skips_delete_when_authorizer_not_found(self, m):138 self.authorizer.module.params = {139 'rest_api_id': 'rest_id',140 'name': 'testify',141 'state': 'absent',142 }143 self.authorizer.process_request()144 self.assertEqual(0, self.authorizer.client.delete_authorizer.call_count)145 @patch.object(ApiGwAuthorizer, '_create_authorizer', return_value=('time', 'lord'))146 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value=None)147 def test_process_request_calls_exit_json_with_expected_value_after_successful_create(self, mra, mca):148 self.authorizer.process_request()149 self.authorizer.module.exit_json.assert_called_once_with(changed='time', authorizer='lord')150 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value=None)151 def test_process_request_returns_create_authorizer_result_when_create_succeeds(self, m):152 self.authorizer.client.create_authorizer = mock.MagicMock(return_value='woot')153 self.authorizer.process_request()154 self.authorizer.module.exit_json.assert_called_once_with(changed=True, authorizer='woot')155 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value=None)156 def test_process_request_calls_create_authorizer_when_state_present_and_authorizer_not_found(self, m):157 self.authorizer.process_request()158 self.authorizer.client.create_authorizer.assert_called_once_with(159 restApiId='rest_id',160 name='testify',161 type='TOKEN',162 authType='yolo',163 authorizerUri='my uri',164 identitySource='source-arn'165 )166 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value=None)167 def test_process_request_calls_fail_json_when_create_authorizer_raises_exception(self, m):168 self.authorizer.client.create_authorizer = mock.MagicMock(side_effect=BotoCoreError())169 self.authorizer.process_request()170 self.authorizer.client.create_authorizer.assert_called_once_with(171 restApiId='rest_id',172 name='testify',173 type='TOKEN',174 authType='yolo',175 authorizerUri='my uri',176 identitySource='source-arn'177 )178 self.authorizer.module.fail_json.assert_called_once_with(179 msg='Error when creating authorizer via boto3: An unspecified error occurred'180 )181 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value=None)182 def test_process_request_skips_create_call_and_returns_changed_True_when_check_mode(self, m):183 self.authorizer.module.check_mode = True184 self.authorizer.process_request()185 self.assertEqual(0, self.authorizer.client.create_authorizer.call_count)186 self.authorizer.module.exit_json.assert_called_once_with(changed=True, authorizer=None)187 @patch.object(ApiGwAuthorizer, '_create_authorizer', return_value=(None, None))188 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer')189 def test_process_request_calls_fail_json_when_state_present_and_required_fields_missing(self, mra, mca):190 orig_params = copy.deepcopy(self.authorizer.module.params)191 tests = [192 {'field': 'type', 'message': 'Field <type> is required when state is present'},193 {'field': 'identity_source', 'message': 'Field <identity_source> is required when state is present'},194 ]195 for t in tests:196 self.authorizer.module.params = copy.deepcopy(orig_params)197 self.authorizer.module.params.pop(t['field'], None)198 self.authorizer.process_request()199 self.authorizer.module.fail_json.assert_called_with(msg=t['message'])200 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer')201 def test_process_request_calls_update_authorizer_when_state_present_and_authorizer_changed(self, m):202 self.authorizer.module.params = {203 'rest_api_id': 'rest_id',204 'name': 'testify',205 'type': 'TOKEN',206 'uri': 'my uri',207 'identity_source': 'source-arn',208 'auth_type': 'yolo',209 'result_ttl_seconds': 12345,210 'identity_validation_expression': 'add me',211 'provider_arns': ['not', 'order', 'in'],212 'state': 'present',213 }214 m.return_value = {215 'authType': 'orig_auth_type',216 'authorizerResultTtlInSeconds': 24601,217 'authorizerUri': 'orig_auth_uri',218 'id': 'id12345',219 'identitySource': 'orig_identity_source',220 'providerARNs': ['not', 'in', 'order'],221 'name': 'testify',222 'type': 'TOKEN',223 'authorizerCredentials': 'orig_creds',224 }225 expected_patches = [226 {'op': 'replace', 'path': '/identityValidationExpression', 'value': 'add me'},227 {'op': 'replace', 'path': '/authorizerCredentials', 'value': ''},228 {'op': 'replace', 'path': '/authorizerUri', 'value': 'my uri'},229 {'op': 'replace', 'path': '/identitySource', 'value': 'source-arn'},230 {'op': 'replace', 'path': '/authorizerResultTtlInSeconds', 'value': '12345'},231 {'op': 'replace', 'path': '/authType', 'value': 'yolo'},232 ]233 self.authorizer.process_request()234 self.authorizer.client.update_authorizer.assert_called_once_with(235 restApiId='rest_id',236 authorizerId='id12345',237 patchOperations=mock.ANY238 )239 self.assertItemsEqual(expected_patches, self.authorizer.client.update_authorizer.call_args[1]['patchOperations'])240 @patch('library.apigw_authorizer.ApiGwAuthorizer._create_patches', return_value=[])241 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'something': 'here'})242 def test_process_request_skips_update_authorizer_and_replies_false_when_no_changes(self, m, mcp):243 self.authorizer.process_request()244 self.assertEqual(0, self.authorizer.client.update_authorizer.call_count)245 self.authorizer.module.exit_json.assert_called_once_with(changed=False, authorizer={'something': 'here'})246 @patch('library.apigw_authorizer.ApiGwAuthorizer._create_patches', return_value=['patches!'])247 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'id': 'hi'})248 def test_process_request_calls_fail_json_when_update_authorizer_raises_exception(self, m, mcp):249 self.authorizer.client.update_authorizer = mock.MagicMock(side_effect=BotoCoreError())250 self.authorizer.process_request()251 self.authorizer.client.update_authorizer.assert_called_once_with(252 restApiId='rest_id',253 authorizerId='hi',254 patchOperations=['patches!']255 )256 self.authorizer.module.fail_json.assert_called_once_with(257 msg='Error when updating authorizer via boto3: An unspecified error occurred'258 )259 @patch('library.apigw_authorizer.ApiGwAuthorizer._create_patches', return_value=['patches!'])260 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', side_effect=[{'id': 'hi'}, 'second call'])261 def test_process_request_returns_result_of_find_when_update_is_successful(self, m, mcp):262 self.authorizer.process_request()263 self.authorizer.client.update_authorizer.assert_called_once_with(264 restApiId='rest_id',265 authorizerId='hi',266 patchOperations=['patches!']267 )268 self.authorizer.module.exit_json.assert_called_once_with(changed=True, authorizer='second call')269 @patch('library.apigw_authorizer.ApiGwAuthorizer._create_patches', return_value=['patches!'])270 @patch.object(ApiGwAuthorizer, '_retrieve_authorizer', return_value={'something': 'here'})271 def test_process_request_skips_update_authorizer_and_replies_true_when_check_mode(self, m, mcp):272 self.authorizer.module.check_mode = True273 self.authorizer.process_request()274 self.assertEqual(0, self.authorizer.client.update_authorizer.call_count)275 self.authorizer.module.exit_json.assert_called_once_with(changed=True, authorizer={'something': 'here'})276 def test_define_argument_spec(self):277 result = ApiGwAuthorizer._define_module_argument_spec()278 self.assertIsInstance(result, dict)279 self.assertEqual(result, dict(280 rest_api_id=dict(required=True),281 name=dict(required=True),282 type=dict(required=False, choices=['TOKEN', 'REQUEST', 'COGNITO_USER_POOLS']),283 uri=dict(required=False),284 identity_source=dict(required=False),285 identity_validation_expression=dict(required=False, default=''),286 provider_arns=dict(required=False, type='list', default=[]),287 auth_type=dict(required=False),288 credentials=dict(required=False),289 result_ttl_seconds=dict(required=False, type='int', default=0),290 state=dict(default='present', choices=['present', 'absent']),291 ))292 @patch.object(apigw_authorizer, 'AnsibleModule')293 @patch.object(apigw_authorizer, 'ApiGwAuthorizer')294 def test_main(self, mock_ApiGwAuthorizer, mock_AnsibleModule):295 mock_ApiGwAuthorizer_instance = mock.MagicMock()296 mock_AnsibleModule_instance = mock.MagicMock()297 mock_ApiGwAuthorizer.return_value = mock_ApiGwAuthorizer_instance298 mock_AnsibleModule.return_value = mock_AnsibleModule_instance299 apigw_authorizer.main()300 mock_ApiGwAuthorizer.assert_called_once_with(mock_AnsibleModule_instance)301 assert mock_ApiGwAuthorizer_instance.process_request.call_count == 1302if __name__ == '__main__':...
apis.py
Source:apis.py
...52 """53 query_auth_result = msg.query_auth_result54 authorizer_appid = query_auth_result['authorization_info']['authorizer_appid']55 info_result = component.get_authorizer_info(authorizer_appid)56 wx_authorizer: WxAuthorizer = update_authorizer(57 authorizer_appid, query_auth_result['authorization_info'], info_result['authorizer_info'])58 content = 'å°ç¨åºæ´æ°ææ, {nick_name}'.format(nick_name=wx_authorizer.nick_name)59 auth_client.message.send_text(60 agent_id=auth_agent_id,61 party_ids=cfg['party_id']['all'],62 user_ids=[],63 content=content64 )65 return Response('success', mimetype='text/plain')66 elif msg.type == 'unauthorized':67 update_unauthorized(msg.authorizer_appid)68 wx_authorizer: WxAuthorizer = WxAuthorizer.query.filter(WxAuthorizer.app_id == msg.authorizer_appid).first()69 content = 'å°ç¨åºåæ¶ææ, {nick_name}'.format(nick_name=wx_authorizer.nick_name)70 auth_client.message.send_text(71 agent_id=auth_agent_id,72 party_ids=cfg['party_id']['all'],73 user_ids=[],74 content=content75 )76 return Response('success', mimetype='text/plain')77def update_unauthorized(authorizer_appid):78 now = datetime.now()79 authorizer = WxAuthorizer.query.filter(WxAuthorizer.app_id == authorizer_appid).first()80 if not authorizer:81 return82 authorizer.is_authorized = False83 authorizer.unauthorized_at = now84 authorizer.modified_at = now85 db.session.commit()86@blueprint.route('/authorized/<string:auth_link_token>', methods=['GET'])87def get_authorized(auth_link_token):88 auth_code = request.args.get('auth_code', '')89 expires_in = request.args.get('expires_in', '')90 auth_cache = AuthLinkCache(token=auth_link_token)91 biz_user_id, biz_id, mark = auth_cache.get('biz_user_id', 'biz_id', 'mark')92 query_auth_result = component.query_auth(auth_code)93 authorizer_appid = query_auth_result['authorization_info']['authorizer_appid']94 info_result = component.get_authorizer_info(authorizer_appid)95 wx_authorizer: WxAuthorizer = update_authorizer(96 authorizer_appid, query_auth_result['authorization_info'], info_result['authorizer_info'])97 now = datetime.now()98 wx_authorizer.biz_id = biz_id99 wx_authorizer.mark = mark100 wx_authorizer.modified_at = now101 db.session.commit()102 content = 'ææ°çå°ç¨åºææ, {nick_name}'.format(nick_name=wx_authorizer.nick_name)103 auth_client.message.send_text(104 agent_id=auth_agent_id,105 party_ids=cfg['party_id']['all'],106 user_ids=[],107 content=content108 )109 redirect_uri = cfg['wxopen']['SUCCESS_REDIRECT_URI']...
test_apigatewayv2_authorizers.py
Source:test_apigatewayv2_authorizers.py
...91 IdentityValidationExpression="ive",92 JwtConfiguration={"Audience": ["a1"], "Issuer": "moto.com"},93 Name="auth1",94 )["AuthorizerId"]95 resp = client.update_authorizer(ApiId=api_id, AuthorizerId=auth_id, Name="auth2")96 resp.should.have.key("AuthorizerId")97 resp.should.have.key("AuthorizerCredentialsArn").equals("auth:creds:arn")98 resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")99 resp.should.have.key("AuthorizerResultTtlInSeconds").equals(3)100 resp.should.have.key("AuthorizerType").equals("REQUEST")101 resp.should.have.key("AuthorizerUri").equals("auth_uri")102 resp.should.have.key("EnableSimpleResponses").equals(True)103 resp.should.have.key("IdentitySource").equals(["$request.header.Authorization"])104 resp.should.have.key("IdentityValidationExpression").equals("ive")105 resp.should.have.key("JwtConfiguration").equals(106 {"Audience": ["a1"], "Issuer": "moto.com"}107 )108 resp.should.have.key("Name").equals("auth2")109@mock_apigatewayv2110def test_update_authorizer_all_attributes():111 client = boto3.client("apigatewayv2", region_name="eu-west-1")112 api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]113 auth_id = client.create_authorizer(114 ApiId=api_id, AuthorizerType="REQUEST", IdentitySource=[], Name="auth1"115 )["AuthorizerId"]116 auth_id = client.update_authorizer(117 ApiId=api_id,118 AuthorizerId=auth_id,119 AuthorizerCredentialsArn="",120 AuthorizerPayloadFormatVersion="3.0",121 AuthorizerResultTtlInSeconds=5,122 AuthorizerType="REQUEST",123 AuthorizerUri="auth_uri",124 EnableSimpleResponses=False,125 IdentitySource=["$request.header.Authentication"],126 IdentityValidationExpression="ive2",127 JwtConfiguration={"Audience": ["a2"], "Issuer": "moto.com"},128 Name="auth1",129 )["AuthorizerId"]130 resp = client.update_authorizer(ApiId=api_id, AuthorizerId=auth_id, Name="auth2")131 resp.should.have.key("AuthorizerId")132 resp.should.have.key("AuthorizerCredentialsArn").equals("")133 resp.should.have.key("AuthorizerPayloadFormatVersion").equals("3.0")134 resp.should.have.key("AuthorizerResultTtlInSeconds").equals(5)135 resp.should.have.key("AuthorizerType").equals("REQUEST")136 resp.should.have.key("AuthorizerUri").equals("auth_uri")137 resp.should.have.key("EnableSimpleResponses").equals(False)138 resp.should.have.key("IdentitySource").equals(["$request.header.Authentication"])139 resp.should.have.key("IdentityValidationExpression").equals("ive2")140 resp.should.have.key("JwtConfiguration").equals(141 {"Audience": ["a2"], "Issuer": "moto.com"}142 )...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
