Update kms and lambda to work with terraform
This commit is contained in:
William Rubel
parent
1095b7d94b
commit
4c6f08d463
@ -500,6 +500,11 @@ class LambdaStorage(object):
|
||||
except ValueError:
|
||||
return self._functions[name]['latest']
|
||||
|
||||
def list_versions_by_function(self, name):
|
||||
if name not in self._functions:
|
||||
return None
|
||||
return [self._functions[name]['latest']]
|
||||
|
||||
def get_arn(self, arn):
|
||||
return self._arns.get(arn, None)
|
||||
|
||||
@ -607,6 +612,9 @@ class LambdaBackend(BaseBackend):
|
||||
def get_function(self, function_name, qualifier=None):
|
||||
return self._lambdas.get_function(function_name, qualifier)
|
||||
|
||||
def list_versions_by_function(self, function_name):
|
||||
return self._lambdas.list_versions_by_function(function_name)
|
||||
|
||||
def get_function_by_arn(self, function_arn):
|
||||
return self._lambdas.get_arn(function_arn)
|
||||
|
||||
|
||||
@ -52,7 +52,11 @@ class LambdaResponse(BaseResponse):
|
||||
self.setup_class(request, full_url, headers)
|
||||
if request.method == 'GET':
|
||||
# This is ListVersionByFunction
|
||||
raise ValueError("Cannot handle request")
|
||||
|
||||
path = request.path if hasattr(request, 'path') else path_url(request.url)
|
||||
function_name = path.split('/')[-2]
|
||||
return self._list_versions_by_function(function_name)
|
||||
|
||||
elif request.method == 'POST':
|
||||
return self._publish_function(request, full_url, headers)
|
||||
else:
|
||||
@ -151,6 +155,17 @@ class LambdaResponse(BaseResponse):
|
||||
|
||||
return 200, {}, json.dumps(result)
|
||||
|
||||
def _list_versions_by_function(self, function_name):
|
||||
result = {
|
||||
'Versions': []
|
||||
}
|
||||
|
||||
for fn in self.lambda_backend.list_versions_by_function(function_name):
|
||||
json_data = fn.get_configuration()
|
||||
result['Versions'].append(json_data)
|
||||
|
||||
return 200, {}, json.dumps(result)
|
||||
|
||||
def _create_function(self, request, full_url, headers):
|
||||
try:
|
||||
fn = self.lambda_backend.create_function(self.json_body)
|
||||
|
||||
@ -21,6 +21,7 @@ class Key(BaseModel):
|
||||
self.account_id = "0123456789012"
|
||||
self.key_rotation_status = False
|
||||
self.deletion_date = None
|
||||
self.tags = {}
|
||||
|
||||
@property
|
||||
def physical_resource_id(self):
|
||||
@ -35,7 +36,7 @@ class Key(BaseModel):
|
||||
"KeyMetadata": {
|
||||
"AWSAccountId": self.account_id,
|
||||
"Arn": self.arn,
|
||||
"CreationDate": "2015-01-01 00:00:00",
|
||||
"CreationDate": datetime.strftime(datetime.utcnow(), "%Y-%m-%dT%H:%M:%SZ"),
|
||||
"Description": self.description,
|
||||
"Enabled": self.enabled,
|
||||
"KeyId": self.id,
|
||||
@ -84,6 +85,19 @@ class KmsBackend(BaseBackend):
|
||||
self.keys[key.id] = key
|
||||
return key
|
||||
|
||||
def update_key_description(self, key_id, description):
|
||||
key = self.keys[self.get_key_id(key_id)]
|
||||
key.description = description
|
||||
|
||||
def tag_resource(self, key_id, tags):
|
||||
key = self.keys[self.get_key_id(key_id)]
|
||||
key.tags = tags
|
||||
|
||||
|
||||
def list_resource_tags(self, key_id):
|
||||
key = self.keys[self.get_key_id(key_id)]
|
||||
return key.tags
|
||||
|
||||
def delete_key(self, key_id):
|
||||
if key_id in self.keys:
|
||||
if key_id in self.key_to_aliases:
|
||||
|
||||
@ -38,6 +38,28 @@ class KmsResponse(BaseResponse):
|
||||
policy, key_usage, description, self.region)
|
||||
return json.dumps(key.to_dict())
|
||||
|
||||
def update_key_description(self):
|
||||
key_id = self.parameters.get('KeyId')
|
||||
description = self.parameters.get('Description')
|
||||
|
||||
self.kms_backend.update_key_description(key_id,description)
|
||||
return json.dumps(None)
|
||||
|
||||
def tag_resource(self):
|
||||
key_id = self.parameters.get('KeyId')
|
||||
tags = self.parameters.get('Tags')
|
||||
self.kms_backend.tag_resource(key_id,tags)
|
||||
return json.dumps({})
|
||||
|
||||
def list_resource_tags(self):
|
||||
key_id = self.parameters.get('KeyId')
|
||||
tags = self.kms_backend.list_resource_tags(key_id)
|
||||
return json.dumps({
|
||||
"Tags": tags,
|
||||
"NextMarker": None,
|
||||
"Truncated": False,
|
||||
})
|
||||
|
||||
def describe_key(self):
|
||||
key_id = self.parameters.get('KeyId')
|
||||
try:
|
||||
|
||||
@ -819,3 +819,37 @@ def get_function_policy():
|
||||
assert isinstance(response['Policy'], str)
|
||||
res = json.loads(response['Policy'])
|
||||
assert res['Statement'][0]['Action'] == 'lambda:InvokeFunction'
|
||||
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
def test_list_versions_by_function():
|
||||
s3_conn = boto3.client('s3', 'us-west-2')
|
||||
s3_conn.create_bucket(Bucket='test-bucket')
|
||||
|
||||
zip_content = get_test_zip_file2()
|
||||
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
conn.create_function(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.lambda_handler',
|
||||
Code={
|
||||
'S3Bucket': 'test-bucket',
|
||||
'S3Key': 'test.zip',
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
)
|
||||
|
||||
conn.publish_version(FunctionName='testFunction')
|
||||
|
||||
versions = conn.list_versions_by_function(FunctionName='testFunction')
|
||||
|
||||
assert versions['Versions'][0]['FunctionArn'] == 'arn:aws:lambda:us-west-2:123456789012:function:testFunction:$LATEST'
|
||||
|
||||
|
||||
@ -717,3 +717,60 @@ def test_cancel_key_deletion():
|
||||
assert result["KeyMetadata"]["Enabled"] == False
|
||||
assert result["KeyMetadata"]["KeyState"] == 'Disabled'
|
||||
assert 'DeletionDate' not in result["KeyMetadata"]
|
||||
|
||||
|
||||
@mock_kms
|
||||
def test_update_key_description():
|
||||
client = boto3.client('kms', region_name='us-east-1')
|
||||
key = client.create_key(Description='old_description')
|
||||
key_id = key['KeyMetadata']['KeyId']
|
||||
|
||||
result = client.update_key_description(KeyId=key_id, Description='new_description')
|
||||
assert 'ResponseMetadata' in result
|
||||
|
||||
|
||||
@mock_kms
|
||||
def test_tag_resource():
|
||||
client = boto3.client('kms', region_name='us-east-1')
|
||||
key = client.create_key(Description='cancel-key-deletion')
|
||||
response = client.schedule_key_deletion(
|
||||
KeyId=key['KeyMetadata']['KeyId']
|
||||
)
|
||||
|
||||
keyid = response['KeyId']
|
||||
response = client.tag_resource(
|
||||
KeyId=keyid,
|
||||
Tags=[
|
||||
{
|
||||
'TagKey': 'string',
|
||||
'TagValue': 'string'
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
# Shouldn't have any data, just header
|
||||
assert len(response.keys()) == 1
|
||||
|
||||
|
||||
@mock_kms
|
||||
def test_list_resource_tags():
|
||||
client = boto3.client('kms', region_name='us-east-1')
|
||||
key = client.create_key(Description='cancel-key-deletion')
|
||||
response = client.schedule_key_deletion(
|
||||
KeyId=key['KeyMetadata']['KeyId']
|
||||
)
|
||||
|
||||
keyid = response['KeyId']
|
||||
response = client.tag_resource(
|
||||
KeyId=keyid,
|
||||
Tags=[
|
||||
{
|
||||
'TagKey': 'string',
|
||||
'TagValue': 'string'
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
response = client.list_resource_tags(KeyId=keyid)
|
||||
assert response['Tags'][0]['TagKey'] == 'string'
|
||||
assert response['Tags'][0]['TagValue'] == 'string'
|
||||
|
||||
Loading…
Reference in New Issue
Block a user