Add get_function_configuration support for Lambda (#3562)
* Add get_function_configuration support for Lambda * remove unnesecary code from test and use _lambda_region when asserting * rename function and skip coping configuration * run black formatting
This commit is contained in:
parent
640df04840
commit
f749f583ee
@ -131,6 +131,8 @@ class LambdaResponse(BaseResponse):
|
||||
self.setup_class(request, full_url, headers)
|
||||
if request.method == "PUT":
|
||||
return self._put_configuration(request)
|
||||
if request.method == "GET":
|
||||
return self._get_function_configuration(request, full_url, headers)
|
||||
else:
|
||||
raise ValueError("Cannot handle request")
|
||||
|
||||
@ -300,6 +302,14 @@ class LambdaResponse(BaseResponse):
|
||||
else:
|
||||
return 404, {}, "{}"
|
||||
|
||||
@staticmethod
|
||||
def _set_configuration_qualifier(configuration, qualifier):
|
||||
if qualifier is None or qualifier == "$LATEST":
|
||||
configuration["Version"] = "$LATEST"
|
||||
if qualifier == "$LATEST":
|
||||
configuration["FunctionArn"] += ":$LATEST"
|
||||
return configuration
|
||||
|
||||
def _get_function(self, request, full_url, headers):
|
||||
function_name = unquote(self.path.rsplit("/", 1)[-1])
|
||||
qualifier = self._get_param("Qualifier", None)
|
||||
@ -308,14 +318,27 @@ class LambdaResponse(BaseResponse):
|
||||
|
||||
if fn:
|
||||
code = fn.get_code()
|
||||
if qualifier is None or qualifier == "$LATEST":
|
||||
code["Configuration"]["Version"] = "$LATEST"
|
||||
if qualifier == "$LATEST":
|
||||
code["Configuration"]["FunctionArn"] += ":$LATEST"
|
||||
code["Configuration"] = self._set_configuration_qualifier(
|
||||
code["Configuration"], qualifier
|
||||
)
|
||||
return 200, {}, json.dumps(code)
|
||||
else:
|
||||
return 404, {"x-amzn-ErrorType": "ResourceNotFoundException"}, "{}"
|
||||
|
||||
def _get_function_configuration(self, request, full_url, headers):
|
||||
function_name = unquote(self.path.rsplit("/", 2)[-2])
|
||||
qualifier = self._get_param("Qualifier", None)
|
||||
|
||||
fn = self.lambda_backend.get_function(function_name, qualifier)
|
||||
|
||||
if fn:
|
||||
configuration = self._set_configuration_qualifier(
|
||||
fn.get_configuration(), qualifier
|
||||
)
|
||||
return 200, {}, json.dumps(configuration)
|
||||
else:
|
||||
return 404, {"x-amzn-ErrorType": "ResourceNotFoundException"}, "{}"
|
||||
|
||||
def _get_aws_region(self, full_url):
|
||||
region = self.region_regex.search(full_url)
|
||||
if region:
|
||||
|
@ -514,6 +514,67 @@ def test_get_function():
|
||||
conn.get_function(FunctionName="junk", Qualifier="$LATEST")
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
@freeze_time("2015-01-01 00:00:00")
|
||||
def test_get_function_configuration():
|
||||
s3_conn = boto3.client("s3", _lambda_region)
|
||||
s3_conn.create_bucket(
|
||||
Bucket="test-bucket",
|
||||
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
|
||||
)
|
||||
|
||||
zip_content = get_test_zip_file1()
|
||||
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
|
||||
conn = boto3.client("lambda", _lambda_region)
|
||||
|
||||
conn.create_function(
|
||||
FunctionName="testFunction",
|
||||
Runtime="python2.7",
|
||||
Role=get_role_name(),
|
||||
Handler="lambda_function.lambda_handler",
|
||||
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
|
||||
Description="test lambda function",
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
Environment={"Variables": {"test_variable": "test_value"}},
|
||||
)
|
||||
|
||||
result = conn.get_function_configuration(FunctionName="testFunction")
|
||||
|
||||
result["CodeSha256"].should.equal(hashlib.sha256(zip_content).hexdigest())
|
||||
result["CodeSize"].should.equal(len(zip_content))
|
||||
result["Description"].should.equal("test lambda function")
|
||||
result.should.contain("FunctionArn")
|
||||
result["FunctionName"].should.equal("testFunction")
|
||||
result["Handler"].should.equal("lambda_function.lambda_handler")
|
||||
result["MemorySize"].should.equal(128)
|
||||
result["Role"].should.equal(get_role_name())
|
||||
result["Runtime"].should.equal("python2.7")
|
||||
result["Timeout"].should.equal(3)
|
||||
result["Version"].should.equal("$LATEST")
|
||||
result.should.contain("VpcConfig")
|
||||
result.should.contain("Environment")
|
||||
result["Environment"].should.contain("Variables")
|
||||
result["Environment"]["Variables"].should.equal({"test_variable": "test_value"})
|
||||
|
||||
# Test get function with qualifier
|
||||
result = conn.get_function_configuration(
|
||||
FunctionName="testFunction", Qualifier="$LATEST"
|
||||
)
|
||||
result["Version"].should.equal("$LATEST")
|
||||
result["FunctionArn"].should.equal(
|
||||
"arn:aws:lambda:{}:{}:function:testFunction:$LATEST".format(
|
||||
_lambda_region, ACCOUNT_ID
|
||||
)
|
||||
)
|
||||
|
||||
# Test get function when can't find function name
|
||||
with pytest.raises(conn.exceptions.ResourceNotFoundException):
|
||||
conn.get_function_configuration(FunctionName="junk", Qualifier="$LATEST")
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
def test_get_function_by_arn():
|
||||
|
Loading…
Reference in New Issue
Block a user