implements APIGateway update_api_key
This commit is contained in:
		
							parent
							
								
									5c34c06d07
								
							
						
					
					
						commit
						7fa5ce3dc3
					
				
							
								
								
									
										1
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										1
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							| @ -15,6 +15,7 @@ python_env | ||||
| .ropeproject/ | ||||
| .pytest_cache/ | ||||
| venv/ | ||||
| env/ | ||||
| .python-version | ||||
| .vscode/ | ||||
| tests/file.tmp | ||||
|  | ||||
| @ -181,7 +181,7 @@ | ||||
| - [ ] test_invoke_method | ||||
| - [ ] untag_resource | ||||
| - [ ] update_account | ||||
| - [ ] update_api_key | ||||
| - [X] update_api_key | ||||
| - [ ] update_authorizer | ||||
| - [ ] update_base_path_mapping | ||||
| - [ ] update_client_certificate | ||||
|  | ||||
| @ -309,6 +309,25 @@ class ApiKey(BaseModel, dict): | ||||
|         self['createdDate'] = self['lastUpdatedDate'] = int(time.time()) | ||||
|         self['stageKeys'] = stageKeys | ||||
| 
 | ||||
|     def update_operations(self, patch_operations): | ||||
|         for op in patch_operations: | ||||
|             if op['op'] == 'replace': | ||||
|                 if '/name' in op['path']: | ||||
|                     self['name'] = op['value'] | ||||
|                 elif '/customerId' in op['path']: | ||||
|                     self['customerId'] = op['value'] | ||||
|                 elif '/description' in op['path']: | ||||
|                     self['description'] = op['value'] | ||||
|                 elif '/enabled' in op['path']: | ||||
|                     self['enabled'] = self._str2bool(op['value']) | ||||
|             else: | ||||
|                 raise Exception( | ||||
|                     'Patch operation "%s" not implemented' % op['op']) | ||||
|         return self | ||||
| 
 | ||||
|     def _str2bool(self, v): | ||||
|         return v.lower() == "true" | ||||
| 
 | ||||
| 
 | ||||
| class UsagePlan(BaseModel, dict): | ||||
| 
 | ||||
| @ -599,6 +618,10 @@ class APIGatewayBackend(BaseBackend): | ||||
|     def get_apikey(self, api_key_id): | ||||
|         return self.keys[api_key_id] | ||||
| 
 | ||||
|     def update_apikey(self, api_key_id, patch_operations): | ||||
|         key = self.keys[api_key_id] | ||||
|         return key.update_operations(patch_operations) | ||||
| 
 | ||||
|     def delete_apikey(self, api_key_id): | ||||
|         self.keys.pop(api_key_id) | ||||
|         return {} | ||||
|  | ||||
| @ -245,6 +245,9 @@ class APIGatewayResponse(BaseResponse): | ||||
| 
 | ||||
|         if self.method == 'GET': | ||||
|             apikey_response = self.backend.get_apikey(apikey) | ||||
|         elif self.method == 'PATCH': | ||||
|             patch_operations = self._get_param('patchOperations') | ||||
|             apikey_response = self.backend.update_apikey(apikey, patch_operations) | ||||
|         elif self.method == 'DELETE': | ||||
|             apikey_response = self.backend.delete_apikey(apikey) | ||||
|         return 200, {}, json.dumps(apikey_response) | ||||
|  | ||||
| @ -988,13 +988,30 @@ def test_api_keys(): | ||||
|     apikey['name'].should.equal(apikey_name) | ||||
|     len(apikey['value']).should.equal(40) | ||||
| 
 | ||||
|     apikey_name = 'TESTKEY3' | ||||
|     payload = {'name': apikey_name } | ||||
|     response = client.create_api_key(**payload) | ||||
|     apikey_id = response['id'] | ||||
| 
 | ||||
|     patch_operations = [ | ||||
|         {'op': 'replace', 'path': '/name', 'value': 'TESTKEY3_CHANGE'}, | ||||
|         {'op': 'replace', 'path': '/customerId', 'value': '12345'}, | ||||
|         {'op': 'replace', 'path': '/description', 'value': 'APIKEY UPDATE TEST'}, | ||||
|         {'op': 'replace', 'path': '/enabled', 'value': 'false'}, | ||||
|     ] | ||||
|     response = client.update_api_key(apiKey=apikey_id, patchOperations=patch_operations) | ||||
|     response['name'].should.equal('TESTKEY3_CHANGE') | ||||
|     response['customerId'].should.equal('12345') | ||||
|     response['description'].should.equal('APIKEY UPDATE TEST') | ||||
|     response['enabled'].should.equal(False) | ||||
| 
 | ||||
|     response = client.get_api_keys() | ||||
|     len(response['items']).should.equal(2) | ||||
|     len(response['items']).should.equal(3) | ||||
| 
 | ||||
|     client.delete_api_key(apiKey=apikey_id) | ||||
| 
 | ||||
|     response = client.get_api_keys() | ||||
|     len(response['items']).should.equal(1) | ||||
|     len(response['items']).should.equal(2) | ||||
| 
 | ||||
| @mock_apigateway | ||||
| def test_usage_plans(): | ||||
|  | ||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user