Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

AWSLambda:list_functions() should only return the latest version by default (#4047)

Browse Source
This commit is contained in:
Bert Blommers 2021-08-28 15:26:44 +01:00 committed by GitHub
parent a43c42ef21
commit 728c0c91b4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 107 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
moto/awslambda/models.py
View File

@ -964,7 +964,7 @@ class LambdaStorage(object):
fn.policy = Policy(fn)
self._arns[fn.function_arn] = fn
def publish_function(self, name_or_arn):
def publish_function(self, name_or_arn, description=""):
function = self.get_function_by_name_or_arn(name_or_arn)
name = function.function_name
if name not in self._functions:
@ -975,6 +975,8 @@ class LambdaStorage(object):
new_version = len(self._functions[name]["versions"]) + 1
fn = copy.copy(self._functions[name]["latest"])
fn.set_version(new_version)
if description:
fn.description = description
self._functions[name]["versions"].append(fn)
self._arns[fn.function_arn] = fn
@ -1028,13 +1030,26 @@ class LambdaStorage(object):
result = []
for function_group in self._functions.values():
if function_group["latest"] is not None:
result.append(function_group["latest"])
latest = copy.deepcopy(function_group["latest"])
latest.function_arn = "{}:$LATEST".format(latest.function_arn)
result.append(latest)
result.extend(function_group["versions"])
return result
def latest(self):
"""
Return the list of functions with version @LATEST
:return:
"""
result = []
for function_group in self._functions.values():
if function_group["latest"] is not None:
result.append(function_group["latest"])
return result
class LayerStorage(object):
def __init__(self):
@ -1091,6 +1106,9 @@ class LambdaBackend(BaseBackend):
if spec.get("Publish"):
ver = self.publish_function(function_name)
fn = copy.deepcopy(
fn
) # We don't want to change the actual version - just the return value
fn.version = ver.version
return fn
@ -1162,8 +1180,8 @@ class LambdaBackend(BaseBackend):
def layers_versions_by_arn(self, layer_version_arn):
return self._layers.get_layer_version_by_arn(layer_version_arn)
def publish_function(self, function_name):
return self._lambdas.publish_function(function_name)
def publish_function(self, function_name, description=""):
return self._lambdas.publish_function(function_name, description)
def get_function(self, function_name_or_arn, qualifier=None):
return self._lambdas.get_function_by_name_or_arn(
@ -1210,8 +1228,10 @@ class LambdaBackend(BaseBackend):
def delete_function(self, function_name, qualifier=None):
return self._lambdas.del_function(function_name, qualifier)
def list_functions(self):
return self._lambdas.all()
def list_functions(self, func_version=None):
if func_version == "ALL":
return self._lambdas.all()
return self._lambdas.latest()
def send_sqs_batch(self, function_arn, messages, queue_arn):
success = True

8
moto/awslambda/responses.py
View File

@ -236,11 +236,12 @@ class LambdaResponse(BaseResponse):
return 404, response_headers, "{}"
def _list_functions(self, request, full_url, headers):
querystring = self.querystring
func_version = querystring.get("FunctionVersion", [None])[0]
result = {"Functions": []}
for fn in self.lambda_backend.list_functions():
for fn in self.lambda_backend.list_functions(func_version):
json_data = fn.get_configuration()
json_data["Version"] = "$LATEST"
result["Functions"].append(json_data)
return 200, {}, json.dumps(result)
@ -298,8 +299,9 @@ class LambdaResponse(BaseResponse):
def _publish_function(self, request, full_url, headers):
function_name = self.path.rsplit("/", 2)[-2]
description = self._get_param("Description")
fn = self.lambda_backend.publish_function(function_name)
fn = self.lambda_backend.publish_function(function_name, description)
if fn:
config = fn.get_configuration()
return 201, {}, json.dumps(config)

88
tests/test_awslambda/test_lambda.py
View File

@ -121,8 +121,40 @@ def test_lambda_regions(region):
@mock_lambda
def test_list_functions():
conn = boto3.client("lambda", _lambda_region)
result = conn.list_functions()
result["Functions"].should.have.length_of(0)
conn.list_functions()["Functions"].should.have.length_of(0)
function_name = "testFunction"
conn.create_function(
FunctionName=function_name,
Runtime="python3.7",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file1()},
)
conn.list_functions()["Functions"].should.have.length_of(1)
conn.publish_version(FunctionName=function_name, Description="v2")
conn.list_functions()["Functions"].should.have.length_of(1)
# FunctionVersion=ALL means we should get a list of all versions
result = conn.list_functions(FunctionVersion="ALL")["Functions"]
result.should.have.length_of(2)
v1 = [f for f in result if f["Version"] == "1"][0]
v1["Description"].should.equal("v2")
v1["FunctionArn"].should.equal(
"arn:aws:lambda:{}:{}:function:{}:1".format(
_lambda_region, ACCOUNT_ID, function_name
)
)
latest = [f for f in result if f["Version"] == "$LATEST"][0]
latest["Description"].should.equal("")
latest["FunctionArn"].should.equal(
"arn:aws:lambda:{}:{}:function:{}:$LATEST".format(
_lambda_region, ACCOUNT_ID, function_name
)
)
@pytest.mark.network
@ -772,14 +804,14 @@ def test_publish():
Publish=False,
)
function_list = conn.list_functions()
function_list = conn.list_functions(FunctionVersion="ALL")
function_list["Functions"].should.have.length_of(1)
latest_arn = function_list["Functions"][0]["FunctionArn"]
res = conn.publish_version(FunctionName="testFunction")
assert res["ResponseMetadata"]["HTTPStatusCode"] == 201
function_list = conn.list_functions()
function_list = conn.list_functions(FunctionVersion="ALL")
function_list["Functions"].should.have.length_of(2)
# #SetComprehension ;-)
@ -815,8 +847,9 @@ def test_list_create_list_get_delete_list():
conn.list_functions()["Functions"].should.have.length_of(0)
function_name = "testFunction"
conn.create_function(
FunctionName="testFunction",
FunctionName=function_name,
Runtime="python2.7",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
@ -837,10 +870,7 @@ def test_list_create_list_get_delete_list():
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
"CodeSize": len(zip_content),
"Description": "test lambda function",
"FunctionArn": "arn:aws:lambda:{}:{}:function:testFunction".format(
_lambda_region, ACCOUNT_ID
),
"FunctionName": "testFunction",
"FunctionName": function_name,
"Handler": "lambda_function.lambda_handler",
"MemorySize": 128,
"Role": get_role_name(),
@ -853,16 +883,48 @@ def test_list_create_list_get_delete_list():
},
"ResponseMetadata": {"HTTPStatusCode": 200},
}
func = conn.list_functions()["Functions"][0]
func.pop("LastModified")
func.should.equal(expected_function_result["Configuration"])
functions = conn.list_functions()["Functions"]
functions.should.have.length_of(1)
functions[0]["FunctionArn"].should.equal(
"arn:aws:lambda:{}:{}:function:{}".format(
_lambda_region, ACCOUNT_ID, function_name
)
)
functions = conn.list_functions(FunctionVersion="ALL")["Functions"]
functions.should.have.length_of(2)
latest = [f for f in functions if f["Version"] == "$LATEST"][0]
latest["FunctionArn"].should.equal(
"arn:aws:lambda:{}:{}:function:{}:$LATEST".format(
_lambda_region, ACCOUNT_ID, function_name
)
)
latest.pop("FunctionArn")
latest.pop("LastModified")
latest.should.equal(expected_function_result["Configuration"])
published = [f for f in functions if f["Version"] != "$LATEST"][0]
published["Version"].should.equal("1")
published["FunctionArn"].should.equal(
"arn:aws:lambda:{}:{}:function:{}:1".format(
_lambda_region, ACCOUNT_ID, function_name
)
)
func = conn.get_function(FunctionName=function_name)
func["Configuration"]["FunctionArn"].should.equal(
"arn:aws:lambda:{}:{}:function:{}".format(
_lambda_region, ACCOUNT_ID, function_name
)
)
func = conn.get_function(FunctionName="testFunction")
# this is hard to match against, so remove it
func["ResponseMetadata"].pop("HTTPHeaders", None)
# Botocore inserts retry attempts not seen in Python27
func["ResponseMetadata"].pop("RetryAttempts", None)
func["Configuration"].pop("LastModified")
func["Configuration"].pop("FunctionArn")
func.should.equal(expected_function_result)
conn.delete_function(FunctionName="testFunction")