AWSLambda - Improve retrieval of qualifiers (#5576)

moto/awslambda/models.py

@@ -895,6 +895,12 @@ class LambdaFunction(CloudFormationModel, DockerModel):
         arn = f"arn:aws:lambda:{self.region}:{self.account_id}:function:{self.function_name}:{name}"
         raise UnknownAliasException(arn)
 
+    def has_alias(self, alias_name) -> bool:
+        try:
+            return self.get_alias(alias_name) is not None
+        except UnknownAliasException:
+            return False
+
     def put_alias(self, name, description, function_version, routing_config):
         alias = LambdaAlias(
             account_id=self.account_id,
@@ -1115,7 +1121,6 @@ class LambdaStorage(object):
     def __init__(self, region_name, account_id):
         # Format 'func_name' {'versions': []}
         self._functions = {}
-        self._aliases = dict()
         self._arns = weakref.WeakValueDictionary()
         self.region_name = region_name
         self.account_id = account_id
@@ -1123,13 +1128,11 @@ class LambdaStorage(object):
     def _get_latest(self, name):
         return self._functions[name]["latest"]
 
-    def _get_version(self, name, version):
+    def _get_version(self, name: str, version: str):
-        index = version - 1
+        for config in self._functions[name]["versions"]:
-
+            if str(config.version) == version or config.has_alias(version):
-        try:
+                return config
-            return self._functions[name]["versions"][index]
+        return None
-        except IndexError:
-            return None
 
     def delete_alias(self, name, function_name):
         fn = self.get_function_by_name_or_arn(function_name)
@@ -1158,11 +1161,11 @@ class LambdaStorage(object):
         if qualifier is None:
             return self._get_latest(name)
 
-        try:
+        if qualifier.lower() == "$latest":
-            return self._get_version(name, int(qualifier))
-        except ValueError:
             return self._functions[name]["latest"]
 
+        return self._get_version(name, qualifier)
+
     def list_versions_by_function(self, name):
         if name not in self._functions:
             return None

tests/test_awslambda/test_lambda.py

@@ -1100,3 +1100,37 @@ def test_remove_unknown_permission_throws_error():
     err = exc.value.response["Error"]
     err["Code"].should.equal("ResourceNotFoundException")
     err["Message"].should.equal("No policy is associated with the given resource.")
+
+
+@mock_lambda
+def test_multiple_qualifiers():
+    client = boto3.client("lambda", "us-east-1")
+
+    zip_content = get_test_zip_file1()
+    fn_name = str(uuid4())[0:6]
+    client.create_function(
+        FunctionName=fn_name,
+        Runtime="python3.7",
+        Role=(get_role_name()),
+        Handler="lambda_function.handler",
+        Code={"ZipFile": zip_content},
+    )
+
+    for _ in range(10):
+        client.publish_version(FunctionName=fn_name)
+
+    resp = client.list_versions_by_function(FunctionName=fn_name)["Versions"]
+    qualis = [fn["FunctionArn"].split(":")[-1] for fn in resp]
+    qualis.should.equal(["$LATEST", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10"])
+
+    client.delete_function(FunctionName=fn_name, Qualifier="4")
+    client.delete_function(FunctionName=fn_name, Qualifier="5")
+
+    resp = client.list_versions_by_function(FunctionName=fn_name)["Versions"]
+    qualis = [fn["FunctionArn"].split(":")[-1] for fn in resp]
+    qualis.should.equal(["$LATEST", "1", "2", "3", "6", "7", "8", "9", "10"])
+
+    fn = client.get_function(FunctionName=fn_name, Qualifier="6")["Configuration"]
+    fn["FunctionArn"].should.equal(
+        f"arn:aws:lambda:us-east-1:{ACCOUNT_ID}:function:{fn_name}:6"
+    )

tests/test_awslambda/test_lambda_alias.py

@@ -297,3 +297,26 @@ def test_update_alias_routingconfig():
     resp.should.have.key("RoutingConfig").equals(
         {"AdditionalVersionWeights": {"2": 0.5}}
     )
+
+
+@mock_lambda
+def test_get_function_using_alias():
+    client = boto3.client("lambda", region_name="us-east-2")
+    fn_name = str(uuid4())[0:6]
+
+    client.create_function(
+        FunctionName=fn_name,
+        Runtime="python3.7",
+        Role=get_role_name(),
+        Handler="lambda_function.lambda_handler",
+        Code={"ZipFile": get_test_zip_file1()},
+    )
+    client.publish_version(FunctionName=fn_name)
+    client.publish_version(FunctionName=fn_name)
+
+    client.create_alias(FunctionName=fn_name, Name="live", FunctionVersion="1")
+
+    fn = client.get_function(FunctionName=fn_name, Qualifier="live")["Configuration"]
+    fn["FunctionArn"].should.equal(
+        f"arn:aws:lambda:us-east-2:{ACCOUNT_ID}:function:{fn_name}:1"
+    )