Lambda: create list_aliases functionality (#6455)
This commit is contained in:
parent
3e11578a72
commit
bf3c9768b2
@ -34,6 +34,13 @@ class PreconditionFailedException(JsonRESTError):
|
|||||||
super().__init__("PreconditionFailedException", message)
|
super().__init__("PreconditionFailedException", message)
|
||||||
|
|
||||||
|
|
||||||
|
class ConflictException(LambdaClientError):
|
||||||
|
code = 409
|
||||||
|
|
||||||
|
def __init__(self, message: str):
|
||||||
|
super().__init__("ConflictException", message)
|
||||||
|
|
||||||
|
|
||||||
class UnknownAliasException(LambdaClientError):
|
class UnknownAliasException(LambdaClientError):
|
||||||
code = 404
|
code = 404
|
||||||
|
|
||||||
|
@ -36,6 +36,7 @@ from moto.ecr.models import ecr_backends
|
|||||||
from moto.s3.exceptions import MissingBucket, MissingKey
|
from moto.s3.exceptions import MissingBucket, MissingKey
|
||||||
from moto import settings
|
from moto import settings
|
||||||
from .exceptions import (
|
from .exceptions import (
|
||||||
|
ConflictException,
|
||||||
CrossAccountNotAllowed,
|
CrossAccountNotAllowed,
|
||||||
FunctionUrlConfigNotFound,
|
FunctionUrlConfigNotFound,
|
||||||
InvalidRoleFormat,
|
InvalidRoleFormat,
|
||||||
@ -514,8 +515,6 @@ class LambdaFunction(CloudFormationModel, DockerModel):
|
|||||||
|
|
||||||
self.tags = spec.get("Tags") or dict()
|
self.tags = spec.get("Tags") or dict()
|
||||||
|
|
||||||
self._aliases: Dict[str, LambdaAlias] = dict()
|
|
||||||
|
|
||||||
def __getstate__(self) -> Dict[str, Any]:
|
def __getstate__(self) -> Dict[str, Any]:
|
||||||
return {
|
return {
|
||||||
k: v
|
k: v
|
||||||
@ -954,43 +953,6 @@ class LambdaFunction(CloudFormationModel, DockerModel):
|
|||||||
def delete(self, account_id: str, region: str) -> None:
|
def delete(self, account_id: str, region: str) -> None:
|
||||||
lambda_backends[account_id][region].delete_function(self.function_name)
|
lambda_backends[account_id][region].delete_function(self.function_name)
|
||||||
|
|
||||||
def delete_alias(self, name: str) -> None:
|
|
||||||
self._aliases.pop(name, None)
|
|
||||||
|
|
||||||
def get_alias(self, name: str) -> LambdaAlias:
|
|
||||||
if name in self._aliases:
|
|
||||||
return self._aliases[name]
|
|
||||||
arn = f"arn:aws:lambda:{self.region}:{self.account_id}:function:{self.function_name}:{name}"
|
|
||||||
raise UnknownAliasException(arn)
|
|
||||||
|
|
||||||
def has_alias(self, alias_name: str) -> bool:
|
|
||||||
try:
|
|
||||||
return self.get_alias(alias_name) is not None
|
|
||||||
except UnknownAliasException:
|
|
||||||
return False
|
|
||||||
|
|
||||||
def put_alias(
|
|
||||||
self, name: str, description: str, function_version: str, routing_config: str
|
|
||||||
) -> LambdaAlias:
|
|
||||||
alias = LambdaAlias(
|
|
||||||
account_id=self.account_id,
|
|
||||||
region=self.region,
|
|
||||||
name=name,
|
|
||||||
function_name=self.function_name,
|
|
||||||
function_version=function_version,
|
|
||||||
description=description,
|
|
||||||
routing_config=routing_config,
|
|
||||||
)
|
|
||||||
self._aliases[name] = alias
|
|
||||||
return alias
|
|
||||||
|
|
||||||
def update_alias(
|
|
||||||
self, name: str, description: str, function_version: str, routing_config: str
|
|
||||||
) -> LambdaAlias:
|
|
||||||
alias = self.get_alias(name)
|
|
||||||
alias.update(description, function_version, routing_config)
|
|
||||||
return alias
|
|
||||||
|
|
||||||
def create_url_config(self, config: Dict[str, Any]) -> "FunctionUrlConfig":
|
def create_url_config(self, config: Dict[str, Any]) -> "FunctionUrlConfig":
|
||||||
self.url_config = FunctionUrlConfig(function=self, config=config)
|
self.url_config = FunctionUrlConfig(function=self, config=config)
|
||||||
return self.url_config # type: ignore[return-value]
|
return self.url_config # type: ignore[return-value]
|
||||||
@ -1205,22 +1167,34 @@ class LambdaStorage(object):
|
|||||||
self.region_name = region_name
|
self.region_name = region_name
|
||||||
self.account_id = account_id
|
self.account_id = account_id
|
||||||
|
|
||||||
|
# function-arn -> alias -> LambdaAlias
|
||||||
|
self._aliases: Dict[str, Dict[str, LambdaAlias]] = defaultdict(lambda: {})
|
||||||
|
|
||||||
def _get_latest(self, name: str) -> LambdaFunction:
|
def _get_latest(self, name: str) -> LambdaFunction:
|
||||||
return self._functions[name]["latest"]
|
return self._functions[name]["latest"]
|
||||||
|
|
||||||
def _get_version(self, name: str, version: str) -> Optional[LambdaFunction]:
|
def _get_version(self, name: str, version: str) -> Optional[LambdaFunction]:
|
||||||
for config in self._functions[name]["versions"]:
|
for config in self._functions[name]["versions"]:
|
||||||
if str(config.version) == version or config.has_alias(version):
|
if str(config.version) == version:
|
||||||
return config
|
return config
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def delete_alias(self, name: str, function_name: str) -> None:
|
def _get_function_aliases(self, function_name: str) -> Dict[str, LambdaAlias]:
|
||||||
fn = self.get_function_by_name_or_arn(function_name)
|
fn = self.get_function_by_name_or_arn(function_name)
|
||||||
return fn.delete_alias(name)
|
return self._aliases[fn.function_arn]
|
||||||
|
|
||||||
|
def delete_alias(self, name: str, function_name: str) -> None:
|
||||||
|
aliases = self._get_function_aliases(function_name)
|
||||||
|
aliases.pop(name, None)
|
||||||
|
|
||||||
def get_alias(self, name: str, function_name: str) -> LambdaAlias:
|
def get_alias(self, name: str, function_name: str) -> LambdaAlias:
|
||||||
fn = self.get_function_by_name_or_arn(function_name)
|
aliases = self._get_function_aliases(function_name)
|
||||||
return fn.get_alias(name)
|
if name in aliases:
|
||||||
|
return aliases[name]
|
||||||
|
|
||||||
|
arn = f"arn:aws:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
|
||||||
|
raise UnknownAliasException(arn)
|
||||||
|
|
||||||
def put_alias(
|
def put_alias(
|
||||||
self,
|
self,
|
||||||
@ -1230,8 +1204,23 @@ class LambdaStorage(object):
|
|||||||
description: str,
|
description: str,
|
||||||
routing_config: str,
|
routing_config: str,
|
||||||
) -> LambdaAlias:
|
) -> LambdaAlias:
|
||||||
fn = self.get_function_by_name_or_arn(function_name)
|
fn = self.get_function_by_name_or_arn(function_name, function_version)
|
||||||
return fn.put_alias(name, description, function_version, routing_config)
|
aliases = self._get_function_aliases(function_name)
|
||||||
|
if name in aliases:
|
||||||
|
arn = f"arn:aws:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
|
||||||
|
raise ConflictException(f"Alias already exists: {arn}")
|
||||||
|
|
||||||
|
alias = LambdaAlias(
|
||||||
|
account_id=self.account_id,
|
||||||
|
region=self.region_name,
|
||||||
|
name=name,
|
||||||
|
function_name=fn.function_name,
|
||||||
|
function_version=function_version,
|
||||||
|
description=description,
|
||||||
|
routing_config=routing_config,
|
||||||
|
)
|
||||||
|
aliases[name] = alias
|
||||||
|
return alias
|
||||||
|
|
||||||
def update_alias(
|
def update_alias(
|
||||||
self,
|
self,
|
||||||
@ -1241,8 +1230,13 @@ class LambdaStorage(object):
|
|||||||
description: str,
|
description: str,
|
||||||
routing_config: str,
|
routing_config: str,
|
||||||
) -> LambdaAlias:
|
) -> LambdaAlias:
|
||||||
fn = self.get_function_by_name_or_arn(function_name)
|
alias = self.get_alias(name, function_name)
|
||||||
return fn.update_alias(name, description, function_version, routing_config)
|
|
||||||
|
# errors if new function version doesn't exist
|
||||||
|
self.get_function_by_name_or_arn(function_name, function_version)
|
||||||
|
|
||||||
|
alias.update(description, function_version, routing_config)
|
||||||
|
return alias
|
||||||
|
|
||||||
def get_function_by_name(
|
def get_function_by_name(
|
||||||
self, name: str, qualifier: Optional[str] = None
|
self, name: str, qualifier: Optional[str] = None
|
||||||
@ -1256,7 +1250,17 @@ class LambdaStorage(object):
|
|||||||
if qualifier.lower() == "$latest":
|
if qualifier.lower() == "$latest":
|
||||||
return self._functions[name]["latest"]
|
return self._functions[name]["latest"]
|
||||||
|
|
||||||
return self._get_version(name, qualifier)
|
found_version = self._get_version(name, qualifier)
|
||||||
|
if found_version:
|
||||||
|
return found_version
|
||||||
|
|
||||||
|
aliases = self._get_function_aliases(name)
|
||||||
|
|
||||||
|
if qualifier in aliases:
|
||||||
|
alias = aliases[qualifier]
|
||||||
|
return self._get_version(name, alias.function_version)
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
def list_versions_by_function(self, name: str) -> Iterable[LambdaFunction]:
|
def list_versions_by_function(self, name: str) -> Iterable[LambdaFunction]:
|
||||||
if name not in self._functions:
|
if name not in self._functions:
|
||||||
@ -1266,6 +1270,10 @@ class LambdaStorage(object):
|
|||||||
latest.function_arn += ":$LATEST"
|
latest.function_arn += ":$LATEST"
|
||||||
return [latest] + self._functions[name]["versions"]
|
return [latest] + self._functions[name]["versions"]
|
||||||
|
|
||||||
|
def list_aliases(self, function_name: str) -> Iterable[LambdaAlias]:
|
||||||
|
aliases = self._get_function_aliases(function_name)
|
||||||
|
return sorted(aliases.values(), key=lambda alias: alias.name)
|
||||||
|
|
||||||
def get_arn(self, arn: str) -> Optional[LambdaFunction]:
|
def get_arn(self, arn: str) -> Optional[LambdaFunction]:
|
||||||
# Function ARN may contain an alias
|
# Function ARN may contain an alias
|
||||||
# arn:aws:lambda:region:account_id:function:<fn_name>:<alias_name>
|
# arn:aws:lambda:region:account_id:function:<fn_name>:<alias_name>
|
||||||
@ -1375,6 +1383,8 @@ class LambdaStorage(object):
|
|||||||
):
|
):
|
||||||
del self._functions[name]
|
del self._functions[name]
|
||||||
|
|
||||||
|
self._aliases[function.function_arn] = {}
|
||||||
|
|
||||||
def all(self) -> Iterable[LambdaFunction]:
|
def all(self) -> Iterable[LambdaFunction]:
|
||||||
result = []
|
result = []
|
||||||
|
|
||||||
@ -1706,6 +1716,9 @@ class LambdaBackend(BaseBackend):
|
|||||||
def list_versions_by_function(self, function_name: str) -> Iterable[LambdaFunction]:
|
def list_versions_by_function(self, function_name: str) -> Iterable[LambdaFunction]:
|
||||||
return self._lambdas.list_versions_by_function(function_name)
|
return self._lambdas.list_versions_by_function(function_name)
|
||||||
|
|
||||||
|
def list_aliases(self, function_name: str) -> Iterable[LambdaAlias]:
|
||||||
|
return self._lambdas.list_aliases(function_name)
|
||||||
|
|
||||||
def get_event_source_mapping(self, uuid: str) -> Optional[EventSourceMapping]:
|
def get_event_source_mapping(self, uuid: str) -> Optional[EventSourceMapping]:
|
||||||
return self._event_source_mappings.get(uuid)
|
return self._event_source_mappings.get(uuid)
|
||||||
|
|
||||||
|
@ -46,8 +46,13 @@ class LambdaResponse(BaseResponse):
|
|||||||
|
|
||||||
def aliases(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
|
def aliases(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
|
||||||
self.setup_class(request, full_url, headers)
|
self.setup_class(request, full_url, headers)
|
||||||
|
|
||||||
if request.method == "POST":
|
if request.method == "POST":
|
||||||
return self._create_alias()
|
return self._create_alias()
|
||||||
|
elif request.method == "GET":
|
||||||
|
path = request.path if hasattr(request, "path") else path_url(request.url)
|
||||||
|
function_name = path.split("/")[-2]
|
||||||
|
return self._list_aliases(function_name)
|
||||||
|
|
||||||
def alias(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
|
def alias(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
|
||||||
self.setup_class(request, full_url, headers)
|
self.setup_class(request, full_url, headers)
|
||||||
@ -296,6 +301,16 @@ class LambdaResponse(BaseResponse):
|
|||||||
|
|
||||||
return 200, {}, json.dumps(result)
|
return 200, {}, json.dumps(result)
|
||||||
|
|
||||||
|
def _list_aliases(self, function_name: str) -> TYPE_RESPONSE:
|
||||||
|
result: Dict[str, Any] = {"Aliases": []}
|
||||||
|
|
||||||
|
aliases = self.backend.list_aliases(function_name)
|
||||||
|
for alias in aliases:
|
||||||
|
json_data = alias.to_json()
|
||||||
|
result["Aliases"].append(json_data)
|
||||||
|
|
||||||
|
return 200, {}, json.dumps(result)
|
||||||
|
|
||||||
def _create_function(self) -> TYPE_RESPONSE:
|
def _create_function(self) -> TYPE_RESPONSE:
|
||||||
fn = self.backend.create_function(self.json_body)
|
fn = self.backend.create_function(self.json_body)
|
||||||
config = fn.get_configuration(on_create=True)
|
config = fn.get_configuration(on_create=True)
|
||||||
|
@ -958,6 +958,107 @@ def test_list_versions_by_function():
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@mock_lambda
|
||||||
|
@mock_s3
|
||||||
|
def test_list_aliases():
|
||||||
|
bucket_name = str(uuid4())
|
||||||
|
s3_conn = boto3.client("s3", _lambda_region)
|
||||||
|
s3_conn.create_bucket(
|
||||||
|
Bucket=bucket_name,
|
||||||
|
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
|
||||||
|
)
|
||||||
|
|
||||||
|
zip_content = get_test_zip_file2()
|
||||||
|
s3_conn.put_object(Bucket=bucket_name, Key="test.zip", Body=zip_content)
|
||||||
|
conn = boto3.client("lambda", _lambda_region)
|
||||||
|
function_name = str(uuid4())[0:6]
|
||||||
|
function_name2 = str(uuid4())[0:6]
|
||||||
|
|
||||||
|
conn.create_function(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Runtime="python2.7",
|
||||||
|
Role=get_role_name(),
|
||||||
|
Handler="lambda_function.lambda_handler",
|
||||||
|
Code={"S3Bucket": bucket_name, "S3Key": "test.zip"},
|
||||||
|
Description="test lambda function",
|
||||||
|
Timeout=3,
|
||||||
|
MemorySize=128,
|
||||||
|
Publish=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
conn.create_function(
|
||||||
|
FunctionName=function_name2,
|
||||||
|
Runtime="python2.7",
|
||||||
|
Role=get_role_name(),
|
||||||
|
Handler="lambda_function.lambda_handler",
|
||||||
|
Code={"S3Bucket": bucket_name, "S3Key": "test.zip"},
|
||||||
|
Description="test lambda function",
|
||||||
|
Timeout=3,
|
||||||
|
MemorySize=128,
|
||||||
|
Publish=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
first_version = conn.publish_version(FunctionName=function_name)["Version"]
|
||||||
|
|
||||||
|
conn.create_alias(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Name="alias1",
|
||||||
|
FunctionVersion=first_version,
|
||||||
|
)
|
||||||
|
|
||||||
|
conn.update_function_code(FunctionName=function_name, ZipFile=get_test_zip_file1())
|
||||||
|
second_version = conn.publish_version(FunctionName=function_name)["Version"]
|
||||||
|
|
||||||
|
conn.create_alias(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Name="alias2",
|
||||||
|
FunctionVersion=second_version,
|
||||||
|
)
|
||||||
|
|
||||||
|
conn.create_alias(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Name="alias0",
|
||||||
|
FunctionVersion=second_version,
|
||||||
|
)
|
||||||
|
|
||||||
|
aliases = conn.list_aliases(FunctionName=function_name)
|
||||||
|
assert len(aliases["Aliases"]) == 3
|
||||||
|
|
||||||
|
# should be ordered by their alias name (as per SDK response)
|
||||||
|
assert (
|
||||||
|
aliases["Aliases"][0]["AliasArn"]
|
||||||
|
== f"arn:aws:lambda:us-west-2:{ACCOUNT_ID}:function:{function_name}:alias0"
|
||||||
|
)
|
||||||
|
assert aliases["Aliases"][0]["FunctionVersion"] == second_version
|
||||||
|
|
||||||
|
assert (
|
||||||
|
aliases["Aliases"][1]["AliasArn"]
|
||||||
|
== f"arn:aws:lambda:us-west-2:{ACCOUNT_ID}:function:{function_name}:alias1"
|
||||||
|
)
|
||||||
|
assert aliases["Aliases"][1]["FunctionVersion"] == first_version
|
||||||
|
|
||||||
|
assert (
|
||||||
|
aliases["Aliases"][2]["AliasArn"]
|
||||||
|
== f"arn:aws:lambda:us-west-2:{ACCOUNT_ID}:function:{function_name}:alias2"
|
||||||
|
)
|
||||||
|
assert aliases["Aliases"][2]["FunctionVersion"] == second_version
|
||||||
|
|
||||||
|
res = conn.publish_version(FunctionName=function_name2)
|
||||||
|
conn.create_alias(
|
||||||
|
FunctionName=function_name2,
|
||||||
|
Name="alias1",
|
||||||
|
FunctionVersion=res["Version"],
|
||||||
|
)
|
||||||
|
|
||||||
|
aliases = conn.list_aliases(FunctionName=function_name2)
|
||||||
|
|
||||||
|
assert len(aliases["Aliases"]) == 1
|
||||||
|
assert (
|
||||||
|
aliases["Aliases"][0]["AliasArn"]
|
||||||
|
== f"arn:aws:lambda:us-west-2:{ACCOUNT_ID}:function:{function_name2}:alias1"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@mock_lambda
|
@mock_lambda
|
||||||
@mock_s3
|
@mock_s3
|
||||||
def test_create_function_with_already_exists():
|
def test_create_function_with_already_exists():
|
||||||
|
@ -10,6 +10,7 @@ from uuid import uuid4
|
|||||||
from .utilities import (
|
from .utilities import (
|
||||||
get_role_name,
|
get_role_name,
|
||||||
get_test_zip_file1,
|
get_test_zip_file1,
|
||||||
|
get_test_zip_file2,
|
||||||
)
|
)
|
||||||
|
|
||||||
# See our Development Tips on writing tests for hints on how to write good tests:
|
# See our Development Tips on writing tests for hints on how to write good tests:
|
||||||
@ -150,6 +151,56 @@ def test_get_alias():
|
|||||||
assert "RevisionId" in resp
|
assert "RevisionId" in resp
|
||||||
|
|
||||||
|
|
||||||
|
@mock_lambda
|
||||||
|
def test_aliases_are_unique_per_function():
|
||||||
|
client = boto3.client("lambda", region_name="us-west-1")
|
||||||
|
function_name = str(uuid4())[0:6]
|
||||||
|
function_name2 = str(uuid4())[0:6]
|
||||||
|
|
||||||
|
client.create_function(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Runtime="python3.7",
|
||||||
|
Role=get_role_name(),
|
||||||
|
Handler="lambda_function.lambda_handler",
|
||||||
|
Code={"ZipFile": get_test_zip_file1()},
|
||||||
|
)
|
||||||
|
client.create_function(
|
||||||
|
FunctionName=function_name2,
|
||||||
|
Runtime="python3.7",
|
||||||
|
Role=get_role_name(),
|
||||||
|
Handler="lambda_function.lambda_handler",
|
||||||
|
Code={"ZipFile": get_test_zip_file1()},
|
||||||
|
)
|
||||||
|
|
||||||
|
client.create_alias(
|
||||||
|
FunctionName=function_name, Name="alias1", FunctionVersion="$LATEST"
|
||||||
|
)
|
||||||
|
client.create_alias(
|
||||||
|
FunctionName=function_name2, Name="alias1", FunctionVersion="$LATEST"
|
||||||
|
)
|
||||||
|
|
||||||
|
client.update_function_code(
|
||||||
|
FunctionName=function_name, ZipFile=get_test_zip_file2()
|
||||||
|
)
|
||||||
|
client.update_function_code(
|
||||||
|
FunctionName=function_name2, ZipFile=get_test_zip_file2()
|
||||||
|
)
|
||||||
|
|
||||||
|
res = client.publish_version(FunctionName=function_name)
|
||||||
|
|
||||||
|
with pytest.raises(ClientError) as exc:
|
||||||
|
client.create_alias(
|
||||||
|
FunctionName=function_name, Name="alias1", FunctionVersion=res["Version"]
|
||||||
|
)
|
||||||
|
|
||||||
|
err = exc.value.response["Error"]
|
||||||
|
assert err["Code"] == "ConflictException"
|
||||||
|
assert (
|
||||||
|
err["Message"]
|
||||||
|
== f"Alias already exists: arn:aws:lambda:us-west-1:{ACCOUNT_ID}:function:{function_name}:alias1"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@mock_lambda
|
@mock_lambda
|
||||||
def test_get_alias_using_function_arn():
|
def test_get_alias_using_function_arn():
|
||||||
client = boto3.client("lambda", region_name="us-west-1")
|
client = boto3.client("lambda", region_name="us-west-1")
|
||||||
@ -250,10 +301,16 @@ def test_update_alias():
|
|||||||
FunctionName=function_name, Name="alias1", FunctionVersion="$LATEST"
|
FunctionName=function_name, Name="alias1", FunctionVersion="$LATEST"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
client.update_function_code(
|
||||||
|
FunctionName=function_name, ZipFile=get_test_zip_file2()
|
||||||
|
)
|
||||||
|
|
||||||
|
new_version = client.publish_version(FunctionName=function_name)["Version"]
|
||||||
|
|
||||||
resp = client.update_alias(
|
resp = client.update_alias(
|
||||||
FunctionName=function_name,
|
FunctionName=function_name,
|
||||||
Name="alias1",
|
Name="alias1",
|
||||||
FunctionVersion="1",
|
FunctionVersion=new_version,
|
||||||
Description="updated desc",
|
Description="updated desc",
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -262,11 +319,43 @@ def test_update_alias():
|
|||||||
== f"arn:aws:lambda:us-east-2:{ACCOUNT_ID}:function:{function_name}:alias1"
|
== f"arn:aws:lambda:us-east-2:{ACCOUNT_ID}:function:{function_name}:alias1"
|
||||||
)
|
)
|
||||||
assert resp["Name"] == "alias1"
|
assert resp["Name"] == "alias1"
|
||||||
assert resp["FunctionVersion"] == "1"
|
assert resp["FunctionVersion"] == new_version
|
||||||
assert resp["Description"] == "updated desc"
|
assert resp["Description"] == "updated desc"
|
||||||
assert "RevisionId" in resp
|
assert "RevisionId" in resp
|
||||||
|
|
||||||
|
|
||||||
|
@mock_lambda
|
||||||
|
def test_update_alias_errors_if_version_doesnt_exist():
|
||||||
|
client = boto3.client("lambda", region_name="us-east-2")
|
||||||
|
function_name = str(uuid4())[0:6]
|
||||||
|
|
||||||
|
client.create_function(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Runtime="python3.7",
|
||||||
|
Role=get_role_name(),
|
||||||
|
Handler="lambda_function.lambda_handler",
|
||||||
|
Code={"ZipFile": get_test_zip_file1()},
|
||||||
|
)
|
||||||
|
|
||||||
|
client.create_alias(
|
||||||
|
FunctionName=function_name, Name="alias1", FunctionVersion="$LATEST"
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(ClientError) as exc:
|
||||||
|
client.update_alias(
|
||||||
|
FunctionName=function_name,
|
||||||
|
Name="alias1",
|
||||||
|
FunctionVersion="1",
|
||||||
|
Description="updated desc",
|
||||||
|
)
|
||||||
|
err = exc.value.response["Error"]
|
||||||
|
assert err["Code"] == "ResourceNotFoundException"
|
||||||
|
assert (
|
||||||
|
err["Message"]
|
||||||
|
== f"Function not found: arn:aws:lambda:us-east-2:{ACCOUNT_ID}:function:{function_name}:1"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@mock_lambda
|
@mock_lambda
|
||||||
def test_update_alias_routingconfig():
|
def test_update_alias_routingconfig():
|
||||||
client = boto3.client("lambda", region_name="us-east-2")
|
client = boto3.client("lambda", region_name="us-east-2")
|
||||||
|
Loading…
Reference in New Issue
Block a user