• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

localstack / localstack / 22048723723

13 Feb 2026 06:59PM UTC coverage: 87.006% (+0.1%) from 86.883%
22048723723

push

github

web-flow
CW Logs: Test suite for service internalization (#13692)

22 of 22 new or added lines in 1 file covered. (100.0%)

928 existing lines in 33 files now uncovered.

69716 of 80128 relevant lines covered (87.01%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.14
/localstack-core/localstack/services/lambda_/provider.py
1
import base64
1✔
2
import dataclasses
1✔
3
import datetime
1✔
4
import itertools
1✔
5
import json
1✔
6
import logging
1✔
7
import re
1✔
8
import threading
1✔
9
import time
1✔
10
from typing import IO, Any
1✔
11

12
from botocore.exceptions import ClientError
1✔
13

14
from localstack import config
1✔
15
from localstack.aws.api import RequestContext, ServiceException, handler
1✔
16
from localstack.aws.api.lambda_ import (
1✔
17
    AccountLimit,
18
    AccountUsage,
19
    AddLayerVersionPermissionResponse,
20
    AddPermissionRequest,
21
    AddPermissionResponse,
22
    Alias,
23
    AliasConfiguration,
24
    AliasRoutingConfiguration,
25
    AllowedPublishers,
26
    Architecture,
27
    Arn,
28
    Blob,
29
    BlobStream,
30
    CapacityProviderConfig,
31
    CodeSigningConfigArn,
32
    CodeSigningConfigNotFoundException,
33
    CodeSigningPolicies,
34
    CompatibleArchitectures,
35
    CompatibleRuntimes,
36
    Concurrency,
37
    Cors,
38
    CreateCodeSigningConfigResponse,
39
    CreateEventSourceMappingRequest,
40
    CreateFunctionRequest,
41
    CreateFunctionUrlConfigResponse,
42
    DeleteCodeSigningConfigResponse,
43
    DeleteFunctionResponse,
44
    Description,
45
    DestinationConfig,
46
    DurableExecutionName,
47
    EventSourceMappingConfiguration,
48
    FunctionCodeLocation,
49
    FunctionConfiguration,
50
    FunctionEventInvokeConfig,
51
    FunctionName,
52
    FunctionUrlAuthType,
53
    FunctionUrlQualifier,
54
    FunctionVersionLatestPublished,
55
    GetAccountSettingsResponse,
56
    GetCodeSigningConfigResponse,
57
    GetFunctionCodeSigningConfigResponse,
58
    GetFunctionConcurrencyResponse,
59
    GetFunctionRecursionConfigResponse,
60
    GetFunctionResponse,
61
    GetFunctionUrlConfigResponse,
62
    GetLayerVersionPolicyResponse,
63
    GetLayerVersionResponse,
64
    GetPolicyResponse,
65
    GetProvisionedConcurrencyConfigResponse,
66
    InvalidParameterValueException,
67
    InvocationResponse,
68
    InvocationType,
69
    InvokeAsyncResponse,
70
    InvokeMode,
71
    LambdaApi,
72
    LambdaManagedInstancesCapacityProviderConfig,
73
    LastUpdateStatus,
74
    LayerName,
75
    LayerPermissionAllowedAction,
76
    LayerPermissionAllowedPrincipal,
77
    LayersListItem,
78
    LayerVersionArn,
79
    LayerVersionContentInput,
80
    LayerVersionNumber,
81
    LicenseInfo,
82
    ListAliasesResponse,
83
    ListCodeSigningConfigsResponse,
84
    ListEventSourceMappingsResponse,
85
    ListFunctionEventInvokeConfigsResponse,
86
    ListFunctionsByCodeSigningConfigResponse,
87
    ListFunctionsResponse,
88
    ListFunctionUrlConfigsResponse,
89
    ListLayersResponse,
90
    ListLayerVersionsResponse,
91
    ListProvisionedConcurrencyConfigsResponse,
92
    ListTagsResponse,
93
    ListVersionsByFunctionResponse,
94
    LogFormat,
95
    LoggingConfig,
96
    LogType,
97
    MasterRegion,
98
    MaxFunctionEventInvokeConfigListItems,
99
    MaximumEventAgeInSeconds,
100
    MaximumRetryAttempts,
101
    MaxItems,
102
    MaxLayerListItems,
103
    MaxListItems,
104
    MaxProvisionedConcurrencyConfigListItems,
105
    NamespacedFunctionName,
106
    NamespacedStatementId,
107
    NumericLatestPublishedOrAliasQualifier,
108
    OnFailure,
109
    OnSuccess,
110
    OrganizationId,
111
    PackageType,
112
    PositiveInteger,
113
    PreconditionFailedException,
114
    ProvisionedConcurrencyConfigListItem,
115
    ProvisionedConcurrencyConfigNotFoundException,
116
    ProvisionedConcurrencyStatusEnum,
117
    PublishLayerVersionResponse,
118
    PutFunctionCodeSigningConfigResponse,
119
    PutFunctionRecursionConfigResponse,
120
    PutProvisionedConcurrencyConfigResponse,
121
    Qualifier,
122
    RecursiveLoop,
123
    ReservedConcurrentExecutions,
124
    ResourceConflictException,
125
    ResourceNotFoundException,
126
    Runtime,
127
    RuntimeVersionConfig,
128
    SnapStart,
129
    SnapStartApplyOn,
130
    SnapStartOptimizationStatus,
131
    SnapStartResponse,
132
    State,
133
    StatementId,
134
    StateReasonCode,
135
    String,
136
    TaggableResource,
137
    TagKeyList,
138
    Tags,
139
    TenantId,
140
    TracingMode,
141
    UnqualifiedFunctionName,
142
    UpdateCodeSigningConfigResponse,
143
    UpdateEventSourceMappingRequest,
144
    UpdateFunctionCodeRequest,
145
    UpdateFunctionConfigurationRequest,
146
    UpdateFunctionUrlConfigResponse,
147
    VersionWithLatestPublished,
148
)
149
from localstack.aws.api.lambda_ import FunctionVersion as FunctionVersionApi
1✔
150
from localstack.aws.api.lambda_ import ServiceException as LambdaServiceException
1✔
151
from localstack.aws.api.pipes import (
1✔
152
    DynamoDBStreamStartPosition,
153
    KinesisStreamStartPosition,
154
)
155
from localstack.aws.connect import connect_to
1✔
156
from localstack.aws.spec import load_service
1✔
157
from localstack.services.edge import ROUTER
1✔
158
from localstack.services.lambda_ import api_utils
1✔
159
from localstack.services.lambda_ import hooks as lambda_hooks
1✔
160
from localstack.services.lambda_.analytics import (
1✔
161
    FunctionInitializationType,
162
    FunctionOperation,
163
    FunctionStatus,
164
    function_counter,
165
)
166
from localstack.services.lambda_.api_utils import (
1✔
167
    ARCHITECTURES,
168
    STATEMENT_ID_REGEX,
169
    SUBNET_ID_REGEX,
170
    function_locators_from_arn,
171
)
172
from localstack.services.lambda_.event_source_mapping.esm_config_factory import (
1✔
173
    EsmConfigFactory,
174
)
175
from localstack.services.lambda_.event_source_mapping.esm_worker import (
1✔
176
    EsmState,
177
    EsmWorker,
178
)
179
from localstack.services.lambda_.event_source_mapping.esm_worker_factory import (
1✔
180
    EsmWorkerFactory,
181
)
182
from localstack.services.lambda_.event_source_mapping.pipe_utils import get_internal_client
1✔
183
from localstack.services.lambda_.invocation import AccessDeniedException
1✔
184
from localstack.services.lambda_.invocation.execution_environment import (
1✔
185
    EnvironmentStartupTimeoutException,
186
)
187
from localstack.services.lambda_.invocation.lambda_models import (
1✔
188
    AliasRoutingConfig,
189
    CodeSigningConfig,
190
    EventInvokeConfig,
191
    Function,
192
    FunctionResourcePolicy,
193
    FunctionUrlConfig,
194
    FunctionVersion,
195
    ImageConfig,
196
    LambdaEphemeralStorage,
197
    Layer,
198
    LayerPolicy,
199
    LayerPolicyStatement,
200
    LayerVersion,
201
    ProvisionedConcurrencyConfiguration,
202
    RequestEntityTooLargeException,
203
    ResourcePolicy,
204
    UpdateStatus,
205
    ValidationException,
206
    VersionAlias,
207
    VersionFunctionConfiguration,
208
    VersionIdentifier,
209
    VersionState,
210
    VpcConfig,
211
)
212
from localstack.services.lambda_.invocation.lambda_service import (
1✔
213
    LambdaService,
214
    create_image_code,
215
    destroy_code_if_not_used,
216
    lambda_stores,
217
    store_lambda_archive,
218
    store_s3_bucket_archive,
219
)
220
from localstack.services.lambda_.invocation.models import CapacityProvider as CapacityProviderModel
1✔
221
from localstack.services.lambda_.invocation.models import LambdaStore
1✔
222
from localstack.services.lambda_.invocation.runtime_executor import get_runtime_executor
1✔
223
from localstack.services.lambda_.lambda_utils import HINT_LOG
1✔
224
from localstack.services.lambda_.layerfetcher.layer_fetcher import LayerFetcher
1✔
225
from localstack.services.lambda_.provider_utils import (
1✔
226
    LambdaLayerVersionIdentifier,
227
    get_function_version,
228
    get_function_version_from_arn,
229
)
230
from localstack.services.lambda_.runtimes import (
1✔
231
    ALL_RUNTIMES,
232
    DEPRECATED_RUNTIMES,
233
    DEPRECATED_RUNTIMES_UPGRADES,
234
    RUNTIMES_AGGREGATED,
235
    SNAP_START_SUPPORTED_RUNTIMES,
236
    VALID_MANAGED_INSTANCE_RUNTIMES,
237
    VALID_RUNTIMES,
238
)
239
from localstack.services.lambda_.urlrouter import FunctionUrlRouter
1✔
240
from localstack.services.plugins import ServiceLifecycleHook
1✔
241
from localstack.state import StateVisitor
1✔
242
from localstack.utils.aws.arns import (
1✔
243
    ArnData,
244
    capacity_provider_arn,
245
    extract_resource_from_arn,
246
    extract_service_from_arn,
247
    get_partition,
248
    lambda_event_source_mapping_arn,
249
    parse_arn,
250
)
251
from localstack.utils.aws.client_types import ServicePrincipal
1✔
252
from localstack.utils.bootstrap import is_api_enabled
1✔
253
from localstack.utils.collections import PaginatedList, merge_recursive
1✔
254
from localstack.utils.event_matcher import validate_event_pattern
1✔
255
from localstack.utils.strings import get_random_hex, short_uid, to_bytes, to_str
1✔
256
from localstack.utils.sync import poll_condition
1✔
257
from localstack.utils.urls import localstack_host
1✔
258

259
LOG = logging.getLogger(__name__)
1✔
260

261
CAPACITY_PROVIDER_ARN_NAME = "arn:aws[a-zA-Z-]*:lambda:(eusc-)?[a-z]{2}((-gov)|(-iso([a-z]?)))?-[a-z]+-\\d{1}:\\d{12}:capacity-provider:[a-zA-Z0-9-_]+"
1✔
262
LAMBDA_DEFAULT_TIMEOUT = 3
1✔
263
LAMBDA_DEFAULT_MEMORY_SIZE = 128
1✔
264

265
LAMBDA_TAG_LIMIT_PER_RESOURCE = 50
1✔
266
LAMBDA_LAYERS_LIMIT_PER_FUNCTION = 5
1✔
267

268
TAG_KEY_CUSTOM_URL = "_custom_id_"
1✔
269
# Requirements (from RFC3986 & co): not longer than 63, first char must be
270
# alpha, then alphanumeric or hyphen, except cannot start or end with hyphen
271
TAG_KEY_CUSTOM_URL_VALIDATOR = re.compile(r"^[A-Za-z]([A-Za-z0-9\-]{0,61}[A-Za-z0-9])?$")
1✔
272

273

274
class LambdaProvider(LambdaApi, ServiceLifecycleHook):
1✔
275
    lambda_service: LambdaService
1✔
276
    create_fn_lock: threading.RLock
1✔
277
    create_layer_lock: threading.RLock
1✔
278
    router: FunctionUrlRouter
1✔
279
    esm_workers: dict[str, EsmWorker]
1✔
280
    layer_fetcher: LayerFetcher | None
1✔
281

282
    def __init__(self) -> None:
1✔
283
        self.lambda_service = LambdaService()
1✔
284
        self.create_fn_lock = threading.RLock()
1✔
285
        self.create_layer_lock = threading.RLock()
1✔
286
        self.router = FunctionUrlRouter(ROUTER, self.lambda_service)
1✔
287
        self.esm_workers = {}
1✔
288
        self.layer_fetcher = None
1✔
289
        lambda_hooks.inject_layer_fetcher.run(self)
1✔
290

291
    def accept_state_visitor(self, visitor: StateVisitor):
1✔
292
        visitor.visit(lambda_stores)
×
293

294
    def on_before_state_reset(self):
1✔
295
        for esm_worker in self.esm_workers.values():
×
296
            esm_worker.stop_for_shutdown()
×
297
        self.esm_workers = {}
×
298
        self.lambda_service.stop()
×
299

300
    def on_after_state_reset(self):
1✔
301
        self.router.lambda_service = self.lambda_service = LambdaService()
×
302

303
    def on_before_state_load(self):
1✔
304
        self.lambda_service.stop()
×
305

306
    def on_after_state_load(self):
1✔
307
        self.lambda_service = LambdaService()
×
308
        self.router.lambda_service = self.lambda_service
×
309

310
        for account_id, account_bundle in lambda_stores.items():
×
311
            for region_name, state in account_bundle.items():
×
312
                for fn in state.functions.values():
×
313
                    # HACK to model a volatile variable that should be ignored for persistence
314
                    # Identifier unique to this function and LocalStack instance.
315
                    # A LocalStack restart or persistence load should create a new instance id.
316
                    # Used for retaining invoke queues across version updates for $LATEST, but
317
                    # separate unrelated instances.
318
                    fn.instance_id = short_uid()
×
319

320
                    for fn_version in fn.versions.values():
×
321
                        try:
×
322
                            # $LATEST is not invokable for Lambda functions with a capacity provider
323
                            # and has a different State (i.e., ActiveNonInvokable)
324
                            is_capacity_provider_latest = (
×
325
                                fn_version.config.capacity_provider_config
326
                                and fn_version.id.qualifier == "$LATEST"
327
                            )
328
                            if not is_capacity_provider_latest:
×
329
                                # Restore the "Pending" state for the function version and start it
330
                                new_state = VersionState(
×
331
                                    state=State.Pending,
332
                                    code=StateReasonCode.Creating,
333
                                    reason="The function is being created.",
334
                                )
335
                                new_config = dataclasses.replace(fn_version.config, state=new_state)
×
336
                                new_version = dataclasses.replace(fn_version, config=new_config)
×
337
                                fn.versions[fn_version.id.qualifier] = new_version
×
338
                                self.lambda_service.create_function_version(fn_version).result(
×
339
                                    timeout=5
340
                                )
341
                        except Exception:
×
342
                            LOG.warning(
×
343
                                "Failed to restore function version %s",
344
                                fn_version.id.qualified_arn(),
345
                                exc_info=LOG.isEnabledFor(logging.DEBUG),
346
                            )
347
                    # restore provisioned concurrency per function considering both versions and aliases
348
                    for (
×
349
                        provisioned_qualifier,
350
                        provisioned_config,
351
                    ) in fn.provisioned_concurrency_configs.items():
352
                        fn_arn = None
×
353
                        try:
×
354
                            if api_utils.qualifier_is_alias(provisioned_qualifier):
×
355
                                alias = fn.aliases.get(provisioned_qualifier)
×
356
                                resolved_version = fn.versions.get(alias.function_version)
×
357
                                fn_arn = resolved_version.id.qualified_arn()
×
358
                            elif api_utils.qualifier_is_version(provisioned_qualifier):
×
359
                                fn_version = fn.versions.get(provisioned_qualifier)
×
360
                                fn_arn = fn_version.id.qualified_arn()
×
361
                            else:
362
                                raise InvalidParameterValueException(
×
363
                                    "Invalid qualifier type:"
364
                                    " Qualifier can only be an alias or a version for provisioned concurrency."
365
                                )
366

367
                            manager = self.lambda_service.get_lambda_version_manager(fn_arn)
×
368
                            manager.update_provisioned_concurrency_config(
×
369
                                provisioned_config.provisioned_concurrent_executions
370
                            )
371
                        except Exception:
×
372
                            LOG.warning(
×
373
                                "Failed to restore provisioned concurrency %s for function %s",
374
                                provisioned_config,
375
                                fn_arn,
376
                                exc_info=LOG.isEnabledFor(logging.DEBUG),
377
                            )
378

379
                for esm in state.event_source_mappings.values():
×
380
                    # Restores event source workers
381
                    function_arn = esm.get("FunctionArn")
×
382

383
                    # TODO: How do we know the event source is up?
384
                    # A basic poll to see if the mapped Lambda function is active/failed
385
                    if not poll_condition(
×
386
                        lambda: (
387
                            get_function_version_from_arn(function_arn).config.state.state
388
                            in [State.Active, State.Failed]
389
                        ),
390
                        timeout=10,
391
                    ):
UNCOV
392
                        LOG.warning(
×
393
                            "Creating ESM for Lambda that is not in running state: %s",
394
                            function_arn,
395
                        )
396

UNCOV
397
                    function_version = get_function_version_from_arn(function_arn)
×
398
                    function_role = function_version.config.role
×
399

UNCOV
400
                    is_esm_enabled = esm.get("State", EsmState.DISABLED) not in (
×
401
                        EsmState.DISABLED,
402
                        EsmState.DISABLING,
403
                    )
UNCOV
404
                    esm_worker = EsmWorkerFactory(
×
405
                        esm, function_role, is_esm_enabled
406
                    ).get_esm_worker()
407

408
                    # Note: a worker is created in the DISABLED state if not enabled
UNCOV
409
                    esm_worker.create()
×
410
                    # TODO: assigning the esm_worker to the dict only works after .create(). Could it cause a race
411
                    #  condition if we get a shutdown here and have a worker thread spawned but not accounted for?
UNCOV
412
                    self.esm_workers[esm_worker.uuid] = esm_worker
×
413

414
    def on_after_init(self):
1✔
415
        self.router.register_routes()
1✔
416
        get_runtime_executor().validate_environment()
1✔
417

418
    def on_before_stop(self) -> None:
1✔
419
        for esm_worker in self.esm_workers.values():
1✔
420
            esm_worker.stop_for_shutdown()
1✔
421

422
        # TODO: should probably unregister routes?
423
        self.lambda_service.stop()
1✔
424

425
    @staticmethod
1✔
426
    def _get_function(function_name: str, account_id: str, region: str) -> Function:
1✔
427
        state = lambda_stores[account_id][region]
1✔
428
        function = state.functions.get(function_name)
1✔
429
        if not function:
1✔
430
            arn = api_utils.unqualified_lambda_arn(
1✔
431
                function_name=function_name,
432
                account=account_id,
433
                region=region,
434
            )
435
            raise ResourceNotFoundException(
1✔
436
                f"Function not found: {arn}",
437
                Type="User",
438
            )
439
        return function
1✔
440

441
    @staticmethod
1✔
442
    def _get_esm(uuid: str, account_id: str, region: str) -> EventSourceMappingConfiguration:
1✔
443
        state = lambda_stores[account_id][region]
1✔
444
        esm = state.event_source_mappings.get(uuid)
1✔
445
        if not esm:
1✔
446
            arn = lambda_event_source_mapping_arn(uuid, account_id, region)
1✔
447
            raise ResourceNotFoundException(
1✔
448
                f"Event source mapping not found: {arn}",
449
                Type="User",
450
            )
451
        return esm
1✔
452

453
    @staticmethod
1✔
454
    def _get_capacity_provider(
1✔
455
        capacity_provider_name: str,
456
        account_id: str,
457
        region: str,
458
        error_msg_template: str = "Capacity provider not found: {}",
459
    ) -> CapacityProviderModel:
460
        state = lambda_stores[account_id][region]
1✔
461
        cp = state.capacity_providers.get(capacity_provider_name)
1✔
462
        if not cp:
1✔
463
            arn = capacity_provider_arn(capacity_provider_name, account_id, region)
1✔
464
            raise ResourceNotFoundException(
1✔
465
                error_msg_template.format(arn),
466
                Type="User",
467
            )
UNCOV
468
        return cp
×
469

470
    @staticmethod
1✔
471
    def _validate_qualifier_expression(qualifier: str) -> None:
1✔
472
        if error_messages := api_utils.validate_qualifier(qualifier):
1✔
UNCOV
473
            raise ValidationException(
×
474
                message=api_utils.construct_validation_exception_message(error_messages)
475
            )
476

477
    @staticmethod
1✔
478
    def _validate_publish_to(publish_to: str):
1✔
UNCOV
479
        if publish_to != FunctionVersionLatestPublished.LATEST_PUBLISHED:
×
UNCOV
480
            raise ValidationException(
×
481
                message=f"1 validation error detected: Value '{publish_to}' at 'publishTo' failed to satisfy constraint: Member must satisfy enum value set: [LATEST_PUBLISHED]"
482
            )
483

484
    @staticmethod
1✔
485
    def _resolve_fn_qualifier(resolved_fn: Function, qualifier: str | None) -> tuple[str, str]:
1✔
486
        """Attempts to resolve a given qualifier and returns a qualifier that exists or
487
        raises an appropriate ResourceNotFoundException.
488

489
        :param resolved_fn: The resolved lambda function
490
        :param qualifier: The qualifier to be resolved or None
491
        :return: Tuple of (resolved qualifier, function arn either qualified or unqualified)"""
492
        function_name = resolved_fn.function_name
1✔
493
        # assuming function versions need to live in the same account and region
494
        account_id = resolved_fn.latest().id.account
1✔
495
        region = resolved_fn.latest().id.region
1✔
496
        fn_arn = api_utils.unqualified_lambda_arn(function_name, account_id, region)
1✔
497
        if qualifier is not None:
1✔
498
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
499
            if api_utils.qualifier_is_alias(qualifier):
1✔
500
                if qualifier not in resolved_fn.aliases:
1✔
501
                    raise ResourceNotFoundException(f"Cannot find alias arn: {fn_arn}", Type="User")
1✔
502
            elif api_utils.qualifier_is_version(qualifier) or qualifier == "$LATEST":
1✔
503
                if qualifier not in resolved_fn.versions:
1✔
504
                    raise ResourceNotFoundException(f"Function not found: {fn_arn}", Type="User")
1✔
505
            else:
506
                # matches qualifier pattern but invalid alias or version
507
                raise ResourceNotFoundException(f"Function not found: {fn_arn}", Type="User")
1✔
508
        resolved_qualifier = qualifier or "$LATEST"
1✔
509
        return resolved_qualifier, fn_arn
1✔
510

511
    @staticmethod
1✔
512
    def _function_revision_id(resolved_fn: Function, resolved_qualifier: str) -> str:
1✔
513
        if api_utils.qualifier_is_alias(resolved_qualifier):
1✔
514
            return resolved_fn.aliases[resolved_qualifier].revision_id
1✔
515
        # Assumes that a non-alias is a version
516
        else:
517
            return resolved_fn.versions[resolved_qualifier].config.revision_id
1✔
518

519
    def _resolve_vpc_id(self, account_id: str, region_name: str, subnet_id: str) -> str:
1✔
520
        ec2_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ec2
1✔
521
        try:
1✔
522
            return ec2_client.describe_subnets(SubnetIds=[subnet_id])["Subnets"][0]["VpcId"]
1✔
523
        except ec2_client.exceptions.ClientError as e:
1✔
524
            code = e.response["Error"]["Code"]
1✔
525
            message = e.response["Error"]["Message"]
1✔
526
            raise InvalidParameterValueException(
1✔
527
                f"Error occurred while DescribeSubnets. EC2 Error Code: {code}. EC2 Error Message: {message}",
528
                Type="User",
529
            )
530

531
    def _build_vpc_config(
1✔
532
        self,
533
        account_id: str,
534
        region_name: str,
535
        vpc_config: dict | None = None,
536
    ) -> VpcConfig | None:
537
        if not vpc_config or not is_api_enabled("ec2"):
1✔
538
            return None
1✔
539

540
        subnet_ids = vpc_config.get("SubnetIds", [])
1✔
541
        if subnet_ids is not None and len(subnet_ids) == 0:
1✔
542
            return VpcConfig(vpc_id="", security_group_ids=[], subnet_ids=[])
1✔
543

544
        subnet_id = subnet_ids[0]
1✔
545
        if not bool(SUBNET_ID_REGEX.match(subnet_id)):
1✔
546
            raise ValidationException(
1✔
547
                f"1 validation error detected: Value '[{subnet_id}]' at 'vpcConfig.subnetIds' failed to satisfy constraint: Member must satisfy constraint: [Member must have length less than or equal to 1024, Member must have length greater than or equal to 0, Member must satisfy regular expression pattern: subnet-[0-9a-z]*]"
548
            )
549

550
        return VpcConfig(
1✔
551
            vpc_id=self._resolve_vpc_id(account_id, region_name, subnet_id),
552
            security_group_ids=vpc_config.get("SecurityGroupIds", []),
553
            subnet_ids=subnet_ids,
554
        )
555

556
    def _create_version_model(
1✔
557
        self,
558
        function_name: str,
559
        region: str,
560
        account_id: str,
561
        description: str | None = None,
562
        revision_id: str | None = None,
563
        code_sha256: str | None = None,
564
        publish_to: FunctionVersionLatestPublished | None = None,
565
        is_active: bool = False,
566
    ) -> tuple[FunctionVersion, bool]:
567
        """
568
        Release a new version to the model if all restrictions are met.
569
        Restrictions:
570
          - CodeSha256, if provided, must equal the current latest version code hash
571
          - RevisionId, if provided, must equal the current latest version revision id
572
          - Some changes have been done to the latest version since last publish
573
        Will return a tuple of the version, and whether the version was published (True) or the latest available version was taken (False).
574
        This can happen if the latest version has not been changed since the last version publish, in this case the last version will be returned.
575

576
        :param function_name: Function name to be published
577
        :param region: Region of the function
578
        :param account_id: Account of the function
579
        :param description: new description of the version (will be the description of the function if missing)
580
        :param revision_id: Revision id, function will raise error if it does not match latest revision id
581
        :param code_sha256: Code sha256, function will raise error if it does not match latest code hash
582
        :return: Tuple of (published version, whether version was released or last released version returned, since nothing changed)
583
        """
584
        current_latest_version = get_function_version(
1✔
585
            function_name=function_name, qualifier="$LATEST", account_id=account_id, region=region
586
        )
587
        if revision_id and current_latest_version.config.revision_id != revision_id:
1✔
588
            raise PreconditionFailedException(
1✔
589
                "The Revision Id provided does not match the latest Revision Id. Call the GetFunction/GetAlias API to retrieve the latest Revision Id",
590
                Type="User",
591
            )
592

593
        # check if code hashes match if they are specified
594
        current_hash = (
1✔
595
            current_latest_version.config.code.code_sha256
596
            if current_latest_version.config.package_type == PackageType.Zip
597
            else current_latest_version.config.image.code_sha256
598
        )
599
        # if the code is a zip package and hot reloaded (hot reloading is currently only supported for zip packagetypes)
600
        # we cannot enforce the codesha256 check
601
        is_hot_reloaded_zip_package = (
1✔
602
            current_latest_version.config.package_type == PackageType.Zip
603
            and current_latest_version.config.code.is_hot_reloading()
604
        )
605
        if code_sha256 and current_hash != code_sha256 and not is_hot_reloaded_zip_package:
1✔
606
            raise InvalidParameterValueException(
1✔
607
                f"CodeSHA256 ({code_sha256}) is different from current CodeSHA256 in $LATEST ({current_hash}). Please try again with the CodeSHA256 in $LATEST.",
608
                Type="User",
609
            )
610

611
        state = lambda_stores[account_id][region]
1✔
612
        function = state.functions.get(function_name)
1✔
613
        changes = {}
1✔
614
        if description is not None:
1✔
615
            changes["description"] = description
1✔
616
        # TODO copy environment instead of restarting one, get rid of all the "Pending"s
617

618
        with function.lock:
1✔
619
            if function.next_version > 1 and (
1✔
620
                prev_version := function.versions.get(str(function.next_version - 1))
621
            ):
622
                if (
1✔
623
                    prev_version.config.internal_revision
624
                    == current_latest_version.config.internal_revision
625
                ):
626
                    return prev_version, False
1✔
627
            # TODO check if there was a change since last version
628
            if publish_to == FunctionVersionLatestPublished.LATEST_PUBLISHED:
1✔
UNCOV
629
                qualifier = "$LATEST.PUBLISHED"
×
630
            else:
631
                qualifier = str(function.next_version)
1✔
632
                function.next_version += 1
1✔
633
            new_id = VersionIdentifier(
1✔
634
                function_name=function_name,
635
                qualifier=qualifier,
636
                region=region,
637
                account=account_id,
638
            )
639

640
            if current_latest_version.config.capacity_provider_config:
1✔
641
                # for lambda managed functions, snap start is not supported
UNCOV
642
                snap_start = None
×
643
            else:
644
                apply_on = current_latest_version.config.snap_start["ApplyOn"]
1✔
645
                optimization_status = SnapStartOptimizationStatus.Off
1✔
646
                if apply_on == SnapStartApplyOn.PublishedVersions:
1✔
UNCOV
647
                    optimization_status = SnapStartOptimizationStatus.On
×
648
                snap_start = SnapStartResponse(
1✔
649
                    ApplyOn=apply_on,
650
                    OptimizationStatus=optimization_status,
651
                )
652

653
            last_update = None
1✔
654
            new_state = VersionState(
1✔
655
                state=State.Pending,
656
                code=StateReasonCode.Creating,
657
                reason="The function is being created.",
658
            )
659
            if publish_to == FunctionVersionLatestPublished.LATEST_PUBLISHED:
1✔
UNCOV
660
                last_update = UpdateStatus(
×
661
                    status=LastUpdateStatus.InProgress,
662
                    code="Updating",
663
                    reason="The function is being updated.",
664
                )
UNCOV
665
                if is_active:
×
UNCOV
666
                    new_state = VersionState(state=State.Active)
×
667
            new_version = dataclasses.replace(
1✔
668
                current_latest_version,
669
                config=dataclasses.replace(
670
                    current_latest_version.config,
671
                    last_update=last_update,
672
                    state=new_state,
673
                    snap_start=snap_start,
674
                    **changes,
675
                ),
676
                id=new_id,
677
            )
678
            function.versions[qualifier] = new_version
1✔
679
        return new_version, True
1✔
680

681
    def _publish_version_from_existing_version(
1✔
682
        self,
683
        function_name: str,
684
        region: str,
685
        account_id: str,
686
        description: str | None = None,
687
        revision_id: str | None = None,
688
        code_sha256: str | None = None,
689
        publish_to: FunctionVersionLatestPublished | None = None,
690
    ) -> FunctionVersion:
691
        """
692
        Publish version from an existing, already initialized LATEST
693

694
        :param function_name: Function name
695
        :param region: region
696
        :param account_id: account id
697
        :param description: description
698
        :param revision_id: revision id (check if current version matches)
699
        :param code_sha256: code sha (check if current code matches)
700
        :return: new version
701
        """
702
        is_active = True if publish_to == FunctionVersionLatestPublished.LATEST_PUBLISHED else False
1✔
703
        new_version, changed = self._create_version_model(
1✔
704
            function_name=function_name,
705
            region=region,
706
            account_id=account_id,
707
            description=description,
708
            revision_id=revision_id,
709
            code_sha256=code_sha256,
710
            publish_to=publish_to,
711
            is_active=is_active,
712
        )
713
        if not changed:
1✔
714
            return new_version
1✔
715

716
        if new_version.config.capacity_provider_config:
1✔
UNCOV
717
            self.lambda_service.publish_version_async(new_version)
×
718
        else:
719
            self.lambda_service.publish_version(new_version)
1✔
720
        state = lambda_stores[account_id][region]
1✔
721
        function = state.functions.get(function_name)
1✔
722

723
        # Update revision id for $LATEST version
724
        # TODO: re-evaluate data model to prevent this dirty hack just for bumping the revision id
725
        latest_version = function.versions["$LATEST"]
1✔
726
        function.versions["$LATEST"] = dataclasses.replace(
1✔
727
            latest_version, config=dataclasses.replace(latest_version.config)
728
        )
729
        if new_version.config.capacity_provider_config:
1✔
730
            # publish_version happens async for functions with a capacity provider.
731
            # Therefore, we return the new_version with State=Pending or LastUpdateStatus=InProgress ($LATEST.PUBLISHED)
UNCOV
732
            return new_version
×
733
        else:
734
            # Regular functions yield an Active state modified during `publish_version` (sync).
735
            # Therefore, we need to get the updated version from the store.
736
            updated_version = function.versions.get(new_version.id.qualifier)
1✔
737
            return updated_version
1✔
738

739
    def _publish_version_with_changes(
1✔
740
        self,
741
        function_name: str,
742
        region: str,
743
        account_id: str,
744
        description: str | None = None,
745
        revision_id: str | None = None,
746
        code_sha256: str | None = None,
747
        publish_to: FunctionVersionLatestPublished | None = None,
748
        is_active: bool = False,
749
    ) -> FunctionVersion:
750
        """
751
        Publish version together with a new latest version (publish on create / update)
752

753
        :param function_name: Function name
754
        :param region: region
755
        :param account_id: account id
756
        :param description: description
757
        :param revision_id: revision id (check if current version matches)
758
        :param code_sha256: code sha (check if current code matches)
759
        :return: new version
760
        """
761
        new_version, changed = self._create_version_model(
1✔
762
            function_name=function_name,
763
            region=region,
764
            account_id=account_id,
765
            description=description,
766
            revision_id=revision_id,
767
            code_sha256=code_sha256,
768
            publish_to=publish_to,
769
            is_active=is_active,
770
        )
771
        if not changed:
1✔
UNCOV
772
            return new_version
×
773
        self.lambda_service.create_function_version(new_version)
1✔
774
        return new_version
1✔
775

776
    @staticmethod
1✔
777
    def _verify_env_variables(env_vars: dict[str, str]):
1✔
778
        dumped_env_vars = json.dumps(env_vars, separators=(",", ":"))
1✔
779
        if (
1✔
780
            len(dumped_env_vars.encode("utf-8"))
781
            > config.LAMBDA_LIMITS_MAX_FUNCTION_ENVVAR_SIZE_BYTES
782
        ):
783
            raise InvalidParameterValueException(
1✔
784
                f"Lambda was unable to configure your environment variables because the environment variables you have provided exceeded the 4KB limit. String measured: {dumped_env_vars}",
785
                Type="User",
786
            )
787

788
    @staticmethod
1✔
789
    def _validate_snapstart(snap_start: SnapStart, runtime: Runtime):
1✔
790
        apply_on = snap_start.get("ApplyOn")
1✔
791
        if apply_on not in [
1✔
792
            SnapStartApplyOn.PublishedVersions,
793
            SnapStartApplyOn.None_,
794
        ]:
795
            raise ValidationException(
1✔
796
                f"1 validation error detected: Value '{apply_on}' at 'snapStart.applyOn' failed to satisfy constraint: Member must satisfy enum value set: [PublishedVersions, None]"
797
            )
798

799
        if runtime not in SNAP_START_SUPPORTED_RUNTIMES:
1✔
UNCOV
800
            raise InvalidParameterValueException(
×
801
                f"{runtime} is not supported for SnapStart enabled functions.", Type="User"
802
            )
803

804
    def _validate_layers(self, new_layers: list[str], region: str, account_id: str):
1✔
805
        if len(new_layers) > LAMBDA_LAYERS_LIMIT_PER_FUNCTION:
1✔
806
            raise InvalidParameterValueException(
1✔
807
                "Cannot reference more than 5 layers.", Type="User"
808
            )
809

810
        visited_layers = {}
1✔
811
        for layer_version_arn in new_layers:
1✔
812
            (
1✔
813
                layer_region,
814
                layer_account_id,
815
                layer_name,
816
                layer_version_str,
817
            ) = api_utils.parse_layer_arn(layer_version_arn)
818
            if layer_version_str is None:
1✔
819
                raise ValidationException(
1✔
820
                    f"1 validation error detected: Value '[{layer_version_arn}]'"
821
                    + " at 'layers' failed to satisfy constraint: Member must satisfy constraint: [Member must have length less than or equal to 2048, Member must have length greater than or equal to 1, Member must satisfy regular expression pattern: "
822
                    + "(arn:(aws[a-zA-Z-]*)?:lambda:(eusc-)?[a-z]{2}((-gov)|(-iso([a-z]?)))?-[a-z]+-\\d{1}:\\d{12}:layer:[a-zA-Z0-9-_]+:[0-9]+)|(arn:[a-zA-Z0-9-]+:lambda:::awslayer:[a-zA-Z0-9-_]+), Member must not be null]",
823
                )
824

825
            state = lambda_stores[layer_account_id][layer_region]
1✔
826
            layer = state.layers.get(layer_name)
1✔
827
            layer_version = None
1✔
828
            if layer is not None:
1✔
829
                layer_version = layer.layer_versions.get(layer_version_str)
1✔
830
            if layer_account_id == account_id:
1✔
831
                if region and layer_region != region:
1✔
832
                    raise InvalidParameterValueException(
1✔
833
                        f"Layers are not in the same region as the function. "
834
                        f"Layers are expected to be in region {region}.",
835
                        Type="User",
836
                    )
837
                if layer is None or layer.layer_versions.get(layer_version_str) is None:
1✔
838
                    raise InvalidParameterValueException(
1✔
839
                        f"Layer version {layer_version_arn} does not exist.", Type="User"
840
                    )
841
            else:  # External layer from other account
842
                # TODO: validate IAM layer policy here, allowing access by default for now and only checking region
843
                if region and layer_region != region:
×
844
                    # TODO: detect user or role from context when IAM users are implemented
UNCOV
845
                    user = "user/localstack-testing"
×
UNCOV
846
                    raise AccessDeniedException(
×
847
                        f"User: arn:{get_partition(region)}:iam::{account_id}:{user} is not authorized to perform: lambda:GetLayerVersion on resource: {layer_version_arn} because no resource-based policy allows the lambda:GetLayerVersion action"
848
                    )
UNCOV
849
                if layer is None or layer_version is None:
×
850
                    # Limitation: cannot fetch external layers when using the same account id as the target layer
851
                    # because we do not want to trigger the layer fetcher for every non-existing layer.
UNCOV
852
                    if self.layer_fetcher is None:
×
853
                        raise NotImplementedError(
854
                            "Fetching shared layers from AWS is a pro feature."
855
                        )
856

UNCOV
857
                    layer = self.layer_fetcher.fetch_layer(layer_version_arn)
×
858
                    if layer is None:
×
859
                        # TODO: detect user or role from context when IAM users are implemented
UNCOV
860
                        user = "user/localstack-testing"
×
UNCOV
861
                        raise AccessDeniedException(
×
862
                            f"User: arn:{get_partition(region)}:iam::{account_id}:{user} is not authorized to perform: lambda:GetLayerVersion on resource: {layer_version_arn} because no resource-based policy allows the lambda:GetLayerVersion action"
863
                        )
864

865
                    # Distinguish between new layer and new layer version
866
                    if layer_version is None:
×
867
                        # Create whole layer from scratch
UNCOV
868
                        state.layers[layer_name] = layer
×
869
                    else:
870
                        # Create layer version if another version of the same layer already exists
UNCOV
871
                        state.layers[layer_name].layer_versions[layer_version_str] = (
×
872
                            layer.layer_versions.get(layer_version_str)
873
                        )
874

875
            # only the first two matches in the array are considered for the error message
876
            layer_arn = ":".join(layer_version_arn.split(":")[:-1])
1✔
877
            if layer_arn in visited_layers:
1✔
878
                conflict_layer_version_arn = visited_layers[layer_arn]
1✔
879
                raise InvalidParameterValueException(
1✔
880
                    f"Two different versions of the same layer are not allowed to be referenced in the same function. {conflict_layer_version_arn} and {layer_version_arn} are versions of the same layer.",
881
                    Type="User",
882
                )
883
            visited_layers[layer_arn] = layer_version_arn
1✔
884

885
    def _validate_capacity_provider_config(
1✔
886
        self, capacity_provider_config: CapacityProviderConfig, context: RequestContext
887
    ):
UNCOV
888
        if not capacity_provider_config.get("LambdaManagedInstancesCapacityProviderConfig"):
×
UNCOV
889
            raise ValidationException(
×
890
                "1 validation error detected: Value null at 'capacityProviderConfig.lambdaManagedInstancesCapacityProviderConfig' failed to satisfy constraint: Member must not be null"
891
            )
892

UNCOV
893
        capacity_provider_arn = capacity_provider_config.get(
×
894
            "LambdaManagedInstancesCapacityProviderConfig", {}
895
        ).get("CapacityProviderArn")
UNCOV
896
        if not capacity_provider_arn:
×
UNCOV
897
            raise ValidationException(
×
898
                "1 validation error detected: Value null at 'capacityProviderConfig.lambdaManagedInstancesCapacityProviderConfig.capacityProviderArn' failed to satisfy constraint: Member must not be null"
899
            )
900

UNCOV
901
        if not re.match(CAPACITY_PROVIDER_ARN_NAME, capacity_provider_arn):
×
UNCOV
902
            raise ValidationException(
×
903
                f"1 validation error detected: Value '{capacity_provider_arn}' at 'capacityProviderConfig.lambdaManagedInstancesCapacityProviderConfig.capacityProviderArn' failed to satisfy constraint: Member must satisfy regular expression pattern: {CAPACITY_PROVIDER_ARN_NAME}"
904
            )
905

UNCOV
906
        capacity_provider_name = capacity_provider_arn.split(":")[-1]
×
UNCOV
907
        self.get_capacity_provider(context, capacity_provider_name)
×
908

909
    @staticmethod
1✔
910
    def map_layers(new_layers: list[str]) -> list[LayerVersion]:
1✔
911
        layers = []
1✔
912
        for layer_version_arn in new_layers:
1✔
913
            region_name, account_id, layer_name, layer_version = api_utils.parse_layer_arn(
1✔
914
                layer_version_arn
915
            )
916
            layer = lambda_stores[account_id][region_name].layers.get(layer_name)
1✔
917
            layer_version = layer.layer_versions.get(layer_version)
1✔
918
            layers.append(layer_version)
1✔
919
        return layers
1✔
920

921
    def get_function_recursion_config(
1✔
922
        self,
923
        context: RequestContext,
924
        function_name: UnqualifiedFunctionName,
925
        **kwargs,
926
    ) -> GetFunctionRecursionConfigResponse:
927
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
928
        function_name = api_utils.get_function_name(function_name, context)
1✔
929
        fn = self._get_function(function_name=function_name, region=region, account_id=account_id)
1✔
930
        return GetFunctionRecursionConfigResponse(RecursiveLoop=fn.recursive_loop)
1✔
931

932
    def put_function_recursion_config(
1✔
933
        self,
934
        context: RequestContext,
935
        function_name: UnqualifiedFunctionName,
936
        recursive_loop: RecursiveLoop,
937
        **kwargs,
938
    ) -> PutFunctionRecursionConfigResponse:
939
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
940
        function_name = api_utils.get_function_name(function_name, context)
1✔
941

942
        fn = self._get_function(function_name=function_name, region=region, account_id=account_id)
1✔
943

944
        allowed_values = list(RecursiveLoop.__members__.values())
1✔
945
        if recursive_loop not in allowed_values:
1✔
946
            raise ValidationException(
1✔
947
                f"1 validation error detected: Value '{recursive_loop}' at 'recursiveLoop' failed to satisfy constraint: "
948
                f"Member must satisfy enum value set: [Terminate, Allow]"
949
            )
950

951
        fn.recursive_loop = recursive_loop
1✔
952
        return PutFunctionRecursionConfigResponse(RecursiveLoop=fn.recursive_loop)
1✔
953

954
    @handler(operation="CreateFunction", expand=False)
1✔
955
    def create_function(
1✔
956
        self,
957
        context: RequestContext,
958
        request: CreateFunctionRequest,
959
    ) -> FunctionConfiguration:
960
        context_region = context.region
1✔
961
        context_account_id = context.account_id
1✔
962

963
        zip_file = request.get("Code", {}).get("ZipFile")
1✔
964
        if zip_file and len(zip_file) > config.LAMBDA_LIMITS_CODE_SIZE_ZIPPED:
1✔
965
            raise RequestEntityTooLargeException(
1✔
966
                f"Zipped size must be smaller than {config.LAMBDA_LIMITS_CODE_SIZE_ZIPPED} bytes"
967
            )
968

969
        if context.request.content_length > config.LAMBDA_LIMITS_CREATE_FUNCTION_REQUEST_SIZE:
1✔
970
            raise RequestEntityTooLargeException(
1✔
971
                f"Request must be smaller than {config.LAMBDA_LIMITS_CREATE_FUNCTION_REQUEST_SIZE} bytes for the CreateFunction operation"
972
            )
973

974
        if architectures := request.get("Architectures"):
1✔
975
            if len(architectures) != 1:
1✔
976
                raise ValidationException(
1✔
977
                    f"1 validation error detected: Value '[{', '.join(architectures)}]' at 'architectures' failed to "
978
                    f"satisfy constraint: Member must have length less than or equal to 1",
979
                )
980
            if architectures[0] not in ARCHITECTURES:
1✔
981
                raise ValidationException(
1✔
982
                    f"1 validation error detected: Value '[{', '.join(architectures)}]' at 'architectures' failed to "
983
                    f"satisfy constraint: Member must satisfy constraint: [Member must satisfy enum value set: "
984
                    f"[x86_64, arm64], Member must not be null]",
985
                )
986

987
        if env_vars := request.get("Environment", {}).get("Variables"):
1✔
988
            self._verify_env_variables(env_vars)
1✔
989

990
        if layers := request.get("Layers", []):
1✔
991
            self._validate_layers(layers, region=context_region, account_id=context_account_id)
1✔
992

993
        if not api_utils.is_role_arn(request.get("Role")):
1✔
994
            raise ValidationException(
1✔
995
                f"1 validation error detected: Value '{request.get('Role')}'"
996
                + " at 'role' failed to satisfy constraint: Member must satisfy regular expression pattern: arn:(aws[a-zA-Z-]*)?:iam::\\d{12}:role/?[a-zA-Z_0-9+=,.@\\-_/]+"
997
            )
998
        if not self.lambda_service.can_assume_role(request.get("Role"), context.region):
1✔
UNCOV
999
            raise InvalidParameterValueException(
×
1000
                "The role defined for the function cannot be assumed by Lambda.", Type="User"
1001
            )
1002
        package_type = request.get("PackageType", PackageType.Zip)
1✔
1003
        runtime = request.get("Runtime")
1✔
1004
        self._validate_runtime(package_type, runtime)
1✔
1005

1006
        request_function_name = request.get("FunctionName")
1✔
1007

1008
        function_name, *_ = api_utils.get_name_and_qualifier(
1✔
1009
            function_arn_or_name=request_function_name,
1010
            qualifier=None,
1011
            context=context,
1012
        )
1013

1014
        if runtime in DEPRECATED_RUNTIMES:
1✔
1015
            LOG.warning(
1✔
1016
                "The Lambda runtime %s} is deprecated. "
1017
                "Please upgrade the runtime for the function %s: "
1018
                "https://docs.aws.amazon.com/lambda/latest/dg/lambda-runtimes.html",
1019
                runtime,
1020
                function_name,
1021
            )
1022
        if snap_start := request.get("SnapStart"):
1✔
1023
            self._validate_snapstart(snap_start, runtime)
1✔
1024
        if publish_to := request.get("PublishTo"):
1✔
UNCOV
1025
            self._validate_publish_to(publish_to)
×
1026
        state = lambda_stores[context_account_id][context_region]
1✔
1027

1028
        with self.create_fn_lock:
1✔
1029
            if function_name in state.functions:
1✔
UNCOV
1030
                raise ResourceConflictException(f"Function already exist: {function_name}")
×
1031
            fn = Function(function_name=function_name)
1✔
1032
            arn = VersionIdentifier(
1✔
1033
                function_name=function_name,
1034
                qualifier="$LATEST",
1035
                region=context_region,
1036
                account=context_account_id,
1037
            )
1038
            # save function code to s3
1039
            code = None
1✔
1040
            image = None
1✔
1041
            image_config = None
1✔
1042
            runtime_version_config = RuntimeVersionConfig(
1✔
1043
                # Limitation: the runtime id (presumably sha256 of image) is currently hardcoded
1044
                # Potential implementation: provide (cached) sha256 hash of used Docker image
1045
                RuntimeVersionArn=f"arn:{context.partition}:lambda:{context_region}::runtime:8eeff65f6809a3ce81507fe733fe09b835899b99481ba22fd75b5a7338290ec1"
1046
            )
1047
            request_code = request.get("Code")
1✔
1048
            if package_type == PackageType.Zip:
1✔
1049
                # TODO verify if correct combination of code is set
1050
                if zip_file := request_code.get("ZipFile"):
1✔
1051
                    code = store_lambda_archive(
1✔
1052
                        archive_file=zip_file,
1053
                        function_name=function_name,
1054
                        region_name=context_region,
1055
                        account_id=context_account_id,
1056
                    )
1057
                elif s3_bucket := request_code.get("S3Bucket"):
1✔
1058
                    s3_key = request_code["S3Key"]
1✔
1059
                    s3_object_version = request_code.get("S3ObjectVersion")
1✔
1060
                    code = store_s3_bucket_archive(
1✔
1061
                        archive_bucket=s3_bucket,
1062
                        archive_key=s3_key,
1063
                        archive_version=s3_object_version,
1064
                        function_name=function_name,
1065
                        region_name=context_region,
1066
                        account_id=context_account_id,
1067
                    )
1068
                else:
UNCOV
1069
                    raise LambdaServiceException("A ZIP file or S3 bucket is required")
×
1070
            elif package_type == PackageType.Image:
1✔
1071
                image = request_code.get("ImageUri")
1✔
1072
                if not image:
1✔
UNCOV
1073
                    raise LambdaServiceException(
×
1074
                        "An image is required when the package type is set to 'image'"
1075
                    )
1076
                image = create_image_code(image_uri=image)
1✔
1077

1078
                image_config_req = request.get("ImageConfig", {})
1✔
1079
                image_config = ImageConfig(
1✔
1080
                    command=image_config_req.get("Command"),
1081
                    entrypoint=image_config_req.get("EntryPoint"),
1082
                    working_directory=image_config_req.get("WorkingDirectory"),
1083
                )
1084
                # Runtime management controls are not available when providing a custom image
1085
                runtime_version_config = None
1✔
1086

1087
            capacity_provider_config = None
1✔
1088
            memory_size = request.get("MemorySize", LAMBDA_DEFAULT_MEMORY_SIZE)
1✔
1089
            if "CapacityProviderConfig" in request:
1✔
1090
                capacity_provider_config = request["CapacityProviderConfig"]
×
UNCOV
1091
                self._validate_capacity_provider_config(capacity_provider_config, context)
×
1092
                self._validate_managed_instances_runtime(runtime)
×
1093

UNCOV
1094
                default_config = CapacityProviderConfig(
×
1095
                    LambdaManagedInstancesCapacityProviderConfig=LambdaManagedInstancesCapacityProviderConfig(
1096
                        ExecutionEnvironmentMemoryGiBPerVCpu=2.0,
1097
                        PerExecutionEnvironmentMaxConcurrency=16,
1098
                    )
1099
                )
1100
                capacity_provider_config = merge_recursive(default_config, capacity_provider_config)
×
1101
                memory_size = 2048
×
UNCOV
1102
                if request.get("LoggingConfig", {}).get("LogFormat") == LogFormat.Text:
×
UNCOV
1103
                    raise InvalidParameterValueException(
×
1104
                        'LogLevel is not supported when LogFormat is set to "Text". Remove LogLevel from your request or change the LogFormat to "JSON" and try again.',
1105
                        Type="User",
1106
                    )
1107
            if "LoggingConfig" in request:
1✔
1108
                logging_config = request["LoggingConfig"]
1✔
1109
                LOG.warning(
1✔
1110
                    "Advanced Lambda Logging Configuration is currently mocked "
1111
                    "and will not impact the logging behavior. "
1112
                    "Please create a feature request if needed."
1113
                )
1114

1115
                # when switching to JSON, app and system level log is auto set to INFO
1116
                if logging_config.get("LogFormat", None) == LogFormat.JSON:
1✔
1117
                    logging_config = {
1✔
1118
                        "ApplicationLogLevel": "INFO",
1119
                        "SystemLogLevel": "INFO",
1120
                        "LogGroup": f"/aws/lambda/{function_name}",
1121
                    } | logging_config
1122
                else:
UNCOV
1123
                    logging_config = (
×
1124
                        LoggingConfig(
1125
                            LogFormat=LogFormat.Text, LogGroup=f"/aws/lambda/{function_name}"
1126
                        )
1127
                        | logging_config
1128
                    )
1129

1130
            elif capacity_provider_config:
1✔
UNCOV
1131
                logging_config = LoggingConfig(
×
1132
                    LogFormat=LogFormat.JSON,
1133
                    LogGroup=f"/aws/lambda/{function_name}",
1134
                    ApplicationLogLevel="INFO",
1135
                    SystemLogLevel="INFO",
1136
                )
1137
            else:
1138
                logging_config = LoggingConfig(
1✔
1139
                    LogFormat=LogFormat.Text, LogGroup=f"/aws/lambda/{function_name}"
1140
                )
1141
            snap_start = (
1✔
1142
                None
1143
                if capacity_provider_config
1144
                else SnapStartResponse(
1145
                    ApplyOn=request.get("SnapStart", {}).get("ApplyOn", SnapStartApplyOn.None_),
1146
                    OptimizationStatus=SnapStartOptimizationStatus.Off,
1147
                )
1148
            )
1149
            version = FunctionVersion(
1✔
1150
                id=arn,
1151
                config=VersionFunctionConfiguration(
1152
                    last_modified=api_utils.format_lambda_date(datetime.datetime.now()),
1153
                    description=request.get("Description", ""),
1154
                    role=request["Role"],
1155
                    timeout=request.get("Timeout", LAMBDA_DEFAULT_TIMEOUT),
1156
                    runtime=request.get("Runtime"),
1157
                    memory_size=memory_size,
1158
                    handler=request.get("Handler"),
1159
                    package_type=package_type,
1160
                    environment=env_vars,
1161
                    architectures=request.get("Architectures") or [Architecture.x86_64],
1162
                    tracing_config_mode=request.get("TracingConfig", {}).get(
1163
                        "Mode", TracingMode.PassThrough
1164
                    ),
1165
                    image=image,
1166
                    image_config=image_config,
1167
                    code=code,
1168
                    layers=self.map_layers(layers),
1169
                    internal_revision=short_uid(),
1170
                    ephemeral_storage=LambdaEphemeralStorage(
1171
                        size=request.get("EphemeralStorage", {}).get("Size", 512)
1172
                    ),
1173
                    snap_start=snap_start,
1174
                    runtime_version_config=runtime_version_config,
1175
                    dead_letter_arn=request.get("DeadLetterConfig", {}).get("TargetArn"),
1176
                    vpc_config=self._build_vpc_config(
1177
                        context_account_id, context_region, request.get("VpcConfig")
1178
                    ),
1179
                    state=VersionState(
1180
                        state=State.Pending,
1181
                        code=StateReasonCode.Creating,
1182
                        reason="The function is being created.",
1183
                    ),
1184
                    logging_config=logging_config,
1185
                    # TODO: might need something like **optional_kwargs if None
1186
                    #   -> Test with regular GetFunction (i.e., without a capacity provider)
1187
                    capacity_provider_config=capacity_provider_config,
1188
                ),
1189
            )
1190
            version_post_response = None
1✔
1191
            if capacity_provider_config:
1✔
UNCOV
1192
                version_post_response = dataclasses.replace(
×
1193
                    version,
1194
                    config=dataclasses.replace(
1195
                        version.config,
1196
                        last_update=UpdateStatus(status=LastUpdateStatus.Successful),
1197
                        state=VersionState(state=State.ActiveNonInvocable),
1198
                    ),
1199
                )
1200
            fn.versions["$LATEST"] = version_post_response or version
1✔
1201
            state.functions[function_name] = fn
1✔
1202
        initialization_type = (
1✔
1203
            FunctionInitializationType.lambda_managed_instances
1204
            if capacity_provider_config
1205
            else FunctionInitializationType.on_demand
1206
        )
1207
        function_counter.labels(
1✔
1208
            operation=FunctionOperation.create,
1209
            runtime=runtime or "n/a",
1210
            status=FunctionStatus.success,
1211
            invocation_type="n/a",
1212
            package_type=package_type,
1213
            initialization_type=initialization_type,
1214
        )
1215
        # TODO: consider potential other side effects of not having a function version for $LATEST
1216
        # Provisioning happens upon publishing for functions using a capacity provider
1217
        if not capacity_provider_config:
1✔
1218
            self.lambda_service.create_function_version(version)
1✔
1219

1220
        if tags := request.get("Tags"):
1✔
1221
            # This will check whether the function exists.
1222
            self._store_tags(arn.unqualified_arn(), tags)
1✔
1223

1224
        if request.get("Publish"):
1✔
1225
            version = self._publish_version_with_changes(
1✔
1226
                function_name=function_name,
1227
                region=context_region,
1228
                account_id=context_account_id,
1229
                publish_to=request.get("PublishTo"),
1230
            )
1231

1232
        if config.LAMBDA_SYNCHRONOUS_CREATE:
1✔
1233
            # block via retrying until "terminal" condition reached before returning
UNCOV
1234
            if not poll_condition(
×
1235
                lambda: (
1236
                    get_function_version(
1237
                        function_name, version.id.qualifier, version.id.account, version.id.region
1238
                    ).config.state.state
1239
                    in [State.Active, State.ActiveNonInvocable, State.Failed]
1240
                ),
1241
                timeout=10,
1242
            ):
UNCOV
1243
                LOG.warning(
×
1244
                    "LAMBDA_SYNCHRONOUS_CREATE is active, but waiting for %s reached timeout.",
1245
                    function_name,
1246
                )
1247

1248
        return api_utils.map_config_out(
1✔
1249
            version, return_qualified_arn=False, return_update_status=False
1250
        )
1251

1252
    def _validate_runtime(self, package_type, runtime):
1✔
1253
        runtimes = ALL_RUNTIMES
1✔
1254
        if config.LAMBDA_RUNTIME_VALIDATION:
1✔
1255
            runtimes = list(itertools.chain(RUNTIMES_AGGREGATED.values()))
1✔
1256

1257
        if package_type == PackageType.Zip and runtime not in runtimes:
1✔
1258
            # deprecated runtimes have different error
1259
            if runtime in DEPRECATED_RUNTIMES:
1✔
1260
                HINT_LOG.info(
1✔
1261
                    "Set env variable LAMBDA_RUNTIME_VALIDATION to 0"
1262
                    " in order to allow usage of deprecated runtimes"
1263
                )
1264
                self._check_for_recomended_migration_target(runtime)
1✔
1265

1266
            raise InvalidParameterValueException(
1✔
1267
                f"Value {runtime} at 'runtime' failed to satisfy constraint: Member must satisfy enum value set: {VALID_RUNTIMES} or be a valid ARN",
1268
                Type="User",
1269
            )
1270

1271
    def _validate_managed_instances_runtime(self, runtime):
1✔
UNCOV
1272
        if runtime not in VALID_MANAGED_INSTANCE_RUNTIMES:
×
UNCOV
1273
            raise InvalidParameterValueException(
×
1274
                f"Runtime Enum {runtime} does not support specified feature: Lambda Managed Instances"
1275
            )
1276

1277
    def _check_for_recomended_migration_target(self, deprecated_runtime):
1✔
1278
        # AWS offers recommended runtime for migration for "newly" deprecated runtimes
1279
        # in order to preserve parity with error messages we need the code bellow
1280
        latest_runtime = DEPRECATED_RUNTIMES_UPGRADES.get(deprecated_runtime)
1✔
1281

1282
        if latest_runtime is not None:
1✔
1283
            LOG.debug(
1✔
1284
                "The Lambda runtime %s is deprecated. Please upgrade to a supported Lambda runtime such as %s.",
1285
                deprecated_runtime,
1286
                latest_runtime,
1287
            )
1288
            raise InvalidParameterValueException(
1✔
1289
                f"The runtime parameter of {deprecated_runtime} is no longer supported for creating or updating AWS Lambda functions. We recommend you use a supported runtime while creating or updating functions.",
1290
                Type="User",
1291
            )
1292

1293
    @handler(operation="UpdateFunctionConfiguration", expand=False)
1✔
1294
    def update_function_configuration(
1✔
1295
        self, context: RequestContext, request: UpdateFunctionConfigurationRequest
1296
    ) -> FunctionConfiguration:
1297
        """updates the $LATEST version of the function"""
1298
        function_name = request.get("FunctionName")
1✔
1299

1300
        # in case we got ARN or partial ARN
1301
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1302
        function_name, qualifier = api_utils.get_name_and_qualifier(function_name, None, context)
1✔
1303
        state = lambda_stores[account_id][region]
1✔
1304

1305
        if function_name not in state.functions:
1✔
UNCOV
1306
            raise ResourceNotFoundException(
×
1307
                f"Function not found: {api_utils.unqualified_lambda_arn(function_name=function_name, region=region, account=account_id)}",
1308
                Type="User",
1309
            )
1310
        function = state.functions[function_name]
1✔
1311

1312
        # TODO: lock modification of latest version
1313
        # TODO: notify service for changes relevant to re-provisioning of $LATEST
1314
        latest_version = function.latest()
1✔
1315
        latest_version_config = latest_version.config
1✔
1316

1317
        revision_id = request.get("RevisionId")
1✔
1318
        if revision_id and revision_id != latest_version.config.revision_id:
1✔
1319
            raise PreconditionFailedException(
1✔
1320
                "The Revision Id provided does not match the latest Revision Id. "
1321
                "Call the GetFunction/GetAlias API to retrieve the latest Revision Id",
1322
                Type="User",
1323
            )
1324

1325
        replace_kwargs = {}
1✔
1326
        if "EphemeralStorage" in request:
1✔
UNCOV
1327
            replace_kwargs["ephemeral_storage"] = LambdaEphemeralStorage(
×
1328
                request.get("EphemeralStorage", {}).get("Size", 512)
1329
            )  # TODO: do defaults here apply as well?
1330

1331
        if "Role" in request:
1✔
1332
            if not api_utils.is_role_arn(request["Role"]):
1✔
1333
                raise ValidationException(
1✔
1334
                    f"1 validation error detected: Value '{request.get('Role')}'"
1335
                    + " at 'role' failed to satisfy constraint: Member must satisfy regular expression pattern: arn:(aws[a-zA-Z-]*)?:iam::\\d{12}:role/?[a-zA-Z_0-9+=,.@\\-_/]+"
1336
                )
1337
            replace_kwargs["role"] = request["Role"]
1✔
1338

1339
        if "Description" in request:
1✔
1340
            replace_kwargs["description"] = request["Description"]
1✔
1341

1342
        if "Timeout" in request:
1✔
1343
            replace_kwargs["timeout"] = request["Timeout"]
1✔
1344

1345
        if "MemorySize" in request:
1✔
1346
            replace_kwargs["memory_size"] = request["MemorySize"]
1✔
1347

1348
        if "DeadLetterConfig" in request:
1✔
1349
            replace_kwargs["dead_letter_arn"] = request.get("DeadLetterConfig", {}).get("TargetArn")
1✔
1350

1351
        if vpc_config := request.get("VpcConfig"):
1✔
1352
            replace_kwargs["vpc_config"] = self._build_vpc_config(account_id, region, vpc_config)
1✔
1353

1354
        if "Handler" in request:
1✔
1355
            replace_kwargs["handler"] = request["Handler"]
1✔
1356

1357
        if "Runtime" in request:
1✔
1358
            runtime = request["Runtime"]
1✔
1359

1360
            if runtime not in ALL_RUNTIMES:
1✔
1361
                raise InvalidParameterValueException(
1✔
1362
                    f"Value {runtime} at 'runtime' failed to satisfy constraint: Member must satisfy enum value set: {VALID_RUNTIMES} or be a valid ARN",
1363
                    Type="User",
1364
                )
1365
            if runtime in DEPRECATED_RUNTIMES:
1✔
UNCOV
1366
                LOG.warning(
×
1367
                    "The Lambda runtime %s is deprecated. "
1368
                    "Please upgrade the runtime for the function %s: "
1369
                    "https://docs.aws.amazon.com/lambda/latest/dg/lambda-runtimes.html",
1370
                    runtime,
1371
                    function_name,
1372
                )
1373
            replace_kwargs["runtime"] = request["Runtime"]
1✔
1374

1375
        if snap_start := request.get("SnapStart"):
1✔
1376
            runtime = replace_kwargs.get("runtime") or latest_version_config.runtime
1✔
1377
            self._validate_snapstart(snap_start, runtime)
1✔
1378
            replace_kwargs["snap_start"] = SnapStartResponse(
1✔
1379
                ApplyOn=snap_start.get("ApplyOn", SnapStartApplyOn.None_),
1380
                OptimizationStatus=SnapStartOptimizationStatus.Off,
1381
            )
1382

1383
        if "Environment" in request:
1✔
1384
            if env_vars := request.get("Environment", {}).get("Variables", {}):
1✔
1385
                self._verify_env_variables(env_vars)
1✔
1386
            replace_kwargs["environment"] = env_vars
1✔
1387

1388
        if "Layers" in request:
1✔
1389
            new_layers = request["Layers"]
1✔
1390
            if new_layers:
1✔
1391
                self._validate_layers(new_layers, region=region, account_id=account_id)
1✔
1392
            replace_kwargs["layers"] = self.map_layers(new_layers)
1✔
1393

1394
        if "ImageConfig" in request:
1✔
1395
            new_image_config = request["ImageConfig"]
1✔
1396
            replace_kwargs["image_config"] = ImageConfig(
1✔
1397
                command=new_image_config.get("Command"),
1398
                entrypoint=new_image_config.get("EntryPoint"),
1399
                working_directory=new_image_config.get("WorkingDirectory"),
1400
            )
1401

1402
        if "LoggingConfig" in request:
1✔
1403
            logging_config = request["LoggingConfig"]
1✔
1404
            LOG.warning(
1✔
1405
                "Advanced Lambda Logging Configuration is currently mocked "
1406
                "and will not impact the logging behavior. "
1407
                "Please create a feature request if needed."
1408
            )
1409

1410
            # when switching to JSON, app and system level log is auto set to INFO
1411
            if logging_config.get("LogFormat", None) == LogFormat.JSON:
1✔
1412
                logging_config = {
1✔
1413
                    "ApplicationLogLevel": "INFO",
1414
                    "SystemLogLevel": "INFO",
1415
                } | logging_config
1416

1417
            last_config = latest_version_config.logging_config
1✔
1418

1419
            # add partial update
1420
            new_logging_config = last_config | logging_config
1✔
1421

1422
            # in case we switched from JSON to Text we need to remove LogLevel keys
1423
            if (
1✔
1424
                new_logging_config.get("LogFormat") == LogFormat.Text
1425
                and last_config.get("LogFormat") == LogFormat.JSON
1426
            ):
1427
                new_logging_config.pop("ApplicationLogLevel", None)
1✔
1428
                new_logging_config.pop("SystemLogLevel", None)
1✔
1429

1430
            replace_kwargs["logging_config"] = new_logging_config
1✔
1431

1432
        if "TracingConfig" in request:
1✔
UNCOV
1433
            new_mode = request.get("TracingConfig", {}).get("Mode")
×
1434
            if new_mode:
×
1435
                replace_kwargs["tracing_config_mode"] = new_mode
×
1436

1437
        if "CapacityProviderConfig" in request:
1✔
UNCOV
1438
            capacity_provider_config = request["CapacityProviderConfig"]
×
UNCOV
1439
            self._validate_capacity_provider_config(capacity_provider_config, context)
×
1440

UNCOV
1441
            if latest_version.config.capacity_provider_config and not request[
×
1442
                "CapacityProviderConfig"
1443
            ].get("LambdaManagedInstancesCapacityProviderConfig"):
1444
                raise ValidationException(
×
1445
                    "1 validation error detected: Value null at 'capacityProviderConfig.lambdaManagedInstancesCapacityProviderConfig' failed to satisfy constraint: Member must not be null"
1446
                )
UNCOV
1447
            if not latest_version.config.capacity_provider_config:
×
UNCOV
1448
                raise InvalidParameterValueException(
×
1449
                    "CapacityProviderConfig isn't supported for Lambda Default functions.",
1450
                    Type="User",
1451
                )
1452

UNCOV
1453
            default_config = CapacityProviderConfig(
×
1454
                LambdaManagedInstancesCapacityProviderConfig=LambdaManagedInstancesCapacityProviderConfig(
1455
                    ExecutionEnvironmentMemoryGiBPerVCpu=2.0,
1456
                    PerExecutionEnvironmentMaxConcurrency=16,
1457
                )
1458
            )
UNCOV
1459
            capacity_provider_config = merge_recursive(default_config, capacity_provider_config)
×
UNCOV
1460
            replace_kwargs["capacity_provider_config"] = capacity_provider_config
×
1461
        new_latest_version = dataclasses.replace(
1✔
1462
            latest_version,
1463
            config=dataclasses.replace(
1464
                latest_version_config,
1465
                last_modified=api_utils.generate_lambda_date(),
1466
                internal_revision=short_uid(),
1467
                last_update=UpdateStatus(
1468
                    status=LastUpdateStatus.InProgress,
1469
                    code="Creating",
1470
                    reason="The function is being created.",
1471
                ),
1472
                **replace_kwargs,
1473
            ),
1474
        )
1475
        function.versions["$LATEST"] = new_latest_version  # TODO: notify
1✔
1476
        self.lambda_service.update_version(new_version=new_latest_version)
1✔
1477

1478
        return api_utils.map_config_out(new_latest_version)
1✔
1479

1480
    @handler(operation="UpdateFunctionCode", expand=False)
1✔
1481
    def update_function_code(
1✔
1482
        self, context: RequestContext, request: UpdateFunctionCodeRequest
1483
    ) -> FunctionConfiguration:
1484
        """updates the $LATEST version of the function"""
1485
        # only supports normal zip packaging atm
1486
        # if request.get("Publish"):
1487
        #     self.lambda_service.create_function_version()
1488

1489
        function_name = request.get("FunctionName")
1✔
1490
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1491
        function_name, qualifier = api_utils.get_name_and_qualifier(function_name, None, context)
1✔
1492

1493
        store = lambda_stores[account_id][region]
1✔
1494
        if function_name not in store.functions:
1✔
UNCOV
1495
            raise ResourceNotFoundException(
×
1496
                f"Function not found: {api_utils.unqualified_lambda_arn(function_name=function_name, region=region, account=account_id)}",
1497
                Type="User",
1498
            )
1499
        function = store.functions[function_name]
1✔
1500

1501
        revision_id = request.get("RevisionId")
1✔
1502
        if revision_id and revision_id != function.latest().config.revision_id:
1✔
1503
            raise PreconditionFailedException(
1✔
1504
                "The Revision Id provided does not match the latest Revision Id. "
1505
                "Call the GetFunction/GetAlias API to retrieve the latest Revision Id",
1506
                Type="User",
1507
            )
1508

1509
        # TODO verify if correct combination of code is set
1510
        image = None
1✔
1511
        if (
1✔
1512
            request.get("ZipFile") or request.get("S3Bucket")
1513
        ) and function.latest().config.package_type == PackageType.Image:
1514
            raise InvalidParameterValueException(
1✔
1515
                "Please provide ImageUri when updating a function with packageType Image.",
1516
                Type="User",
1517
            )
1518
        elif request.get("ImageUri") and function.latest().config.package_type == PackageType.Zip:
1✔
1519
            raise InvalidParameterValueException(
1✔
1520
                "Please don't provide ImageUri when updating a function with packageType Zip.",
1521
                Type="User",
1522
            )
1523

1524
        if publish_to := request.get("PublishTo"):
1✔
UNCOV
1525
            self._validate_publish_to(publish_to)
×
1526

1527
        if zip_file := request.get("ZipFile"):
1✔
1528
            code = store_lambda_archive(
1✔
1529
                archive_file=zip_file,
1530
                function_name=function_name,
1531
                region_name=region,
1532
                account_id=account_id,
1533
            )
1534
        elif s3_bucket := request.get("S3Bucket"):
1✔
1535
            s3_key = request["S3Key"]
1✔
1536
            s3_object_version = request.get("S3ObjectVersion")
1✔
1537
            code = store_s3_bucket_archive(
1✔
1538
                archive_bucket=s3_bucket,
1539
                archive_key=s3_key,
1540
                archive_version=s3_object_version,
1541
                function_name=function_name,
1542
                region_name=region,
1543
                account_id=account_id,
1544
            )
1545
        elif image := request.get("ImageUri"):
1✔
1546
            code = None
1✔
1547
            image = create_image_code(image_uri=image)
1✔
1548
        else:
UNCOV
1549
            raise LambdaServiceException("A ZIP file, S3 bucket, or image is required")
×
1550

1551
        old_function_version = function.versions.get("$LATEST")
1✔
1552
        replace_kwargs = {"code": code} if code else {"image": image}
1✔
1553

1554
        if architectures := request.get("Architectures"):
1✔
UNCOV
1555
            if len(architectures) != 1:
×
UNCOV
1556
                raise ValidationException(
×
1557
                    f"1 validation error detected: Value '[{', '.join(architectures)}]' at 'architectures' failed to "
1558
                    f"satisfy constraint: Member must have length less than or equal to 1",
1559
                )
1560
            # An empty list of architectures is also forbidden. Further exceptions are tested here for create_function:
1561
            # tests.aws.services.lambda_.test_lambda_api.TestLambdaFunction.test_create_lambda_exceptions
UNCOV
1562
            if architectures[0] not in ARCHITECTURES:
×
UNCOV
1563
                raise ValidationException(
×
1564
                    f"1 validation error detected: Value '[{', '.join(architectures)}]' at 'architectures' failed to "
1565
                    f"satisfy constraint: Member must satisfy constraint: [Member must satisfy enum value set: "
1566
                    f"[x86_64, arm64], Member must not be null]",
1567
                )
UNCOV
1568
            replace_kwargs["architectures"] = architectures
×
1569

1570
        config = dataclasses.replace(
1✔
1571
            old_function_version.config,
1572
            internal_revision=short_uid(),
1573
            last_modified=api_utils.generate_lambda_date(),
1574
            last_update=UpdateStatus(
1575
                status=LastUpdateStatus.InProgress,
1576
                code="Creating",
1577
                reason="The function is being created.",
1578
            ),
1579
            **replace_kwargs,
1580
        )
1581
        function_version = dataclasses.replace(old_function_version, config=config)
1✔
1582
        function.versions["$LATEST"] = function_version
1✔
1583

1584
        self.lambda_service.update_version(new_version=function_version)
1✔
1585
        if request.get("Publish"):
1✔
1586
            function_version = self._publish_version_with_changes(
1✔
1587
                function_name=function_name,
1588
                region=region,
1589
                account_id=account_id,
1590
                publish_to=publish_to,
1591
                is_active=True,
1592
            )
1593
        return api_utils.map_config_out(
1✔
1594
            function_version, return_qualified_arn=bool(request.get("Publish"))
1595
        )
1596

1597
    # TODO: does deleting the latest published version affect the next versions number?
1598
    # TODO: what happens when we call this with a qualifier and a fully qualified ARN? (+ conflicts?)
1599
    # TODO: test different ARN patterns (shorthand ARN?)
1600
    # TODO: test deleting across regions?
1601
    # TODO: test mismatch between context region and region in ARN
1602
    # TODO: test qualifier $LATEST, alias-name and version
1603
    def delete_function(
1✔
1604
        self,
1605
        context: RequestContext,
1606
        function_name: NamespacedFunctionName,
1607
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
1608
        **kwargs,
1609
    ) -> DeleteFunctionResponse:
1610
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1611
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
1612
            function_name, qualifier, context
1613
        )
1614

1615
        if qualifier and api_utils.qualifier_is_alias(qualifier):
1✔
UNCOV
1616
            raise InvalidParameterValueException(
×
1617
                "Deletion of aliases is not currently supported.",
1618
                Type="User",
1619
            )
1620

1621
        store = lambda_stores[account_id][region]
1✔
1622
        if qualifier == "$LATEST":
1✔
1623
            raise InvalidParameterValueException(
1✔
1624
                "$LATEST version cannot be deleted without deleting the function.", Type="User"
1625
            )
1626

1627
        if function_name not in store.functions:
1✔
1628
            e = ResourceNotFoundException(
1✔
1629
                f"Function not found: {api_utils.unqualified_lambda_arn(function_name=function_name, region=region, account=account_id)}",
1630
                Type="User",
1631
            )
1632
            raise e
1✔
1633
        function = store.functions.get(function_name)
1✔
1634

1635
        function_has_capacity_provider = False
1✔
1636
        if qualifier:
1✔
1637
            # delete a version of the function
1638
            version = function.versions.pop(qualifier, None)
1✔
1639
            if version:
1✔
1640
                if version.config.capacity_provider_config:
1✔
UNCOV
1641
                    function_has_capacity_provider = True
×
1642
                self.lambda_service.stop_version(version.id.qualified_arn())
1✔
1643
                destroy_code_if_not_used(code=version.config.code, function=function)
1✔
1644
        else:
1645
            # delete the whole function
1646
            # TODO: introduce locking for safe deletion: We could create a new version at the API layer before
1647
            #  the old version gets cleaned up in the internal lambda service.
1648
            function = store.functions.pop(function_name)
1✔
1649
            for version in function.versions.values():
1✔
1650
                # Functions with a capacity provider do NOT have a version manager for $LATEST because only
1651
                # published versions are invokable.
1652
                if version.config.capacity_provider_config:
1✔
UNCOV
1653
                    function_has_capacity_provider = True
×
UNCOV
1654
                    if version.id.qualifier == "$LATEST":
×
UNCOV
1655
                        pass
×
1656
                else:
1657
                    self.lambda_service.stop_version(qualified_arn=version.id.qualified_arn())
1✔
1658
                # we can safely destroy the code here
1659
                if version.config.code:
1✔
1660
                    version.config.code.destroy()
1✔
1661

1662
        return DeleteFunctionResponse(StatusCode=202 if function_has_capacity_provider else 204)
1✔
1663

1664
    def list_functions(
1✔
1665
        self,
1666
        context: RequestContext,
1667
        master_region: MasterRegion = None,  # (only relevant for lambda@edge)
1668
        function_version: FunctionVersionApi = None,
1669
        marker: String = None,
1670
        max_items: MaxListItems = None,
1671
        **kwargs,
1672
    ) -> ListFunctionsResponse:
1673
        state = lambda_stores[context.account_id][context.region]
1✔
1674

1675
        if function_version and function_version != FunctionVersionApi.ALL:
1✔
1676
            raise ValidationException(
1✔
1677
                f"1 validation error detected: Value '{function_version}'"
1678
                + " at 'functionVersion' failed to satisfy constraint: Member must satisfy enum value set: [ALL]"
1679
            )
1680

1681
        if function_version == FunctionVersionApi.ALL:
1✔
1682
            # include all versions for all function
1683
            versions = [v for f in state.functions.values() for v in f.versions.values()]
1✔
1684
            return_qualified_arn = True
1✔
1685
        else:
1686
            versions = [f.latest() for f in state.functions.values()]
1✔
1687
            return_qualified_arn = False
1✔
1688

1689
        versions = [
1✔
1690
            api_utils.map_to_list_response(
1691
                api_utils.map_config_out(fc, return_qualified_arn=return_qualified_arn)
1692
            )
1693
            for fc in versions
1694
        ]
1695
        versions = PaginatedList(versions)
1✔
1696
        page, token = versions.get_page(
1✔
1697
            lambda version: version["FunctionArn"],
1698
            marker,
1699
            max_items,
1700
        )
1701
        return ListFunctionsResponse(Functions=page, NextMarker=token)
1✔
1702

1703
    def get_function(
1✔
1704
        self,
1705
        context: RequestContext,
1706
        function_name: NamespacedFunctionName,
1707
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
1708
        **kwargs,
1709
    ) -> GetFunctionResponse:
1710
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1711
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
1712
            function_name, qualifier, context
1713
        )
1714

1715
        fn = lambda_stores[account_id][region].functions.get(function_name)
1✔
1716
        if fn is None:
1✔
1717
            if qualifier is None:
1✔
1718
                raise ResourceNotFoundException(
1✔
1719
                    f"Function not found: {api_utils.unqualified_lambda_arn(function_name, account_id, region)}",
1720
                    Type="User",
1721
                )
1722
            else:
1723
                raise ResourceNotFoundException(
1✔
1724
                    f"Function not found: {api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)}",
1725
                    Type="User",
1726
                )
1727
        alias_name = None
1✔
1728
        if qualifier and api_utils.qualifier_is_alias(qualifier):
1✔
1729
            if qualifier not in fn.aliases:
1✔
1730
                alias_arn = api_utils.qualified_lambda_arn(
1✔
1731
                    function_name, qualifier, account_id, region
1732
                )
1733
                raise ResourceNotFoundException(f"Function not found: {alias_arn}", Type="User")
1✔
1734
            alias_name = qualifier
1✔
1735
            qualifier = fn.aliases[alias_name].function_version
1✔
1736

1737
        version = get_function_version(
1✔
1738
            function_name=function_name,
1739
            qualifier=qualifier,
1740
            account_id=account_id,
1741
            region=region,
1742
        )
1743
        tags = self._get_tags(api_utils.unqualified_lambda_arn(function_name, account_id, region))
1✔
1744
        additional_fields = {}
1✔
1745
        if tags:
1✔
1746
            additional_fields["Tags"] = tags
1✔
1747
        code_location = None
1✔
1748
        if code := version.config.code:
1✔
1749
            code_location = FunctionCodeLocation(
1✔
1750
                Location=code.generate_presigned_url(endpoint_url=config.external_service_url()),
1751
                RepositoryType="S3",
1752
            )
1753
        elif image := version.config.image:
1✔
1754
            code_location = FunctionCodeLocation(
1✔
1755
                ImageUri=image.image_uri,
1756
                RepositoryType=image.repository_type,
1757
                ResolvedImageUri=image.resolved_image_uri,
1758
            )
1759
        concurrency = None
1✔
1760
        if fn.reserved_concurrent_executions:
1✔
1761
            concurrency = Concurrency(
1✔
1762
                ReservedConcurrentExecutions=fn.reserved_concurrent_executions
1763
            )
1764

1765
        return GetFunctionResponse(
1✔
1766
            Configuration=api_utils.map_config_out(
1767
                version, return_qualified_arn=bool(qualifier), alias_name=alias_name
1768
            ),
1769
            Code=code_location,  # TODO
1770
            Concurrency=concurrency,
1771
            **additional_fields,
1772
        )
1773

1774
    def get_function_configuration(
1✔
1775
        self,
1776
        context: RequestContext,
1777
        function_name: NamespacedFunctionName,
1778
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
1779
        **kwargs,
1780
    ) -> FunctionConfiguration:
1781
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1782
        # CAVE: THIS RETURN VALUE IS *NOT* THE SAME AS IN get_function (!) but seems to be only configuration part?
1783
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
1784
            function_name, qualifier, context
1785
        )
1786
        version = get_function_version(
1✔
1787
            function_name=function_name,
1788
            qualifier=qualifier,
1789
            account_id=account_id,
1790
            region=region,
1791
        )
1792
        return api_utils.map_config_out(version, return_qualified_arn=bool(qualifier))
1✔
1793

1794
    def invoke(
1✔
1795
        self,
1796
        context: RequestContext,
1797
        function_name: NamespacedFunctionName,
1798
        invocation_type: InvocationType | None = None,
1799
        log_type: LogType | None = None,
1800
        client_context: String | None = None,
1801
        durable_execution_name: DurableExecutionName | None = None,
1802
        payload: IO[Blob] | None = None,
1803
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
1804
        tenant_id: TenantId | None = None,
1805
        **kwargs,
1806
    ) -> InvocationResponse:
1807
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1808
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
1809
            function_name, qualifier, context
1810
        )
1811

1812
        user_agent = context.request.user_agent.string
1✔
1813

1814
        time_before = time.perf_counter()
1✔
1815
        try:
1✔
1816
            invocation_result = self.lambda_service.invoke(
1✔
1817
                function_name=function_name,
1818
                qualifier=qualifier,
1819
                region=region,
1820
                account_id=account_id,
1821
                invocation_type=invocation_type,
1822
                client_context=client_context,
1823
                request_id=context.request_id,
1824
                trace_context=context.trace_context,
1825
                payload=payload.read() if payload else None,
1826
                user_agent=user_agent,
1827
            )
1828
        except ServiceException:
1✔
1829
            raise
1✔
1830
        except EnvironmentStartupTimeoutException as e:
1✔
1831
            raise LambdaServiceException(
1✔
1832
                f"[{context.request_id}] Timeout while starting up lambda environment for function {function_name}:{qualifier}"
1833
            ) from e
1834
        except Exception as e:
1✔
1835
            LOG.error(
1✔
1836
                "[%s] Error while invoking lambda %s",
1837
                context.request_id,
1838
                function_name,
1839
                exc_info=LOG.isEnabledFor(logging.DEBUG),
1840
            )
1841
            raise LambdaServiceException(
1✔
1842
                f"[{context.request_id}] Internal error while executing lambda {function_name}:{qualifier}. Caused by {type(e).__name__}: {e}"
1843
            ) from e
1844

1845
        if invocation_type == InvocationType.Event:
1✔
1846
            # This happens when invocation type is event
1847
            return InvocationResponse(StatusCode=202)
1✔
1848
        if invocation_type == InvocationType.DryRun:
1✔
1849
            # This happens when invocation type is dryrun
1850
            return InvocationResponse(StatusCode=204)
1✔
1851
        LOG.debug("Lambda invocation duration: %0.2fms", (time.perf_counter() - time_before) * 1000)
1✔
1852

1853
        response = InvocationResponse(
1✔
1854
            StatusCode=200,
1855
            Payload=invocation_result.payload,
1856
            ExecutedVersion=invocation_result.executed_version,
1857
        )
1858

1859
        if invocation_result.is_error:
1✔
1860
            response["FunctionError"] = "Unhandled"
1✔
1861

1862
        if log_type == LogType.Tail:
1✔
1863
            response["LogResult"] = to_str(
1✔
1864
                base64.b64encode(to_bytes(invocation_result.logs)[-4096:])
1865
            )
1866

1867
        return response
1✔
1868

1869
    # Version operations
1870
    def publish_version(
1✔
1871
        self,
1872
        context: RequestContext,
1873
        function_name: FunctionName,
1874
        code_sha256: String | None = None,
1875
        description: Description | None = None,
1876
        revision_id: String | None = None,
1877
        publish_to: FunctionVersionLatestPublished | None = None,
1878
        **kwargs,
1879
    ) -> FunctionConfiguration:
1880
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1881
        function_name = api_utils.get_function_name(function_name, context)
1✔
1882
        if publish_to:
1✔
UNCOV
1883
            self._validate_publish_to(publish_to)
×
1884
        new_version = self._publish_version_from_existing_version(
1✔
1885
            function_name=function_name,
1886
            description=description,
1887
            account_id=account_id,
1888
            region=region,
1889
            revision_id=revision_id,
1890
            code_sha256=code_sha256,
1891
            publish_to=publish_to,
1892
        )
1893
        return api_utils.map_config_out(new_version, return_qualified_arn=True)
1✔
1894

1895
    def list_versions_by_function(
1✔
1896
        self,
1897
        context: RequestContext,
1898
        function_name: NamespacedFunctionName,
1899
        marker: String = None,
1900
        max_items: MaxListItems = None,
1901
        **kwargs,
1902
    ) -> ListVersionsByFunctionResponse:
1903
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1904
        function_name = api_utils.get_function_name(function_name, context)
1✔
1905
        function = self._get_function(
1✔
1906
            function_name=function_name, region=region, account_id=account_id
1907
        )
1908
        versions = [
1✔
1909
            api_utils.map_to_list_response(
1910
                api_utils.map_config_out(version=version, return_qualified_arn=True)
1911
            )
1912
            for version in function.versions.values()
1913
        ]
1914
        items = PaginatedList(versions)
1✔
1915
        page, token = items.get_page(
1✔
1916
            lambda item: item,
1917
            marker,
1918
            max_items,
1919
        )
1920
        return ListVersionsByFunctionResponse(Versions=page, NextMarker=token)
1✔
1921

1922
    # Alias
1923

1924
    def _create_routing_config_model(
1✔
1925
        self, routing_config_dict: dict[str, float], function_version: FunctionVersion
1926
    ):
1927
        if len(routing_config_dict) > 1:
1✔
1928
            raise InvalidParameterValueException(
1✔
1929
                "Number of items in AdditionalVersionWeights cannot be greater than 1",
1930
                Type="User",
1931
            )
1932
        # should be exactly one item here, still iterating, might be supported in the future
1933
        for key, value in routing_config_dict.items():
1✔
1934
            if value < 0.0 or value >= 1.0:
1✔
1935
                raise ValidationException(
1✔
1936
                    f"1 validation error detected: Value '{{{key}={value}}}' at 'routingConfig.additionalVersionWeights' failed to satisfy constraint: Map value must satisfy constraint: [Member must have value less than or equal to 1.0, Member must have value greater than or equal to 0.0, Member must not be null]"
1937
                )
1938
            if key == function_version.id.qualifier:
1✔
1939
                raise InvalidParameterValueException(
1✔
1940
                    f"Invalid function version {function_version.id.qualifier}. Function version {function_version.id.qualifier} is already included in routing configuration.",
1941
                    Type="User",
1942
                )
1943
            # check if version target is latest, then no routing config is allowed
1944
            if function_version.id.qualifier == "$LATEST":
1✔
1945
                raise InvalidParameterValueException(
1✔
1946
                    "$LATEST is not supported for an alias pointing to more than 1 version"
1947
                )
1948
            if not api_utils.qualifier_is_version(key):
1✔
1949
                raise ValidationException(
1✔
1950
                    f"1 validation error detected: Value '{{{key}={value}}}' at 'routingConfig.additionalVersionWeights' failed to satisfy constraint: Map keys must satisfy constraint: [Member must have length less than or equal to 1024, Member must have length greater than or equal to 1, Member must satisfy regular expression pattern: [0-9]+]"
1951
                )
1952

1953
            # checking if the version in the config exists
1954
            get_function_version(
1✔
1955
                function_name=function_version.id.function_name,
1956
                qualifier=key,
1957
                region=function_version.id.region,
1958
                account_id=function_version.id.account,
1959
            )
1960
        return AliasRoutingConfig(version_weights=routing_config_dict)
1✔
1961

1962
    def create_alias(
1✔
1963
        self,
1964
        context: RequestContext,
1965
        function_name: FunctionName,
1966
        name: Alias,
1967
        function_version: VersionWithLatestPublished,
1968
        description: Description = None,
1969
        routing_config: AliasRoutingConfiguration = None,
1970
        **kwargs,
1971
    ) -> AliasConfiguration:
1972
        if not api_utils.qualifier_is_alias(name):
1✔
1973
            raise ValidationException(
1✔
1974
                f"1 validation error detected: Value '{name}' at 'name' failed to satisfy constraint: Member must satisfy regular expression pattern: (?!^[0-9]+$)([a-zA-Z0-9-_]+)"
1975
            )
1976

1977
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
1978
        function_name = api_utils.get_function_name(function_name, context)
1✔
1979
        target_version = get_function_version(
1✔
1980
            function_name=function_name,
1981
            qualifier=function_version,
1982
            region=region,
1983
            account_id=account_id,
1984
        )
1985
        function = self._get_function(
1✔
1986
            function_name=function_name, region=region, account_id=account_id
1987
        )
1988
        # description is always present, if not specified it's an empty string
1989
        description = description or ""
1✔
1990
        with function.lock:
1✔
1991
            if existing_alias := function.aliases.get(name):
1✔
1992
                raise ResourceConflictException(
1✔
1993
                    f"Alias already exists: {api_utils.map_alias_out(alias=existing_alias, function=function)['AliasArn']}",
1994
                    Type="User",
1995
                )
1996
            # checking if the version exists
1997
            routing_configuration = None
1✔
1998
            if routing_config and (
1✔
1999
                routing_config_dict := routing_config.get("AdditionalVersionWeights")
2000
            ):
2001
                routing_configuration = self._create_routing_config_model(
1✔
2002
                    routing_config_dict, target_version
2003
                )
2004

2005
            alias = VersionAlias(
1✔
2006
                name=name,
2007
                function_version=function_version,
2008
                description=description,
2009
                routing_configuration=routing_configuration,
2010
            )
2011
            function.aliases[name] = alias
1✔
2012
        return api_utils.map_alias_out(alias=alias, function=function)
1✔
2013

2014
    def list_aliases(
1✔
2015
        self,
2016
        context: RequestContext,
2017
        function_name: FunctionName,
2018
        function_version: VersionWithLatestPublished = None,
2019
        marker: String = None,
2020
        max_items: MaxListItems = None,
2021
        **kwargs,
2022
    ) -> ListAliasesResponse:
2023
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2024
        function_name = api_utils.get_function_name(function_name, context)
1✔
2025
        function = self._get_function(
1✔
2026
            function_name=function_name, region=region, account_id=account_id
2027
        )
2028
        aliases = [
1✔
2029
            api_utils.map_alias_out(alias, function)
2030
            for alias in function.aliases.values()
2031
            if function_version is None or alias.function_version == function_version
2032
        ]
2033

2034
        aliases = PaginatedList(aliases)
1✔
2035
        page, token = aliases.get_page(
1✔
2036
            lambda alias: alias["AliasArn"],
2037
            marker,
2038
            max_items,
2039
        )
2040

2041
        return ListAliasesResponse(Aliases=page, NextMarker=token)
1✔
2042

2043
    def delete_alias(
1✔
2044
        self, context: RequestContext, function_name: FunctionName, name: Alias, **kwargs
2045
    ) -> None:
2046
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2047
        function_name = api_utils.get_function_name(function_name, context)
1✔
2048
        function = self._get_function(
1✔
2049
            function_name=function_name, region=region, account_id=account_id
2050
        )
2051
        version_alias = function.aliases.pop(name, None)
1✔
2052

2053
        # cleanup related resources
2054
        if name in function.provisioned_concurrency_configs:
1✔
2055
            function.provisioned_concurrency_configs.pop(name)
1✔
2056

2057
        # TODO: Allow for deactivating/unregistering specific Lambda URLs
2058
        if version_alias and name in function.function_url_configs:
1✔
2059
            url_config = function.function_url_configs.pop(name)
1✔
2060
            LOG.debug(
1✔
2061
                "Stopping aliased Lambda Function URL %s for %s",
2062
                url_config.url,
2063
                url_config.function_name,
2064
            )
2065

2066
    def get_alias(
1✔
2067
        self, context: RequestContext, function_name: FunctionName, name: Alias, **kwargs
2068
    ) -> AliasConfiguration:
2069
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2070
        function_name = api_utils.get_function_name(function_name, context)
1✔
2071
        function = self._get_function(
1✔
2072
            function_name=function_name, region=region, account_id=account_id
2073
        )
2074
        if not (alias := function.aliases.get(name)):
1✔
2075
            raise ResourceNotFoundException(
1✔
2076
                f"Cannot find alias arn: {api_utils.qualified_lambda_arn(function_name=function_name, qualifier=name, region=region, account=account_id)}",
2077
                Type="User",
2078
            )
2079
        return api_utils.map_alias_out(alias=alias, function=function)
1✔
2080

2081
    def update_alias(
1✔
2082
        self,
2083
        context: RequestContext,
2084
        function_name: FunctionName,
2085
        name: Alias,
2086
        function_version: VersionWithLatestPublished = None,
2087
        description: Description = None,
2088
        routing_config: AliasRoutingConfiguration = None,
2089
        revision_id: String = None,
2090
        **kwargs,
2091
    ) -> AliasConfiguration:
2092
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2093
        function_name = api_utils.get_function_name(function_name, context)
1✔
2094
        function = self._get_function(
1✔
2095
            function_name=function_name, region=region, account_id=account_id
2096
        )
2097
        if not (alias := function.aliases.get(name)):
1✔
2098
            fn_arn = api_utils.qualified_lambda_arn(function_name, name, account_id, region)
1✔
2099
            raise ResourceNotFoundException(
1✔
2100
                f"Alias not found: {fn_arn}",
2101
                Type="User",
2102
            )
2103
        if revision_id and alias.revision_id != revision_id:
1✔
2104
            raise PreconditionFailedException(
1✔
2105
                "The Revision Id provided does not match the latest Revision Id. "
2106
                "Call the GetFunction/GetAlias API to retrieve the latest Revision Id",
2107
                Type="User",
2108
            )
2109
        changes = {}
1✔
2110
        if function_version is not None:
1✔
2111
            changes |= {"function_version": function_version}
1✔
2112
        if description is not None:
1✔
2113
            changes |= {"description": description}
1✔
2114
        if routing_config is not None:
1✔
2115
            # if it is an empty dict or AdditionalVersionWeights is empty, set routing config to None
2116
            new_routing_config = None
1✔
2117
            if routing_config_dict := routing_config.get("AdditionalVersionWeights"):
1✔
UNCOV
2118
                new_routing_config = self._create_routing_config_model(routing_config_dict)
×
2119
            changes |= {"routing_configuration": new_routing_config}
1✔
2120
        # even if no changes are done, we have to update revision id for some reason
2121
        old_alias = alias
1✔
2122
        alias = dataclasses.replace(alias, **changes)
1✔
2123
        function.aliases[name] = alias
1✔
2124

2125
        # TODO: signal lambda service that pointer potentially changed
2126
        self.lambda_service.update_alias(old_alias=old_alias, new_alias=alias, function=function)
1✔
2127

2128
        return api_utils.map_alias_out(alias=alias, function=function)
1✔
2129

2130
    # =======================================
2131
    # ======= EVENT SOURCE MAPPINGS =========
2132
    # =======================================
2133
    def check_service_resource_exists(
1✔
2134
        self, service: str, resource_arn: str, function_arn: str, function_role_arn: str
2135
    ):
2136
        """
2137
        Check if the service resource exists and if the function has access to it.
2138

2139
        Raises:
2140
            InvalidParameterValueException: If the service resource does not exist or the function does not have access to it.
2141
        """
2142
        arn = parse_arn(resource_arn)
1✔
2143
        source_client = get_internal_client(
1✔
2144
            arn=resource_arn,
2145
            role_arn=function_role_arn,
2146
            service_principal=ServicePrincipal.lambda_,
2147
            source_arn=function_arn,
2148
        )
2149
        if service in ["sqs", "sqs-fifo"]:
1✔
2150
            try:
1✔
2151
                # AWS uses `GetQueueAttributes` internally to verify the queue existence, but we need the `QueueUrl`
2152
                # which is not given directly. We build out a dummy `QueueUrl` which can be parsed by SQS to return
2153
                # the right value
2154
                queue_name = arn["resource"].split("/")[-1]
1✔
2155
                queue_url = f"http://sqs.{arn['region']}.domain/{arn['account']}/{queue_name}"
1✔
2156
                source_client.get_queue_attributes(QueueUrl=queue_url)
1✔
2157
            except ClientError as e:
1✔
2158
                error_code = e.response["Error"]["Code"]
1✔
2159
                if error_code == "AWS.SimpleQueueService.NonExistentQueue":
1✔
2160
                    raise InvalidParameterValueException(
1✔
2161
                        f"Error occurred while ReceiveMessage. SQS Error Code: {error_code}. SQS Error Message: {e.response['Error']['Message']}",
2162
                        Type="User",
2163
                    )
UNCOV
2164
                raise e
×
2165
        elif service in ["kinesis"]:
1✔
2166
            try:
1✔
2167
                source_client.describe_stream(StreamARN=resource_arn)
1✔
2168
            except ClientError as e:
1✔
2169
                if e.response["Error"]["Code"] == "ResourceNotFoundException":
1✔
2170
                    raise InvalidParameterValueException(
1✔
2171
                        f"Stream not found: {resource_arn}",
2172
                        Type="User",
2173
                    )
UNCOV
2174
                raise e
×
2175
        elif service in ["dynamodb"]:
1✔
2176
            try:
1✔
2177
                source_client.describe_stream(StreamArn=resource_arn)
1✔
2178
            except ClientError as e:
1✔
2179
                if e.response["Error"]["Code"] == "ResourceNotFoundException":
1✔
2180
                    raise InvalidParameterValueException(
1✔
2181
                        f"Stream not found: {resource_arn}",
2182
                        Type="User",
2183
                    )
UNCOV
2184
                raise e
×
2185

2186
    @handler("CreateEventSourceMapping", expand=False)
1✔
2187
    def create_event_source_mapping(
1✔
2188
        self,
2189
        context: RequestContext,
2190
        request: CreateEventSourceMappingRequest,
2191
    ) -> EventSourceMappingConfiguration:
2192
        return self.create_event_source_mapping_v2(context, request)
1✔
2193

2194
    def create_event_source_mapping_v2(
1✔
2195
        self,
2196
        context: RequestContext,
2197
        request: CreateEventSourceMappingRequest,
2198
    ) -> EventSourceMappingConfiguration:
2199
        # Validations
2200
        function_arn, function_name, state, function_version, function_role = (
1✔
2201
            self.validate_event_source_mapping(context, request)
2202
        )
2203

2204
        esm_config = EsmConfigFactory(request, context, function_arn).get_esm_config()
1✔
2205

2206
        # Copy esm_config to avoid a race condition with potential async update in the store
2207
        state.event_source_mappings[esm_config["UUID"]] = esm_config.copy()
1✔
2208
        enabled = request.get("Enabled", True)
1✔
2209
        # TODO: check for potential async race condition update -> think about locking
2210
        esm_worker = EsmWorkerFactory(esm_config, function_role, enabled).get_esm_worker()
1✔
2211
        self.esm_workers[esm_worker.uuid] = esm_worker
1✔
2212
        # TODO: check StateTransitionReason, LastModified, LastProcessingResult (concurrent updates requires locking!)
2213
        if tags := request.get("Tags"):
1✔
2214
            self._store_tags(esm_config.get("EventSourceMappingArn"), tags)
1✔
2215
        esm_worker.create()
1✔
2216
        return esm_config
1✔
2217

2218
    def validate_event_source_mapping(self, context, request):
1✔
2219
        # TODO: test whether stream ARNs are valid sources for Pipes or ESM or whether only DynamoDB table ARNs work
2220
        # TODO: Validate MaxRecordAgeInSeconds (i.e cannot subceed 60s but can be -1) and MaxRetryAttempts parameters.
2221
        # See https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-maximumrecordageinseconds
2222
        is_create_esm_request = context.operation.name == self.create_event_source_mapping.operation
1✔
2223

2224
        if destination_config := request.get("DestinationConfig"):
1✔
2225
            if "OnSuccess" in destination_config:
1✔
2226
                raise InvalidParameterValueException(
1✔
2227
                    "Unsupported DestinationConfig parameter for given event source mapping type.",
2228
                    Type="User",
2229
                )
2230

2231
        service = None
1✔
2232
        if "SelfManagedEventSource" in request:
1✔
UNCOV
2233
            service = "kafka"
×
UNCOV
2234
            if "SourceAccessConfigurations" not in request:
×
UNCOV
2235
                raise InvalidParameterValueException(
×
2236
                    "Required 'sourceAccessConfigurations' parameter is missing.", Type="User"
2237
                )
2238
        if service is None and "EventSourceArn" not in request:
1✔
2239
            raise InvalidParameterValueException("Unrecognized event source.", Type="User")
1✔
2240
        if service is None:
1✔
2241
            service = extract_service_from_arn(request["EventSourceArn"])
1✔
2242

2243
        batch_size = api_utils.validate_and_set_batch_size(service, request.get("BatchSize"))
1✔
2244
        if service in ["dynamodb", "kinesis"]:
1✔
2245
            starting_position = request.get("StartingPosition")
1✔
2246
            if not starting_position:
1✔
2247
                raise InvalidParameterValueException(
1✔
2248
                    "1 validation error detected: Value null at 'startingPosition' failed to satisfy constraint: Member must not be null.",
2249
                    Type="User",
2250
                )
2251

2252
            if starting_position not in KinesisStreamStartPosition.__members__:
1✔
2253
                raise ValidationException(
1✔
2254
                    f"1 validation error detected: Value '{starting_position}' at 'startingPosition' failed to satisfy constraint: Member must satisfy enum value set: [LATEST, AT_TIMESTAMP, TRIM_HORIZON]"
2255
                )
2256
            # AT_TIMESTAMP is not allowed for DynamoDB Streams
2257
            elif (
1✔
2258
                service == "dynamodb"
2259
                and starting_position not in DynamoDBStreamStartPosition.__members__
2260
            ):
2261
                raise InvalidParameterValueException(
1✔
2262
                    f"Unsupported starting position for arn type: {request['EventSourceArn']}",
2263
                    Type="User",
2264
                )
2265

2266
        if service in ["sqs", "sqs-fifo"]:
1✔
2267
            if batch_size > 10 and request.get("MaximumBatchingWindowInSeconds", 0) == 0:
1✔
2268
                raise InvalidParameterValueException(
1✔
2269
                    "Maximum batch window in seconds must be greater than 0 if maximum batch size is greater than 10",
2270
                    Type="User",
2271
                )
2272

2273
        if (filter_criteria := request.get("FilterCriteria")) is not None:
1✔
2274
            for filter_ in filter_criteria.get("Filters", []):
1✔
2275
                pattern_str = filter_.get("Pattern")
1✔
2276
                if not pattern_str or not isinstance(pattern_str, str):
1✔
UNCOV
2277
                    raise InvalidParameterValueException(
×
2278
                        "Invalid filter pattern definition.", Type="User"
2279
                    )
2280

2281
                if not validate_event_pattern(pattern_str):
1✔
2282
                    raise InvalidParameterValueException(
1✔
2283
                        "Invalid filter pattern definition.", Type="User"
2284
                    )
2285

2286
        # Can either have a FunctionName (i.e CreateEventSourceMapping request) or
2287
        # an internal EventSourceMappingConfiguration representation
2288
        request_function_name = request.get("FunctionName") or request.get("FunctionArn")
1✔
2289
        # can be either a partial arn or a full arn for the version/alias
2290
        function_name, qualifier, account, region = function_locators_from_arn(
1✔
2291
            request_function_name
2292
        )
2293
        # TODO: validate `context.region` vs. `region(request["FunctionName"])` vs. `region(request["EventSourceArn"])`
2294
        account = account or context.account_id
1✔
2295
        region = region or context.region
1✔
2296
        state = lambda_stores[account][region]
1✔
2297
        fn = state.functions.get(function_name)
1✔
2298
        if not fn:
1✔
2299
            raise InvalidParameterValueException("Function does not exist", Type="User")
1✔
2300

2301
        if qualifier:
1✔
2302
            # make sure the function version/alias exists
2303
            if api_utils.qualifier_is_alias(qualifier):
1✔
2304
                fn_alias = fn.aliases.get(qualifier)
1✔
2305
                if not fn_alias:
1✔
2306
                    raise Exception("unknown alias")  # TODO: cover via test
×
2307
            elif api_utils.qualifier_is_version(qualifier):
1✔
2308
                fn_version = fn.versions.get(qualifier)
1✔
2309
                if not fn_version:
1✔
2310
                    raise Exception("unknown version")  # TODO: cover via test
×
2311
            elif qualifier == "$LATEST":
1✔
2312
                pass
1✔
2313
            elif qualifier == "$LATEST.PUBLISHED":
×
UNCOV
2314
                if fn.versions.get(qualifier):
×
UNCOV
2315
                    pass
×
2316
            else:
UNCOV
2317
                raise Exception("invalid functionname")  # TODO: cover via test
×
2318
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account, region)
1✔
2319

2320
        else:
2321
            fn_arn = api_utils.unqualified_lambda_arn(function_name, account, region)
1✔
2322

2323
        function_version = get_function_version_from_arn(fn_arn)
1✔
2324
        function_role = function_version.config.role
1✔
2325

2326
        if source_arn := request.get("EventSourceArn"):
1✔
2327
            self.check_service_resource_exists(service, source_arn, fn_arn, function_role)
1✔
2328
        # Check we are validating a CreateEventSourceMapping request
2329
        if is_create_esm_request:
1✔
2330

2331
            def _get_mapping_sources(mapping: dict[str, Any]) -> list[str]:
1✔
2332
                if event_source_arn := mapping.get("EventSourceArn"):
1✔
2333
                    return [event_source_arn]
1✔
UNCOV
2334
                return (
×
2335
                    mapping.get("SelfManagedEventSource", {})
2336
                    .get("Endpoints", {})
2337
                    .get("KAFKA_BOOTSTRAP_SERVERS", [])
2338
                )
2339

2340
            # check for event source duplicates
2341
            # TODO: currently validated for sqs, kinesis, and dynamodb
2342
            service_id = load_service(service).service_id
1✔
2343
            for uuid, mapping in state.event_source_mappings.items():
1✔
2344
                mapping_sources = _get_mapping_sources(mapping)
1✔
2345
                request_sources = _get_mapping_sources(request)
1✔
2346
                if mapping["FunctionArn"] == fn_arn and (
1✔
2347
                    set(mapping_sources).intersection(request_sources)
2348
                ):
2349
                    if service == "sqs":
1✔
2350
                        # *shakes fist at SQS*
2351
                        raise ResourceConflictException(
1✔
2352
                            f'An event source mapping with {service_id} arn (" {mapping["EventSourceArn"]} ") '
2353
                            f'and function (" {function_name} ") already exists. Please update or delete the '
2354
                            f"existing mapping with UUID {uuid}",
2355
                            Type="User",
2356
                        )
2357
                    elif service == "kafka":
1✔
UNCOV
2358
                        if set(mapping["Topics"]).intersection(request["Topics"]):
×
UNCOV
2359
                            raise ResourceConflictException(
×
2360
                                f'An event source mapping with event source ("{",".join(request_sources)}"), '
2361
                                f'function ("{fn_arn}"), '
2362
                                f'topics ("{",".join(request["Topics"])}") already exists. Please update or delete the '
2363
                                f"existing mapping with UUID {uuid}",
2364
                                Type="User",
2365
                            )
2366
                    else:
2367
                        raise ResourceConflictException(
1✔
2368
                            f'The event source arn (" {mapping["EventSourceArn"]} ") and function '
2369
                            f'(" {function_name} ") provided mapping already exists. Please update or delete the '
2370
                            f"existing mapping with UUID {uuid}",
2371
                            Type="User",
2372
                        )
2373
        return fn_arn, function_name, state, function_version, function_role
1✔
2374

2375
    @handler("UpdateEventSourceMapping", expand=False)
1✔
2376
    def update_event_source_mapping(
1✔
2377
        self,
2378
        context: RequestContext,
2379
        request: UpdateEventSourceMappingRequest,
2380
    ) -> EventSourceMappingConfiguration:
2381
        return self.update_event_source_mapping_v2(context, request)
1✔
2382

2383
    def update_event_source_mapping_v2(
1✔
2384
        self,
2385
        context: RequestContext,
2386
        request: UpdateEventSourceMappingRequest,
2387
    ) -> EventSourceMappingConfiguration:
2388
        # TODO: test and implement this properly (quite complex with many validations and limitations!)
2389
        LOG.warning(
1✔
2390
            "Updating Lambda Event Source Mapping is in experimental state and not yet fully tested."
2391
        )
2392
        state = lambda_stores[context.account_id][context.region]
1✔
2393
        request_data = {**request}
1✔
2394
        uuid = request_data.pop("UUID", None)
1✔
2395
        if not uuid:
1✔
UNCOV
2396
            raise ResourceNotFoundException(
×
2397
                "The resource you requested does not exist.", Type="User"
2398
            )
2399
        old_event_source_mapping = state.event_source_mappings.get(uuid)
1✔
2400
        esm_worker = self.esm_workers.get(uuid)
1✔
2401
        if old_event_source_mapping is None or esm_worker is None:
1✔
2402
            raise ResourceNotFoundException(
1✔
2403
                "The resource you requested does not exist.", Type="User"
2404
            )  # TODO: test?
2405

2406
        # normalize values to overwrite
2407
        event_source_mapping = old_event_source_mapping | request_data
1✔
2408

2409
        temp_params = {}  # values only set for the returned response, not saved internally (e.g. transient state)
1✔
2410

2411
        # Validate the newly updated ESM object. We ignore the output here since we only care whether an Exception is raised.
2412
        function_arn, _, _, function_version, function_role = self.validate_event_source_mapping(
1✔
2413
            context, event_source_mapping
2414
        )
2415

2416
        # remove the FunctionName field
2417
        event_source_mapping.pop("FunctionName", None)
1✔
2418

2419
        if function_arn:
1✔
2420
            event_source_mapping["FunctionArn"] = function_arn
1✔
2421

2422
        # Only apply update if the desired state differs
2423
        enabled = request.get("Enabled")
1✔
2424
        if enabled is not None:
1✔
2425
            if enabled and old_event_source_mapping["State"] != EsmState.ENABLED:
1✔
2426
                event_source_mapping["State"] = EsmState.ENABLING
1✔
2427
            # TODO: What happens when trying to update during an update or failed state?!
2428
            elif not enabled and old_event_source_mapping["State"] == EsmState.ENABLED:
1✔
2429
                event_source_mapping["State"] = EsmState.DISABLING
1✔
2430
        else:
2431
            event_source_mapping["State"] = EsmState.UPDATING
1✔
2432

2433
        # To ensure parity, certain responses need to be immediately returned
2434
        temp_params["State"] = event_source_mapping["State"]
1✔
2435

2436
        state.event_source_mappings[uuid] = event_source_mapping
1✔
2437

2438
        # TODO: Currently, we re-create the entire ESM worker. Look into approach with better performance.
2439
        worker_factory = EsmWorkerFactory(
1✔
2440
            event_source_mapping, function_role, request.get("Enabled", esm_worker.enabled)
2441
        )
2442

2443
        # Get a new ESM worker object but do not active it, since the factory holds all logic for creating new worker from configuration.
2444
        updated_esm_worker = worker_factory.get_esm_worker()
1✔
2445
        self.esm_workers[uuid] = updated_esm_worker
1✔
2446

2447
        # We should stop() the worker since the delete() will remove the ESM from the state mapping.
2448
        esm_worker.stop()
1✔
2449
        # This will either create an EsmWorker in the CREATING state if enabled. Otherwise, the DISABLING state is set.
2450
        updated_esm_worker.create()
1✔
2451

2452
        return {**event_source_mapping, **temp_params}
1✔
2453

2454
    def delete_event_source_mapping(
1✔
2455
        self, context: RequestContext, uuid: String, **kwargs
2456
    ) -> EventSourceMappingConfiguration:
2457
        state = lambda_stores[context.account_id][context.region]
1✔
2458
        event_source_mapping = state.event_source_mappings.get(uuid)
1✔
2459
        if not event_source_mapping:
1✔
2460
            raise ResourceNotFoundException(
1✔
2461
                "The resource you requested does not exist.", Type="User"
2462
            )
2463
        esm = state.event_source_mappings[uuid]
1✔
2464
        # TODO: add proper locking
2465
        esm_worker = self.esm_workers.pop(uuid, None)
1✔
2466
        # Asynchronous delete in v2
2467
        if not esm_worker:
1✔
UNCOV
2468
            raise ResourceNotFoundException(
×
2469
                "The resource you requested does not exist.", Type="User"
2470
            )
2471
        esm_worker.delete()
1✔
2472
        return {**esm, "State": EsmState.DELETING}
1✔
2473

2474
    def get_event_source_mapping(
1✔
2475
        self, context: RequestContext, uuid: String, **kwargs
2476
    ) -> EventSourceMappingConfiguration:
2477
        state = lambda_stores[context.account_id][context.region]
1✔
2478
        event_source_mapping = state.event_source_mappings.get(uuid)
1✔
2479
        if not event_source_mapping:
1✔
2480
            raise ResourceNotFoundException(
1✔
2481
                "The resource you requested does not exist.", Type="User"
2482
            )
2483
        esm_worker = self.esm_workers.get(uuid)
1✔
2484
        if not esm_worker:
1✔
UNCOV
2485
            raise ResourceNotFoundException(
×
2486
                "The resource you requested does not exist.", Type="User"
2487
            )
2488
        event_source_mapping["State"] = esm_worker.current_state
1✔
2489
        event_source_mapping["StateTransitionReason"] = esm_worker.state_transition_reason
1✔
2490
        return event_source_mapping
1✔
2491

2492
    def list_event_source_mappings(
1✔
2493
        self,
2494
        context: RequestContext,
2495
        event_source_arn: Arn = None,
2496
        function_name: FunctionName = None,
2497
        marker: String = None,
2498
        max_items: MaxListItems = None,
2499
        **kwargs,
2500
    ) -> ListEventSourceMappingsResponse:
2501
        state = lambda_stores[context.account_id][context.region]
1✔
2502

2503
        esms = state.event_source_mappings.values()
1✔
2504
        # TODO: update and test State and StateTransitionReason for ESM v2
2505

2506
        if event_source_arn:  # TODO: validate pattern
1✔
2507
            esms = [e for e in esms if e.get("EventSourceArn") == event_source_arn]
1✔
2508

2509
        if function_name:
1✔
2510
            esms = [e for e in esms if function_name in e["FunctionArn"]]
1✔
2511

2512
        esms = PaginatedList(esms)
1✔
2513
        page, token = esms.get_page(
1✔
2514
            lambda x: x["UUID"],
2515
            marker,
2516
            max_items,
2517
        )
2518
        return ListEventSourceMappingsResponse(EventSourceMappings=page, NextMarker=token)
1✔
2519

2520
    def get_source_type_from_request(self, request: dict[str, Any]) -> str:
1✔
2521
        if event_source_arn := request.get("EventSourceArn", ""):
×
2522
            service = extract_service_from_arn(event_source_arn)
×
2523
            if service == "sqs" and "fifo" in event_source_arn:
×
UNCOV
2524
                service = "sqs-fifo"
×
UNCOV
2525
            return service
×
UNCOV
2526
        elif request.get("SelfManagedEventSource"):
×
UNCOV
2527
            return "kafka"
×
2528

2529
    # =======================================
2530
    # ============ FUNCTION URLS ============
2531
    # =======================================
2532

2533
    @staticmethod
1✔
2534
    def _validate_qualifier(qualifier: str) -> None:
1✔
2535
        if qualifier in ["$LATEST", "$LATEST.PUBLISHED"] or (
1✔
2536
            qualifier and api_utils.qualifier_is_version(qualifier)
2537
        ):
2538
            raise ValidationException(
1✔
2539
                f"1 validation error detected: Value '{qualifier}' at 'qualifier' failed to satisfy constraint: Member must satisfy regular expression pattern: ((?!^\\d+$)^[0-9a-zA-Z-_]+$)"
2540
            )
2541

2542
    @staticmethod
1✔
2543
    def _validate_invoke_mode(invoke_mode: str) -> None:
1✔
2544
        if invoke_mode and invoke_mode not in [InvokeMode.BUFFERED, InvokeMode.RESPONSE_STREAM]:
1✔
2545
            raise ValidationException(
1✔
2546
                f"1 validation error detected: Value '{invoke_mode}' at 'invokeMode' failed to satisfy constraint: Member must satisfy enum value set: [RESPONSE_STREAM, BUFFERED]"
2547
            )
2548
        if invoke_mode == InvokeMode.RESPONSE_STREAM:
1✔
2549
            # TODO should we actually fail for setting RESPONSE_STREAM?
2550
            #  It should trigger InvokeWithResponseStream which is not implemented
2551
            LOG.warning(
1✔
2552
                "The invokeMode 'RESPONSE_STREAM' is not yet supported on LocalStack. The property is only mocked, the execution will still be 'BUFFERED'"
2553
            )
2554

2555
    # TODO: what happens if function state is not active?
2556
    def create_function_url_config(
1✔
2557
        self,
2558
        context: RequestContext,
2559
        function_name: FunctionName,
2560
        auth_type: FunctionUrlAuthType,
2561
        qualifier: FunctionUrlQualifier = None,
2562
        cors: Cors = None,
2563
        invoke_mode: InvokeMode = None,
2564
        **kwargs,
2565
    ) -> CreateFunctionUrlConfigResponse:
2566
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2567
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
2568
            function_name, qualifier, context
2569
        )
2570
        state = lambda_stores[account_id][region]
1✔
2571
        self._validate_qualifier(qualifier)
1✔
2572
        self._validate_invoke_mode(invoke_mode)
1✔
2573

2574
        fn = state.functions.get(function_name)
1✔
2575
        if fn is None:
1✔
2576
            raise ResourceNotFoundException("Function does not exist", Type="User")
1✔
2577

2578
        url_config = fn.function_url_configs.get(qualifier or "$LATEST")
1✔
2579
        if url_config:
1✔
2580
            raise ResourceConflictException(
1✔
2581
                f"Failed to create function url config for [functionArn = {url_config.function_arn}]. Error message:  FunctionUrlConfig exists for this Lambda function",
2582
                Type="User",
2583
            )
2584

2585
        if qualifier and qualifier != "$LATEST" and qualifier not in fn.aliases:
1✔
2586
            raise ResourceNotFoundException("Function does not exist", Type="User")
1✔
2587

2588
        normalized_qualifier = qualifier or "$LATEST"
1✔
2589

2590
        function_arn = (
1✔
2591
            api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
2592
            if qualifier
2593
            else api_utils.unqualified_lambda_arn(function_name, account_id, region)
2594
        )
2595

2596
        custom_id: str | None = None
1✔
2597

2598
        tags = self._get_tags(api_utils.unqualified_lambda_arn(function_name, account_id, region))
1✔
2599
        if TAG_KEY_CUSTOM_URL in tags:
1✔
2600
            # Note: I really wanted to add verification here that the
2601
            # url_id is unique, so we could surface that to the user ASAP.
2602
            # However, it seems like that information isn't available yet,
2603
            # since (as far as I can tell) we call
2604
            # self.router.register_routes() once, in a single shot, for all
2605
            # of the routes -- and we need to verify that it's unique not
2606
            # just for this particular lambda function, but for the entire
2607
            # lambda provider. Therefore... that idea proved non-trivial!
2608
            custom_id_tag_value = (
1✔
2609
                f"{tags[TAG_KEY_CUSTOM_URL]}-{qualifier}" if qualifier else tags[TAG_KEY_CUSTOM_URL]
2610
            )
2611
            if TAG_KEY_CUSTOM_URL_VALIDATOR.match(custom_id_tag_value):
1✔
2612
                custom_id = custom_id_tag_value
1✔
2613

2614
            else:
2615
                # Note: we're logging here instead of raising to prioritize
2616
                # strict parity with AWS over the localstack-only custom_id
2617
                LOG.warning(
1✔
2618
                    "Invalid custom ID tag value for lambda URL (%s=%s). "
2619
                    "Replaced with default (random id)",
2620
                    TAG_KEY_CUSTOM_URL,
2621
                    custom_id_tag_value,
2622
                )
2623

2624
        # The url_id is the subdomain used for the URL we're creating. This
2625
        # is either created randomly (as in AWS), or can be passed as a tag
2626
        # to the lambda itself (localstack-only).
2627
        url_id: str
2628
        if custom_id is None:
1✔
2629
            url_id = api_utils.generate_random_url_id()
1✔
2630
        else:
2631
            url_id = custom_id
1✔
2632

2633
        host_definition = localstack_host(custom_port=config.GATEWAY_LISTEN[0].port)
1✔
2634
        fn.function_url_configs[normalized_qualifier] = FunctionUrlConfig(
1✔
2635
            function_arn=function_arn,
2636
            function_name=function_name,
2637
            cors=cors,
2638
            url_id=url_id,
2639
            url=f"http://{url_id}.lambda-url.{context.region}.{host_definition.host_and_port()}/",  # TODO: https support
2640
            auth_type=auth_type,
2641
            creation_time=api_utils.generate_lambda_date(),
2642
            last_modified_time=api_utils.generate_lambda_date(),
2643
            invoke_mode=invoke_mode,
2644
        )
2645

2646
        # persist and start URL
2647
        # TODO: implement URL invoke
2648
        api_url_config = api_utils.map_function_url_config(
1✔
2649
            fn.function_url_configs[normalized_qualifier]
2650
        )
2651

2652
        return CreateFunctionUrlConfigResponse(
1✔
2653
            FunctionUrl=api_url_config["FunctionUrl"],
2654
            FunctionArn=api_url_config["FunctionArn"],
2655
            AuthType=api_url_config["AuthType"],
2656
            Cors=api_url_config["Cors"],
2657
            CreationTime=api_url_config["CreationTime"],
2658
            InvokeMode=api_url_config["InvokeMode"],
2659
        )
2660

2661
    def get_function_url_config(
1✔
2662
        self,
2663
        context: RequestContext,
2664
        function_name: FunctionName,
2665
        qualifier: FunctionUrlQualifier = None,
2666
        **kwargs,
2667
    ) -> GetFunctionUrlConfigResponse:
2668
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2669
        state = lambda_stores[account_id][region]
1✔
2670

2671
        fn_name, qualifier = api_utils.get_name_and_qualifier(function_name, qualifier, context)
1✔
2672

2673
        self._validate_qualifier(qualifier)
1✔
2674

2675
        resolved_fn = state.functions.get(fn_name)
1✔
2676
        if not resolved_fn:
1✔
2677
            raise ResourceNotFoundException(
1✔
2678
                "The resource you requested does not exist.", Type="User"
2679
            )
2680

2681
        qualifier = qualifier or "$LATEST"
1✔
2682
        url_config = resolved_fn.function_url_configs.get(qualifier)
1✔
2683
        if not url_config:
1✔
2684
            raise ResourceNotFoundException(
1✔
2685
                "The resource you requested does not exist.", Type="User"
2686
            )
2687

2688
        return api_utils.map_function_url_config(url_config)
1✔
2689

2690
    def update_function_url_config(
1✔
2691
        self,
2692
        context: RequestContext,
2693
        function_name: FunctionName,
2694
        qualifier: FunctionUrlQualifier = None,
2695
        auth_type: FunctionUrlAuthType = None,
2696
        cors: Cors = None,
2697
        invoke_mode: InvokeMode = None,
2698
        **kwargs,
2699
    ) -> UpdateFunctionUrlConfigResponse:
2700
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2701
        state = lambda_stores[account_id][region]
1✔
2702

2703
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
2704
            function_name, qualifier, context
2705
        )
2706
        self._validate_qualifier(qualifier)
1✔
2707
        self._validate_invoke_mode(invoke_mode)
1✔
2708

2709
        fn = state.functions.get(function_name)
1✔
2710
        if not fn:
1✔
2711
            raise ResourceNotFoundException("Function does not exist", Type="User")
1✔
2712

2713
        normalized_qualifier = qualifier or "$LATEST"
1✔
2714

2715
        if (
1✔
2716
            api_utils.qualifier_is_alias(normalized_qualifier)
2717
            and normalized_qualifier not in fn.aliases
2718
        ):
2719
            raise ResourceNotFoundException("Function does not exist", Type="User")
1✔
2720

2721
        url_config = fn.function_url_configs.get(normalized_qualifier)
1✔
2722
        if not url_config:
1✔
2723
            raise ResourceNotFoundException(
1✔
2724
                "The resource you requested does not exist.", Type="User"
2725
            )
2726

2727
        changes = {
1✔
2728
            "last_modified_time": api_utils.generate_lambda_date(),
2729
            **({"cors": cors} if cors is not None else {}),
2730
            **({"auth_type": auth_type} if auth_type is not None else {}),
2731
        }
2732

2733
        if invoke_mode:
1✔
2734
            changes["invoke_mode"] = invoke_mode
1✔
2735

2736
        new_url_config = dataclasses.replace(url_config, **changes)
1✔
2737
        fn.function_url_configs[normalized_qualifier] = new_url_config
1✔
2738

2739
        return UpdateFunctionUrlConfigResponse(
1✔
2740
            FunctionUrl=new_url_config.url,
2741
            FunctionArn=new_url_config.function_arn,
2742
            AuthType=new_url_config.auth_type,
2743
            Cors=new_url_config.cors,
2744
            CreationTime=new_url_config.creation_time,
2745
            LastModifiedTime=new_url_config.last_modified_time,
2746
            InvokeMode=new_url_config.invoke_mode,
2747
        )
2748

2749
    def delete_function_url_config(
1✔
2750
        self,
2751
        context: RequestContext,
2752
        function_name: FunctionName,
2753
        qualifier: FunctionUrlQualifier = None,
2754
        **kwargs,
2755
    ) -> None:
2756
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2757
        state = lambda_stores[account_id][region]
1✔
2758

2759
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
2760
            function_name, qualifier, context
2761
        )
2762
        self._validate_qualifier(qualifier)
1✔
2763

2764
        resolved_fn = state.functions.get(function_name)
1✔
2765
        if not resolved_fn:
1✔
2766
            raise ResourceNotFoundException(
1✔
2767
                "The resource you requested does not exist.", Type="User"
2768
            )
2769

2770
        qualifier = qualifier or "$LATEST"
1✔
2771
        url_config = resolved_fn.function_url_configs.get(qualifier)
1✔
2772
        if not url_config:
1✔
2773
            raise ResourceNotFoundException(
1✔
2774
                "The resource you requested does not exist.", Type="User"
2775
            )
2776

2777
        del resolved_fn.function_url_configs[qualifier]
1✔
2778

2779
    def list_function_url_configs(
1✔
2780
        self,
2781
        context: RequestContext,
2782
        function_name: FunctionName,
2783
        marker: String = None,
2784
        max_items: MaxItems = None,
2785
        **kwargs,
2786
    ) -> ListFunctionUrlConfigsResponse:
2787
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2788
        state = lambda_stores[account_id][region]
1✔
2789

2790
        fn_name = api_utils.get_function_name(function_name, context)
1✔
2791
        resolved_fn = state.functions.get(fn_name)
1✔
2792
        if not resolved_fn:
1✔
2793
            raise ResourceNotFoundException("Function does not exist", Type="User")
1✔
2794

2795
        url_configs = [
1✔
2796
            api_utils.map_function_url_config(fn_conf)
2797
            for fn_conf in resolved_fn.function_url_configs.values()
2798
        ]
2799
        url_configs = PaginatedList(url_configs)
1✔
2800
        page, token = url_configs.get_page(
1✔
2801
            lambda url_config: url_config["FunctionArn"],
2802
            marker,
2803
            max_items,
2804
        )
2805
        url_configs = page
1✔
2806
        return ListFunctionUrlConfigsResponse(FunctionUrlConfigs=url_configs, NextMarker=token)
1✔
2807

2808
    # =======================================
2809
    # ============  Permissions  ============
2810
    # =======================================
2811

2812
    @handler("AddPermission", expand=False)
1✔
2813
    def add_permission(
1✔
2814
        self,
2815
        context: RequestContext,
2816
        request: AddPermissionRequest,
2817
    ) -> AddPermissionResponse:
2818
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
2819
            request.get("FunctionName"), request.get("Qualifier"), context
2820
        )
2821

2822
        # validate qualifier
2823
        if qualifier is not None:
1✔
2824
            self._validate_qualifier_expression(qualifier)
1✔
2825
            if qualifier == "$LATEST":
1✔
2826
                raise InvalidParameterValueException(
1✔
2827
                    "We currently do not support adding policies for $LATEST.", Type="User"
2828
                )
2829
        account_id, region = api_utils.get_account_and_region(request.get("FunctionName"), context)
1✔
2830

2831
        resolved_fn = self._get_function(function_name, account_id, region)
1✔
2832
        resolved_qualifier, fn_arn = self._resolve_fn_qualifier(resolved_fn, qualifier)
1✔
2833

2834
        revision_id = request.get("RevisionId")
1✔
2835
        if revision_id:
1✔
2836
            fn_revision_id = self._function_revision_id(resolved_fn, resolved_qualifier)
1✔
2837
            if revision_id != fn_revision_id:
1✔
2838
                raise PreconditionFailedException(
1✔
2839
                    "The Revision Id provided does not match the latest Revision Id. "
2840
                    "Call the GetPolicy API to retrieve the latest Revision Id",
2841
                    Type="User",
2842
                )
2843

2844
        request_sid = request["StatementId"]
1✔
2845
        if not bool(STATEMENT_ID_REGEX.match(request_sid)):
1✔
2846
            raise ValidationException(
1✔
2847
                f"1 validation error detected: Value '{request_sid}' at 'statementId' failed to satisfy constraint: Member must satisfy regular expression pattern: ([a-zA-Z0-9-_]+)"
2848
            )
2849
        # check for an already existing policy and any conflicts in existing statements
2850
        existing_policy = resolved_fn.permissions.get(resolved_qualifier)
1✔
2851
        if existing_policy:
1✔
2852
            if request_sid in [s["Sid"] for s in existing_policy.policy.Statement]:
1✔
2853
                # uniqueness scope: statement id needs to be unique per qualified function ($LATEST, version, or alias)
2854
                # Counterexample: the same sid can exist within $LATEST, version, and alias
2855
                raise ResourceConflictException(
1✔
2856
                    f"The statement id ({request_sid}) provided already exists. Please provide a new statement id, or remove the existing statement.",
2857
                    Type="User",
2858
                )
2859

2860
        permission_statement = api_utils.build_statement(
1✔
2861
            partition=context.partition,
2862
            resource_arn=fn_arn,
2863
            statement_id=request["StatementId"],
2864
            action=request["Action"],
2865
            principal=request["Principal"],
2866
            source_arn=request.get("SourceArn"),
2867
            source_account=request.get("SourceAccount"),
2868
            principal_org_id=request.get("PrincipalOrgID"),
2869
            event_source_token=request.get("EventSourceToken"),
2870
            auth_type=request.get("FunctionUrlAuthType"),
2871
        )
2872
        new_policy = existing_policy
1✔
2873
        if not existing_policy:
1✔
2874
            new_policy = FunctionResourcePolicy(
1✔
2875
                policy=ResourcePolicy(Version="2012-10-17", Id="default", Statement=[])
2876
            )
2877
        new_policy.policy.Statement.append(permission_statement)
1✔
2878
        if not existing_policy:
1✔
2879
            resolved_fn.permissions[resolved_qualifier] = new_policy
1✔
2880

2881
        # Update revision id of alias or version
2882
        # TODO: re-evaluate data model to prevent this dirty hack just for bumping the revision id
2883
        # TODO: does that need a `with function.lock` for atomic updates of the policy + revision_id?
2884
        if api_utils.qualifier_is_alias(resolved_qualifier):
1✔
2885
            resolved_alias = resolved_fn.aliases[resolved_qualifier]
1✔
2886
            resolved_fn.aliases[resolved_qualifier] = dataclasses.replace(resolved_alias)
1✔
2887
        # Assumes that a non-alias is a version
2888
        else:
2889
            resolved_version = resolved_fn.versions[resolved_qualifier]
1✔
2890
            resolved_fn.versions[resolved_qualifier] = dataclasses.replace(
1✔
2891
                resolved_version, config=dataclasses.replace(resolved_version.config)
2892
            )
2893
        return AddPermissionResponse(Statement=json.dumps(permission_statement))
1✔
2894

2895
    def remove_permission(
1✔
2896
        self,
2897
        context: RequestContext,
2898
        function_name: NamespacedFunctionName,
2899
        statement_id: NamespacedStatementId,
2900
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
2901
        revision_id: String | None = None,
2902
        **kwargs,
2903
    ) -> None:
2904
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2905
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
2906
            function_name, qualifier, context
2907
        )
2908
        if qualifier is not None:
1✔
2909
            self._validate_qualifier_expression(qualifier)
1✔
2910

2911
        state = lambda_stores[account_id][region]
1✔
2912
        resolved_fn = state.functions.get(function_name)
1✔
2913
        if resolved_fn is None:
1✔
2914
            fn_arn = api_utils.unqualified_lambda_arn(function_name, account_id, region)
1✔
2915
            raise ResourceNotFoundException(f"No policy found for: {fn_arn}", Type="User")
1✔
2916

2917
        resolved_qualifier, _ = self._resolve_fn_qualifier(resolved_fn, qualifier)
1✔
2918
        function_permission = resolved_fn.permissions.get(resolved_qualifier)
1✔
2919
        if not function_permission:
1✔
2920
            raise ResourceNotFoundException(
1✔
2921
                "No policy is associated with the given resource.", Type="User"
2922
            )
2923

2924
        # try to find statement in policy and delete it
2925
        statement = None
1✔
2926
        for s in function_permission.policy.Statement:
1✔
2927
            if s["Sid"] == statement_id:
1✔
2928
                statement = s
1✔
2929
                break
1✔
2930

2931
        if not statement:
1✔
2932
            raise ResourceNotFoundException(
1✔
2933
                f"Statement {statement_id} is not found in resource policy.", Type="User"
2934
            )
2935
        fn_revision_id = self._function_revision_id(resolved_fn, resolved_qualifier)
1✔
2936
        if revision_id and revision_id != fn_revision_id:
1✔
UNCOV
2937
            raise PreconditionFailedException(
×
2938
                "The Revision Id provided does not match the latest Revision Id. "
2939
                "Call the GetFunction/GetAlias API to retrieve the latest Revision Id",
2940
                Type="User",
2941
            )
2942
        function_permission.policy.Statement.remove(statement)
1✔
2943

2944
        # Update revision id for alias or version
2945
        # TODO: re-evaluate data model to prevent this dirty hack just for bumping the revision id
2946
        # TODO: does that need a `with function.lock` for atomic updates of the policy + revision_id?
2947
        if api_utils.qualifier_is_alias(resolved_qualifier):
1✔
UNCOV
2948
            resolved_alias = resolved_fn.aliases[resolved_qualifier]
×
UNCOV
2949
            resolved_fn.aliases[resolved_qualifier] = dataclasses.replace(resolved_alias)
×
2950
        # Assumes that a non-alias is a version
2951
        else:
2952
            resolved_version = resolved_fn.versions[resolved_qualifier]
1✔
2953
            resolved_fn.versions[resolved_qualifier] = dataclasses.replace(
1✔
2954
                resolved_version, config=dataclasses.replace(resolved_version.config)
2955
            )
2956

2957
        # remove the policy as a whole when there's no statement left in it
2958
        if len(function_permission.policy.Statement) == 0:
1✔
2959
            del resolved_fn.permissions[resolved_qualifier]
1✔
2960

2961
    def get_policy(
1✔
2962
        self,
2963
        context: RequestContext,
2964
        function_name: NamespacedFunctionName,
2965
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
2966
        **kwargs,
2967
    ) -> GetPolicyResponse:
2968
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
2969
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
2970
            function_name, qualifier, context
2971
        )
2972

2973
        if qualifier is not None:
1✔
2974
            self._validate_qualifier_expression(qualifier)
1✔
2975

2976
        resolved_fn = self._get_function(function_name, account_id, region)
1✔
2977

2978
        resolved_qualifier = qualifier or "$LATEST"
1✔
2979
        function_permission = resolved_fn.permissions.get(resolved_qualifier)
1✔
2980
        if not function_permission:
1✔
2981
            raise ResourceNotFoundException(
1✔
2982
                "The resource you requested does not exist.", Type="User"
2983
            )
2984

2985
        fn_revision_id = None
1✔
2986
        if api_utils.qualifier_is_alias(resolved_qualifier):
1✔
2987
            resolved_alias = resolved_fn.aliases[resolved_qualifier]
1✔
2988
            fn_revision_id = resolved_alias.revision_id
1✔
2989
        # Assumes that a non-alias is a version
2990
        else:
2991
            resolved_version = resolved_fn.versions[resolved_qualifier]
1✔
2992
            fn_revision_id = resolved_version.config.revision_id
1✔
2993

2994
        return GetPolicyResponse(
1✔
2995
            Policy=json.dumps(dataclasses.asdict(function_permission.policy)),
2996
            RevisionId=fn_revision_id,
2997
        )
2998

2999
    # =======================================
3000
    # ========  Code signing config  ========
3001
    # =======================================
3002

3003
    def create_code_signing_config(
1✔
3004
        self,
3005
        context: RequestContext,
3006
        allowed_publishers: AllowedPublishers,
3007
        description: Description | None = None,
3008
        code_signing_policies: CodeSigningPolicies | None = None,
3009
        tags: Tags | None = None,
3010
        **kwargs,
3011
    ) -> CreateCodeSigningConfigResponse:
3012
        account = context.account_id
1✔
3013
        region = context.region
1✔
3014

3015
        state = lambda_stores[account][region]
1✔
3016
        # TODO: can there be duplicates?
3017
        csc_id = f"csc-{get_random_hex(17)}"  # e.g. 'csc-077c33b4c19e26036'
1✔
3018
        csc_arn = f"arn:{context.partition}:lambda:{region}:{account}:code-signing-config:{csc_id}"
1✔
3019
        csc = CodeSigningConfig(
1✔
3020
            csc_id=csc_id,
3021
            arn=csc_arn,
3022
            allowed_publishers=allowed_publishers,
3023
            policies=code_signing_policies,
3024
            last_modified=api_utils.generate_lambda_date(),
3025
            description=description,
3026
        )
3027
        state.code_signing_configs[csc_arn] = csc
1✔
3028
        return CreateCodeSigningConfigResponse(CodeSigningConfig=api_utils.map_csc(csc))
1✔
3029

3030
    def put_function_code_signing_config(
1✔
3031
        self,
3032
        context: RequestContext,
3033
        code_signing_config_arn: CodeSigningConfigArn,
3034
        function_name: NamespacedFunctionName,
3035
        **kwargs,
3036
    ) -> PutFunctionCodeSigningConfigResponse:
3037
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3038
        state = lambda_stores[account_id][region]
1✔
3039
        function_name = api_utils.get_function_name(function_name, context)
1✔
3040

3041
        csc = state.code_signing_configs.get(code_signing_config_arn)
1✔
3042
        if not csc:
1✔
3043
            raise CodeSigningConfigNotFoundException(
1✔
3044
                f"The code signing configuration cannot be found. Check that the provided configuration is not deleted: {code_signing_config_arn}.",
3045
                Type="User",
3046
            )
3047

3048
        fn = state.functions.get(function_name)
1✔
3049
        fn_arn = api_utils.unqualified_lambda_arn(function_name, account_id, region)
1✔
3050
        if not fn:
1✔
3051
            raise ResourceNotFoundException(f"Function not found: {fn_arn}", Type="User")
1✔
3052

3053
        fn.code_signing_config_arn = code_signing_config_arn
1✔
3054
        return PutFunctionCodeSigningConfigResponse(
1✔
3055
            CodeSigningConfigArn=code_signing_config_arn, FunctionName=function_name
3056
        )
3057

3058
    def update_code_signing_config(
1✔
3059
        self,
3060
        context: RequestContext,
3061
        code_signing_config_arn: CodeSigningConfigArn,
3062
        description: Description = None,
3063
        allowed_publishers: AllowedPublishers = None,
3064
        code_signing_policies: CodeSigningPolicies = None,
3065
        **kwargs,
3066
    ) -> UpdateCodeSigningConfigResponse:
3067
        state = lambda_stores[context.account_id][context.region]
1✔
3068
        csc = state.code_signing_configs.get(code_signing_config_arn)
1✔
3069
        if not csc:
1✔
3070
            raise ResourceNotFoundException(
1✔
3071
                f"The Lambda code signing configuration {code_signing_config_arn} can not be found."
3072
            )
3073

3074
        changes = {
1✔
3075
            **(
3076
                {"allowed_publishers": allowed_publishers} if allowed_publishers is not None else {}
3077
            ),
3078
            **({"policies": code_signing_policies} if code_signing_policies is not None else {}),
3079
            **({"description": description} if description is not None else {}),
3080
        }
3081
        new_csc = dataclasses.replace(
1✔
3082
            csc, last_modified=api_utils.generate_lambda_date(), **changes
3083
        )
3084
        state.code_signing_configs[code_signing_config_arn] = new_csc
1✔
3085

3086
        return UpdateCodeSigningConfigResponse(CodeSigningConfig=api_utils.map_csc(new_csc))
1✔
3087

3088
    def get_code_signing_config(
1✔
3089
        self, context: RequestContext, code_signing_config_arn: CodeSigningConfigArn, **kwargs
3090
    ) -> GetCodeSigningConfigResponse:
3091
        state = lambda_stores[context.account_id][context.region]
1✔
3092
        csc = state.code_signing_configs.get(code_signing_config_arn)
1✔
3093
        if not csc:
1✔
3094
            raise ResourceNotFoundException(
1✔
3095
                f"The Lambda code signing configuration {code_signing_config_arn} can not be found."
3096
            )
3097

3098
        return GetCodeSigningConfigResponse(CodeSigningConfig=api_utils.map_csc(csc))
1✔
3099

3100
    def get_function_code_signing_config(
1✔
3101
        self, context: RequestContext, function_name: NamespacedFunctionName, **kwargs
3102
    ) -> GetFunctionCodeSigningConfigResponse:
3103
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3104
        state = lambda_stores[account_id][region]
1✔
3105
        function_name = api_utils.get_function_name(function_name, context)
1✔
3106
        fn = state.functions.get(function_name)
1✔
3107
        fn_arn = api_utils.unqualified_lambda_arn(function_name, account_id, region)
1✔
3108
        if not fn:
1✔
3109
            raise ResourceNotFoundException(f"Function not found: {fn_arn}", Type="User")
1✔
3110

3111
        if fn.code_signing_config_arn:
1✔
3112
            return GetFunctionCodeSigningConfigResponse(
1✔
3113
                CodeSigningConfigArn=fn.code_signing_config_arn, FunctionName=function_name
3114
            )
3115

3116
        return GetFunctionCodeSigningConfigResponse()
1✔
3117

3118
    def delete_function_code_signing_config(
1✔
3119
        self, context: RequestContext, function_name: NamespacedFunctionName, **kwargs
3120
    ) -> None:
3121
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3122
        state = lambda_stores[account_id][region]
1✔
3123
        function_name = api_utils.get_function_name(function_name, context)
1✔
3124
        fn = state.functions.get(function_name)
1✔
3125
        fn_arn = api_utils.unqualified_lambda_arn(function_name, account_id, region)
1✔
3126
        if not fn:
1✔
3127
            raise ResourceNotFoundException(f"Function not found: {fn_arn}", Type="User")
1✔
3128

3129
        fn.code_signing_config_arn = None
1✔
3130

3131
    def delete_code_signing_config(
1✔
3132
        self, context: RequestContext, code_signing_config_arn: CodeSigningConfigArn, **kwargs
3133
    ) -> DeleteCodeSigningConfigResponse:
3134
        state = lambda_stores[context.account_id][context.region]
1✔
3135

3136
        csc = state.code_signing_configs.get(code_signing_config_arn)
1✔
3137
        if not csc:
1✔
3138
            raise ResourceNotFoundException(
1✔
3139
                f"The Lambda code signing configuration {code_signing_config_arn} can not be found."
3140
            )
3141

3142
        del state.code_signing_configs[code_signing_config_arn]
1✔
3143

3144
        return DeleteCodeSigningConfigResponse()
1✔
3145

3146
    def list_code_signing_configs(
1✔
3147
        self,
3148
        context: RequestContext,
3149
        marker: String = None,
3150
        max_items: MaxListItems = None,
3151
        **kwargs,
3152
    ) -> ListCodeSigningConfigsResponse:
3153
        state = lambda_stores[context.account_id][context.region]
1✔
3154

3155
        cscs = [api_utils.map_csc(csc) for csc in state.code_signing_configs.values()]
1✔
3156
        cscs = PaginatedList(cscs)
1✔
3157
        page, token = cscs.get_page(
1✔
3158
            lambda csc: csc["CodeSigningConfigId"],
3159
            marker,
3160
            max_items,
3161
        )
3162
        return ListCodeSigningConfigsResponse(CodeSigningConfigs=page, NextMarker=token)
1✔
3163

3164
    def list_functions_by_code_signing_config(
1✔
3165
        self,
3166
        context: RequestContext,
3167
        code_signing_config_arn: CodeSigningConfigArn,
3168
        marker: String = None,
3169
        max_items: MaxListItems = None,
3170
        **kwargs,
3171
    ) -> ListFunctionsByCodeSigningConfigResponse:
3172
        account = context.account_id
1✔
3173
        region = context.region
1✔
3174

3175
        state = lambda_stores[account][region]
1✔
3176

3177
        if code_signing_config_arn not in state.code_signing_configs:
1✔
3178
            raise ResourceNotFoundException(
1✔
3179
                f"The Lambda code signing configuration {code_signing_config_arn} can not be found."
3180
            )
3181

3182
        fn_arns = [
1✔
3183
            api_utils.unqualified_lambda_arn(fn.function_name, account, region)
3184
            for fn in state.functions.values()
3185
            if fn.code_signing_config_arn == code_signing_config_arn
3186
        ]
3187

3188
        cscs = PaginatedList(fn_arns)
1✔
3189
        page, token = cscs.get_page(
1✔
3190
            lambda x: x,
3191
            marker,
3192
            max_items,
3193
        )
3194
        return ListFunctionsByCodeSigningConfigResponse(FunctionArns=page, NextMarker=token)
1✔
3195

3196
    # =======================================
3197
    # =========  Account Settings   =========
3198
    # =======================================
3199

3200
    # CAVE: these settings & usages are *per* region!
3201
    # Lambda quotas: https://docs.aws.amazon.com/lambda/latest/dg/gettingstarted-limits.html
3202
    def get_account_settings(self, context: RequestContext, **kwargs) -> GetAccountSettingsResponse:
1✔
3203
        state = lambda_stores[context.account_id][context.region]
1✔
3204

3205
        fn_count = 0
1✔
3206
        code_size_sum = 0
1✔
3207
        reserved_concurrency_sum = 0
1✔
3208
        for fn in state.functions.values():
1✔
3209
            fn_count += 1
1✔
3210
            for fn_version in fn.versions.values():
1✔
3211
                # Image-based Lambdas do not have a code attribute and count against the ECR quotas instead
3212
                if fn_version.config.package_type == PackageType.Zip:
1✔
3213
                    code_size_sum += fn_version.config.code.code_size
1✔
3214
            if fn.reserved_concurrent_executions is not None:
1✔
3215
                reserved_concurrency_sum += fn.reserved_concurrent_executions
1✔
3216
            for c in fn.provisioned_concurrency_configs.values():
1✔
3217
                reserved_concurrency_sum += c.provisioned_concurrent_executions
1✔
3218
        for layer in state.layers.values():
1✔
3219
            for layer_version in layer.layer_versions.values():
1✔
3220
                code_size_sum += layer_version.code.code_size
1✔
3221
        return GetAccountSettingsResponse(
1✔
3222
            AccountLimit=AccountLimit(
3223
                TotalCodeSize=config.LAMBDA_LIMITS_TOTAL_CODE_SIZE,
3224
                CodeSizeZipped=config.LAMBDA_LIMITS_CODE_SIZE_ZIPPED,
3225
                CodeSizeUnzipped=config.LAMBDA_LIMITS_CODE_SIZE_UNZIPPED,
3226
                ConcurrentExecutions=config.LAMBDA_LIMITS_CONCURRENT_EXECUTIONS,
3227
                UnreservedConcurrentExecutions=config.LAMBDA_LIMITS_CONCURRENT_EXECUTIONS
3228
                - reserved_concurrency_sum,
3229
            ),
3230
            AccountUsage=AccountUsage(
3231
                TotalCodeSize=code_size_sum,
3232
                FunctionCount=fn_count,
3233
            ),
3234
        )
3235

3236
    # =======================================
3237
    # ==  Provisioned Concurrency Config   ==
3238
    # =======================================
3239

3240
    def _get_provisioned_config(
1✔
3241
        self, context: RequestContext, function_name: str, qualifier: str
3242
    ) -> ProvisionedConcurrencyConfiguration | None:
3243
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3244
        state = lambda_stores[account_id][region]
1✔
3245
        function_name = api_utils.get_function_name(function_name, context)
1✔
3246
        fn = state.functions.get(function_name)
1✔
3247
        if api_utils.qualifier_is_alias(qualifier):
1✔
3248
            fn_alias = None
1✔
3249
            if fn:
1✔
3250
                fn_alias = fn.aliases.get(qualifier)
1✔
3251
            if fn_alias is None:
1✔
3252
                raise ResourceNotFoundException(
1✔
3253
                    f"Cannot find alias arn: {api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)}",
3254
                    Type="User",
3255
                )
3256
        elif api_utils.qualifier_is_version(qualifier):
1✔
3257
            fn_version = None
1✔
3258
            if fn:
1✔
3259
                fn_version = fn.versions.get(qualifier)
1✔
3260
            if fn_version is None:
1✔
3261
                raise ResourceNotFoundException(
1✔
3262
                    f"Function not found: {api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)}",
3263
                    Type="User",
3264
                )
3265

3266
        return fn.provisioned_concurrency_configs.get(qualifier)
1✔
3267

3268
    def put_provisioned_concurrency_config(
1✔
3269
        self,
3270
        context: RequestContext,
3271
        function_name: FunctionName,
3272
        qualifier: Qualifier,
3273
        provisioned_concurrent_executions: PositiveInteger,
3274
        **kwargs,
3275
    ) -> PutProvisionedConcurrencyConfigResponse:
3276
        if provisioned_concurrent_executions <= 0:
1✔
3277
            raise ValidationException(
1✔
3278
                f"1 validation error detected: Value '{provisioned_concurrent_executions}' at 'provisionedConcurrentExecutions' failed to satisfy constraint: Member must have value greater than or equal to 1"
3279
            )
3280

3281
        if qualifier == "$LATEST":
1✔
3282
            raise InvalidParameterValueException(
1✔
3283
                "Provisioned Concurrency Configs cannot be applied to unpublished function versions.",
3284
                Type="User",
3285
            )
3286
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3287
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3288
            function_name, qualifier, context
3289
        )
3290
        state = lambda_stores[account_id][region]
1✔
3291
        fn = state.functions.get(function_name)
1✔
3292

3293
        provisioned_config = self._get_provisioned_config(context, function_name, qualifier)
1✔
3294

3295
        if provisioned_config:  # TODO: merge?
1✔
3296
            # TODO: add a test for partial updates (if possible)
3297
            LOG.warning(
1✔
3298
                "Partial update of provisioned concurrency config is currently not supported."
3299
            )
3300

3301
        other_provisioned_sum = sum(
1✔
3302
            [
3303
                provisioned_configs.provisioned_concurrent_executions
3304
                for provisioned_qualifier, provisioned_configs in fn.provisioned_concurrency_configs.items()
3305
                if provisioned_qualifier != qualifier
3306
            ]
3307
        )
3308

3309
        if (
1✔
3310
            fn.reserved_concurrent_executions is not None
3311
            and fn.reserved_concurrent_executions
3312
            < other_provisioned_sum + provisioned_concurrent_executions
3313
        ):
3314
            raise InvalidParameterValueException(
1✔
3315
                "Requested Provisioned Concurrency should not be greater than the reservedConcurrentExecution for function",
3316
                Type="User",
3317
            )
3318

3319
        if provisioned_concurrent_executions > config.LAMBDA_LIMITS_CONCURRENT_EXECUTIONS:
1✔
3320
            raise InvalidParameterValueException(
1✔
3321
                f"Specified ConcurrentExecutions for function is greater than account's unreserved concurrency"
3322
                f" [{config.LAMBDA_LIMITS_CONCURRENT_EXECUTIONS}]."
3323
            )
3324

3325
        settings = self.get_account_settings(context)
1✔
3326
        unreserved_concurrent_executions = settings["AccountLimit"][
1✔
3327
            "UnreservedConcurrentExecutions"
3328
        ]
3329
        if (
1✔
3330
            unreserved_concurrent_executions - provisioned_concurrent_executions
3331
            < config.LAMBDA_LIMITS_MINIMUM_UNRESERVED_CONCURRENCY
3332
        ):
3333
            raise InvalidParameterValueException(
1✔
3334
                f"Specified ConcurrentExecutions for function decreases account's UnreservedConcurrentExecution below"
3335
                f" its minimum value of [{config.LAMBDA_LIMITS_MINIMUM_UNRESERVED_CONCURRENCY}]."
3336
            )
3337

3338
        provisioned_config = ProvisionedConcurrencyConfiguration(
1✔
3339
            provisioned_concurrent_executions, api_utils.generate_lambda_date()
3340
        )
3341
        fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3342

3343
        if api_utils.qualifier_is_alias(qualifier):
1✔
3344
            alias = fn.aliases.get(qualifier)
1✔
3345
            resolved_version = fn.versions.get(alias.function_version)
1✔
3346

3347
            if (
1✔
3348
                resolved_version
3349
                and fn.provisioned_concurrency_configs.get(alias.function_version) is not None
3350
            ):
3351
                raise ResourceConflictException(
1✔
3352
                    "Alias can't be used for Provisioned Concurrency configuration on an already Provisioned version",
3353
                    Type="User",
3354
                )
3355
            fn_arn = resolved_version.id.qualified_arn()
1✔
3356
        elif api_utils.qualifier_is_version(qualifier):
1✔
3357
            fn_version = fn.versions.get(qualifier)
1✔
3358

3359
            # TODO: might be useful other places, utilize
3360
            pointing_aliases = []
1✔
3361
            for alias in fn.aliases.values():
1✔
3362
                if (
1✔
3363
                    alias.function_version == qualifier
3364
                    and fn.provisioned_concurrency_configs.get(alias.name) is not None
3365
                ):
3366
                    pointing_aliases.append(alias.name)
1✔
3367
            if pointing_aliases:
1✔
3368
                raise ResourceConflictException(
1✔
3369
                    "Version is pointed by a Provisioned Concurrency alias", Type="User"
3370
                )
3371

3372
            fn_arn = fn_version.id.qualified_arn()
1✔
3373

3374
        manager = self.lambda_service.get_lambda_version_manager(fn_arn)
1✔
3375

3376
        fn.provisioned_concurrency_configs[qualifier] = provisioned_config
1✔
3377

3378
        manager.update_provisioned_concurrency_config(
1✔
3379
            provisioned_config.provisioned_concurrent_executions
3380
        )
3381

3382
        return PutProvisionedConcurrencyConfigResponse(
1✔
3383
            RequestedProvisionedConcurrentExecutions=provisioned_config.provisioned_concurrent_executions,
3384
            AvailableProvisionedConcurrentExecutions=0,
3385
            AllocatedProvisionedConcurrentExecutions=0,
3386
            Status=ProvisionedConcurrencyStatusEnum.IN_PROGRESS,
3387
            # StatusReason=manager.provisioned_state.status_reason,
3388
            LastModified=provisioned_config.last_modified,  # TODO: does change with configuration or also with state changes?
3389
        )
3390

3391
    def get_provisioned_concurrency_config(
1✔
3392
        self, context: RequestContext, function_name: FunctionName, qualifier: Qualifier, **kwargs
3393
    ) -> GetProvisionedConcurrencyConfigResponse:
3394
        if qualifier == "$LATEST":
1✔
3395
            raise InvalidParameterValueException(
1✔
3396
                "The function resource provided must be an alias or a published version.",
3397
                Type="User",
3398
            )
3399
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3400
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3401
            function_name, qualifier, context
3402
        )
3403

3404
        provisioned_config = self._get_provisioned_config(context, function_name, qualifier)
1✔
3405
        if not provisioned_config:
1✔
3406
            raise ProvisionedConcurrencyConfigNotFoundException(
1✔
3407
                "No Provisioned Concurrency Config found for this function", Type="User"
3408
            )
3409

3410
        # TODO: make this compatible with alias pointer migration on update
3411
        if api_utils.qualifier_is_alias(qualifier):
1✔
3412
            state = lambda_stores[account_id][region]
1✔
3413
            fn = state.functions.get(function_name)
1✔
3414
            alias = fn.aliases.get(qualifier)
1✔
3415
            fn_arn = api_utils.qualified_lambda_arn(
1✔
3416
                function_name, alias.function_version, account_id, region
3417
            )
3418
        else:
3419
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3420

3421
        ver_manager = self.lambda_service.get_lambda_version_manager(fn_arn)
1✔
3422

3423
        return GetProvisionedConcurrencyConfigResponse(
1✔
3424
            RequestedProvisionedConcurrentExecutions=provisioned_config.provisioned_concurrent_executions,
3425
            LastModified=provisioned_config.last_modified,
3426
            AvailableProvisionedConcurrentExecutions=ver_manager.provisioned_state.available,
3427
            AllocatedProvisionedConcurrentExecutions=ver_manager.provisioned_state.allocated,
3428
            Status=ver_manager.provisioned_state.status,
3429
            StatusReason=ver_manager.provisioned_state.status_reason,
3430
        )
3431

3432
    def list_provisioned_concurrency_configs(
1✔
3433
        self,
3434
        context: RequestContext,
3435
        function_name: FunctionName,
3436
        marker: String = None,
3437
        max_items: MaxProvisionedConcurrencyConfigListItems = None,
3438
        **kwargs,
3439
    ) -> ListProvisionedConcurrencyConfigsResponse:
3440
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3441
        state = lambda_stores[account_id][region]
1✔
3442

3443
        function_name = api_utils.get_function_name(function_name, context)
1✔
3444
        fn = state.functions.get(function_name)
1✔
3445
        if fn is None:
1✔
3446
            raise ResourceNotFoundException(
1✔
3447
                f"Function not found: {api_utils.unqualified_lambda_arn(function_name, account_id, region)}",
3448
                Type="User",
3449
            )
3450

3451
        configs = []
1✔
3452
        for qualifier, pc_config in fn.provisioned_concurrency_configs.items():
1✔
UNCOV
3453
            if api_utils.qualifier_is_alias(qualifier):
×
UNCOV
3454
                alias = fn.aliases.get(qualifier)
×
3455
                fn_arn = api_utils.qualified_lambda_arn(
×
3456
                    function_name, alias.function_version, account_id, region
3457
                )
3458
            else:
3459
                fn_arn = api_utils.qualified_lambda_arn(
×
3460
                    function_name, qualifier, account_id, region
3461
                )
3462

UNCOV
3463
            manager = self.lambda_service.get_lambda_version_manager(fn_arn)
×
3464

UNCOV
3465
            configs.append(
×
3466
                ProvisionedConcurrencyConfigListItem(
3467
                    FunctionArn=api_utils.qualified_lambda_arn(
3468
                        function_name, qualifier, account_id, region
3469
                    ),
3470
                    RequestedProvisionedConcurrentExecutions=pc_config.provisioned_concurrent_executions,
3471
                    AvailableProvisionedConcurrentExecutions=manager.provisioned_state.available,
3472
                    AllocatedProvisionedConcurrentExecutions=manager.provisioned_state.allocated,
3473
                    Status=manager.provisioned_state.status,
3474
                    StatusReason=manager.provisioned_state.status_reason,
3475
                    LastModified=pc_config.last_modified,
3476
                )
3477
            )
3478

3479
        provisioned_concurrency_configs = configs
1✔
3480
        provisioned_concurrency_configs = PaginatedList(provisioned_concurrency_configs)
1✔
3481
        page, token = provisioned_concurrency_configs.get_page(
1✔
3482
            lambda x: x,
3483
            marker,
3484
            max_items,
3485
        )
3486
        return ListProvisionedConcurrencyConfigsResponse(
1✔
3487
            ProvisionedConcurrencyConfigs=page, NextMarker=token
3488
        )
3489

3490
    def delete_provisioned_concurrency_config(
1✔
3491
        self, context: RequestContext, function_name: FunctionName, qualifier: Qualifier, **kwargs
3492
    ) -> None:
3493
        if qualifier == "$LATEST":
1✔
3494
            raise InvalidParameterValueException(
1✔
3495
                "The function resource provided must be an alias or a published version.",
3496
                Type="User",
3497
            )
3498
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3499
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3500
            function_name, qualifier, context
3501
        )
3502
        state = lambda_stores[account_id][region]
1✔
3503
        fn = state.functions.get(function_name)
1✔
3504

3505
        provisioned_config = self._get_provisioned_config(context, function_name, qualifier)
1✔
3506
        # delete is idempotent and doesn't actually care about the provisioned concurrency config not existing
3507
        if provisioned_config:
1✔
3508
            fn.provisioned_concurrency_configs.pop(qualifier)
1✔
3509
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3510
            manager = self.lambda_service.get_lambda_version_manager(fn_arn)
1✔
3511
            manager.update_provisioned_concurrency_config(0)
1✔
3512

3513
    # =======================================
3514
    # =======  Event Invoke Config   ========
3515
    # =======================================
3516

3517
    # "1 validation error detected: Value 'arn:aws:_-/!lambda:<region>:111111111111:function:<function-name:1>' at 'destinationConfig.onFailure.destination' failed to satisfy constraint: Member must satisfy regular expression pattern: ^$|arn:(aws[a-zA-Z0-9-]*):([a-zA-Z0-9\\-])+:([a-z]{2}((-gov)|(-iso(b?)))?-[a-z]+-\\d{1})?:(\\d{12})?:(.*)"
3518
    # "1 validation error detected: Value 'arn:aws:_-/!lambda:<region>:111111111111:function:<function-name:1>' at 'destinationConfig.onFailure.destination' failed to satisfy constraint: Member must satisfy regular expression pattern: ^$|arn:(aws[a-zA-Z0-9-]*):([a-zA-Z0-9\\-])+:([a-z]2((-gov)|(-iso(b?)))?-[a-z]+-\\d1)?:(\\d12)?:(.*)" ... (expected → actual)
3519

3520
    def _validate_destination_config(
1✔
3521
        self, store: LambdaStore, function_name: str, destination_config: DestinationConfig
3522
    ):
3523
        def _validate_destination_arn(destination_arn) -> bool:
1✔
3524
            if not api_utils.DESTINATION_ARN_PATTERN.match(destination_arn):
1✔
3525
                # technically we shouldn't handle this in the provider
3526
                raise ValidationException(
1✔
3527
                    "1 validation error detected: Value '"
3528
                    + destination_arn
3529
                    + "' at 'destinationConfig.onFailure.destination' failed to satisfy constraint: Member must satisfy regular expression pattern: "
3530
                    + "$|kafka://([^.]([a-zA-Z0-9\\-_.]{0,248}))|arn:(aws[a-zA-Z0-9-]*):([a-zA-Z0-9\\-])+:((eusc-)?[a-z]{2}((-gov)|(-iso([a-z]?)))?-[a-z]+-\\d{1})?:(\\d{12})?:(.*)"
3531
                )
3532

3533
            match destination_arn.split(":")[2]:
1✔
3534
                case "lambda":
1✔
3535
                    fn_parts = api_utils.FULL_FN_ARN_PATTERN.search(destination_arn).groupdict()
1✔
3536
                    if fn_parts:
1✔
3537
                        # check if it exists
3538
                        fn = store.functions.get(fn_parts["function_name"])
1✔
3539
                        if not fn:
1✔
3540
                            raise InvalidParameterValueException(
1✔
3541
                                f"The destination ARN {destination_arn} is invalid.", Type="User"
3542
                            )
3543
                        if fn_parts["function_name"] == function_name:
1✔
3544
                            raise InvalidParameterValueException(
1✔
3545
                                "You can't specify the function as a destination for itself.",
3546
                                Type="User",
3547
                            )
3548
                case "sns" | "sqs" | "events":
1✔
3549
                    pass
1✔
3550
                case _:
1✔
3551
                    return False
1✔
3552
            return True
1✔
3553

3554
        validation_err = False
1✔
3555

3556
        failure_destination = destination_config.get("OnFailure", {}).get("Destination")
1✔
3557
        if failure_destination:
1✔
3558
            validation_err = validation_err or not _validate_destination_arn(failure_destination)
1✔
3559

3560
        success_destination = destination_config.get("OnSuccess", {}).get("Destination")
1✔
3561
        if success_destination:
1✔
3562
            validation_err = validation_err or not _validate_destination_arn(success_destination)
1✔
3563

3564
        if validation_err:
1✔
3565
            on_success_part = (
1✔
3566
                f"OnSuccess(destination={success_destination})" if success_destination else "null"
3567
            )
3568
            on_failure_part = (
1✔
3569
                f"OnFailure(destination={failure_destination})" if failure_destination else "null"
3570
            )
3571
            raise InvalidParameterValueException(
1✔
3572
                f"The provided destination config DestinationConfig(onSuccess={on_success_part}, onFailure={on_failure_part}) is invalid.",
3573
                Type="User",
3574
            )
3575

3576
    def put_function_event_invoke_config(
1✔
3577
        self,
3578
        context: RequestContext,
3579
        function_name: FunctionName,
3580
        qualifier: NumericLatestPublishedOrAliasQualifier = None,
3581
        maximum_retry_attempts: MaximumRetryAttempts = None,
3582
        maximum_event_age_in_seconds: MaximumEventAgeInSeconds = None,
3583
        destination_config: DestinationConfig = None,
3584
        **kwargs,
3585
    ) -> FunctionEventInvokeConfig:
3586
        """
3587
        Destination ARNs can be:
3588
        * SQS arn
3589
        * SNS arn
3590
        * Lambda arn
3591
        * EventBridge arn
3592

3593
        Differences between put_ and update_:
3594
            * put overwrites any existing config
3595
            * update allows changes only single values while keeping the rest of existing ones
3596
            * update fails on non-existing configs
3597

3598
        Differences between destination and DLQ
3599
            * "However, a dead-letter queue is part of a function's version-specific configuration, so it is locked in when you publish a version."
3600
            *  "On-failure destinations also support additional targets and include details about the function's response in the invocation record."
3601

3602
        """
3603
        if (
1✔
3604
            maximum_event_age_in_seconds is None
3605
            and maximum_retry_attempts is None
3606
            and destination_config is None
3607
        ):
3608
            raise InvalidParameterValueException(
1✔
3609
                "You must specify at least one of error handling or destination setting.",
3610
                Type="User",
3611
            )
3612
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3613
        state = lambda_stores[account_id][region]
1✔
3614
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3615
            function_name, qualifier, context
3616
        )
3617
        fn = state.functions.get(function_name)
1✔
3618
        if not fn or (qualifier and not (qualifier in fn.aliases or qualifier in fn.versions)):
1✔
3619
            raise ResourceNotFoundException("The function doesn't exist.", Type="User")
1✔
3620

3621
        qualifier = qualifier or "$LATEST"
1✔
3622

3623
        # validate and normalize destination config
3624
        if destination_config:
1✔
3625
            self._validate_destination_config(state, function_name, destination_config)
1✔
3626

3627
        destination_config = DestinationConfig(
1✔
3628
            OnSuccess=OnSuccess(
3629
                Destination=(destination_config or {}).get("OnSuccess", {}).get("Destination")
3630
            ),
3631
            OnFailure=OnFailure(
3632
                Destination=(destination_config or {}).get("OnFailure", {}).get("Destination")
3633
            ),
3634
        )
3635

3636
        config = EventInvokeConfig(
1✔
3637
            function_name=function_name,
3638
            qualifier=qualifier,
3639
            maximum_event_age_in_seconds=maximum_event_age_in_seconds,
3640
            maximum_retry_attempts=maximum_retry_attempts,
3641
            last_modified=api_utils.generate_lambda_date(),
3642
            destination_config=destination_config,
3643
        )
3644
        fn.event_invoke_configs[qualifier] = config
1✔
3645

3646
        return FunctionEventInvokeConfig(
1✔
3647
            LastModified=datetime.datetime.strptime(
3648
                config.last_modified, api_utils.LAMBDA_DATE_FORMAT
3649
            ),
3650
            FunctionArn=api_utils.qualified_lambda_arn(
3651
                function_name, qualifier or "$LATEST", account_id, region
3652
            ),
3653
            DestinationConfig=destination_config,
3654
            MaximumEventAgeInSeconds=maximum_event_age_in_seconds,
3655
            MaximumRetryAttempts=maximum_retry_attempts,
3656
        )
3657

3658
    def get_function_event_invoke_config(
1✔
3659
        self,
3660
        context: RequestContext,
3661
        function_name: FunctionName,
3662
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
3663
        **kwargs,
3664
    ) -> FunctionEventInvokeConfig:
3665
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3666
        state = lambda_stores[account_id][region]
1✔
3667
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3668
            function_name, qualifier, context
3669
        )
3670

3671
        qualifier = qualifier or "$LATEST"
1✔
3672
        fn = state.functions.get(function_name)
1✔
3673
        if not fn:
1✔
3674
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3675
            raise ResourceNotFoundException(
1✔
3676
                f"The function {fn_arn} doesn't have an EventInvokeConfig", Type="User"
3677
            )
3678

3679
        config = fn.event_invoke_configs.get(qualifier)
1✔
3680
        if not config:
1✔
3681
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3682
            raise ResourceNotFoundException(
1✔
3683
                f"The function {fn_arn} doesn't have an EventInvokeConfig", Type="User"
3684
            )
3685

3686
        return FunctionEventInvokeConfig(
1✔
3687
            LastModified=datetime.datetime.strptime(
3688
                config.last_modified, api_utils.LAMBDA_DATE_FORMAT
3689
            ),
3690
            FunctionArn=api_utils.qualified_lambda_arn(
3691
                function_name, qualifier, account_id, region
3692
            ),
3693
            DestinationConfig=config.destination_config,
3694
            MaximumEventAgeInSeconds=config.maximum_event_age_in_seconds,
3695
            MaximumRetryAttempts=config.maximum_retry_attempts,
3696
        )
3697

3698
    def list_function_event_invoke_configs(
1✔
3699
        self,
3700
        context: RequestContext,
3701
        function_name: FunctionName,
3702
        marker: String = None,
3703
        max_items: MaxFunctionEventInvokeConfigListItems = None,
3704
        **kwargs,
3705
    ) -> ListFunctionEventInvokeConfigsResponse:
3706
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3707
        state = lambda_stores[account_id][region]
1✔
3708
        fn = state.functions.get(function_name)
1✔
3709
        if not fn:
1✔
3710
            raise ResourceNotFoundException("The function doesn't exist.", Type="User")
1✔
3711

3712
        event_invoke_configs = [
1✔
3713
            FunctionEventInvokeConfig(
3714
                LastModified=c.last_modified,
3715
                FunctionArn=api_utils.qualified_lambda_arn(
3716
                    function_name, c.qualifier, account_id, region
3717
                ),
3718
                MaximumEventAgeInSeconds=c.maximum_event_age_in_seconds,
3719
                MaximumRetryAttempts=c.maximum_retry_attempts,
3720
                DestinationConfig=c.destination_config,
3721
            )
3722
            for c in fn.event_invoke_configs.values()
3723
        ]
3724

3725
        event_invoke_configs = PaginatedList(event_invoke_configs)
1✔
3726
        page, token = event_invoke_configs.get_page(
1✔
3727
            lambda x: x["FunctionArn"],
3728
            marker,
3729
            max_items,
3730
        )
3731
        return ListFunctionEventInvokeConfigsResponse(
1✔
3732
            FunctionEventInvokeConfigs=page, NextMarker=token
3733
        )
3734

3735
    def delete_function_event_invoke_config(
1✔
3736
        self,
3737
        context: RequestContext,
3738
        function_name: FunctionName,
3739
        qualifier: NumericLatestPublishedOrAliasQualifier | None = None,
3740
        **kwargs,
3741
    ) -> None:
3742
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3743
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3744
            function_name, qualifier, context
3745
        )
3746
        state = lambda_stores[account_id][region]
1✔
3747
        fn = state.functions.get(function_name)
1✔
3748
        resolved_qualifier = qualifier or "$LATEST"
1✔
3749
        fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3750
        if not fn:
1✔
3751
            raise ResourceNotFoundException(
1✔
3752
                f"The function {fn_arn} doesn't have an EventInvokeConfig", Type="User"
3753
            )
3754

3755
        config = fn.event_invoke_configs.get(resolved_qualifier)
1✔
3756
        if not config:
1✔
3757
            raise ResourceNotFoundException(
1✔
3758
                f"The function {fn_arn} doesn't have an EventInvokeConfig", Type="User"
3759
            )
3760

3761
        del fn.event_invoke_configs[resolved_qualifier]
1✔
3762

3763
    def update_function_event_invoke_config(
1✔
3764
        self,
3765
        context: RequestContext,
3766
        function_name: FunctionName,
3767
        qualifier: NumericLatestPublishedOrAliasQualifier = None,
3768
        maximum_retry_attempts: MaximumRetryAttempts = None,
3769
        maximum_event_age_in_seconds: MaximumEventAgeInSeconds = None,
3770
        destination_config: DestinationConfig = None,
3771
        **kwargs,
3772
    ) -> FunctionEventInvokeConfig:
3773
        # like put but only update single fields via replace
3774
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
3775
        state = lambda_stores[account_id][region]
1✔
3776
        function_name, qualifier = api_utils.get_name_and_qualifier(
1✔
3777
            function_name, qualifier, context
3778
        )
3779

3780
        if (
1✔
3781
            maximum_event_age_in_seconds is None
3782
            and maximum_retry_attempts is None
3783
            and destination_config is None
3784
        ):
UNCOV
3785
            raise InvalidParameterValueException(
×
3786
                "You must specify at least one of error handling or destination setting.",
3787
                Type="User",
3788
            )
3789

3790
        fn = state.functions.get(function_name)
1✔
3791
        if not fn or (qualifier and not (qualifier in fn.aliases or qualifier in fn.versions)):
1✔
3792
            raise ResourceNotFoundException("The function doesn't exist.", Type="User")
1✔
3793

3794
        qualifier = qualifier or "$LATEST"
1✔
3795

3796
        config = fn.event_invoke_configs.get(qualifier)
1✔
3797
        if not config:
1✔
3798
            fn_arn = api_utils.qualified_lambda_arn(function_name, qualifier, account_id, region)
1✔
3799
            raise ResourceNotFoundException(
1✔
3800
                f"The function {fn_arn} doesn't have an EventInvokeConfig", Type="User"
3801
            )
3802

3803
        if destination_config:
1✔
UNCOV
3804
            self._validate_destination_config(state, function_name, destination_config)
×
3805

3806
        optional_kwargs = {
1✔
3807
            k: v
3808
            for k, v in {
3809
                "destination_config": destination_config,
3810
                "maximum_retry_attempts": maximum_retry_attempts,
3811
                "maximum_event_age_in_seconds": maximum_event_age_in_seconds,
3812
            }.items()
3813
            if v is not None
3814
        }
3815

3816
        new_config = dataclasses.replace(
1✔
3817
            config, last_modified=api_utils.generate_lambda_date(), **optional_kwargs
3818
        )
3819
        fn.event_invoke_configs[qualifier] = new_config
1✔
3820

3821
        return FunctionEventInvokeConfig(
1✔
3822
            LastModified=datetime.datetime.strptime(
3823
                new_config.last_modified, api_utils.LAMBDA_DATE_FORMAT
3824
            ),
3825
            FunctionArn=api_utils.qualified_lambda_arn(
3826
                function_name, qualifier or "$LATEST", account_id, region
3827
            ),
3828
            DestinationConfig=new_config.destination_config,
3829
            MaximumEventAgeInSeconds=new_config.maximum_event_age_in_seconds,
3830
            MaximumRetryAttempts=new_config.maximum_retry_attempts,
3831
        )
3832

3833
    # =======================================
3834
    # ======  Layer & Layer Versions  =======
3835
    # =======================================
3836

3837
    @staticmethod
1✔
3838
    def _resolve_layer(
1✔
3839
        layer_name_or_arn: str, context: RequestContext
3840
    ) -> tuple[str, str, str, str | None]:
3841
        """
3842
        Return locator attributes for a given Lambda layer.
3843

3844
        :param layer_name_or_arn: Layer name or ARN
3845
        :param context: Request context
3846
        :return: Tuple of region, account ID, layer name, layer version
3847
        """
3848
        if api_utils.is_layer_arn(layer_name_or_arn):
1✔
3849
            return api_utils.parse_layer_arn(layer_name_or_arn)
1✔
3850

3851
        return context.region, context.account_id, layer_name_or_arn, None
1✔
3852

3853
    def publish_layer_version(
1✔
3854
        self,
3855
        context: RequestContext,
3856
        layer_name: LayerName,
3857
        content: LayerVersionContentInput,
3858
        description: Description | None = None,
3859
        compatible_runtimes: CompatibleRuntimes | None = None,
3860
        license_info: LicenseInfo | None = None,
3861
        compatible_architectures: CompatibleArchitectures | None = None,
3862
        **kwargs,
3863
    ) -> PublishLayerVersionResponse:
3864
        """
3865
        On first use of a LayerName a new layer is created and for each subsequent call with the same LayerName a new version is created.
3866
        Note that there are no $LATEST versions with layers!
3867

3868
        """
3869
        account = context.account_id
1✔
3870
        region = context.region
1✔
3871

3872
        validation_errors = api_utils.validate_layer_runtimes_and_architectures(
1✔
3873
            compatible_runtimes, compatible_architectures
3874
        )
3875
        if validation_errors:
1✔
3876
            raise ValidationException(
1✔
3877
                f"{len(validation_errors)} validation error{'s' if len(validation_errors) > 1 else ''} detected: {'; '.join(validation_errors)}"
3878
            )
3879

3880
        state = lambda_stores[account][region]
1✔
3881
        with self.create_layer_lock:
1✔
3882
            if layer_name not in state.layers:
1✔
3883
                # we don't have a version so create new layer object
3884
                # lock is required to avoid creating two v1 objects for the same name
3885
                layer = Layer(
1✔
3886
                    arn=api_utils.layer_arn(layer_name=layer_name, account=account, region=region)
3887
                )
3888
                state.layers[layer_name] = layer
1✔
3889

3890
        layer = state.layers[layer_name]
1✔
3891
        with layer.next_version_lock:
1✔
3892
            next_version = LambdaLayerVersionIdentifier(
1✔
3893
                account_id=account, region=region, layer_name=layer_name
3894
            ).generate(next_version=layer.next_version)
3895
            # When creating a layer with user defined layer version, it is possible that we
3896
            # create layer versions out of order.
3897
            # ie. a user could replicate layer v2 then layer v1. It is important to always keep the maximum possible
3898
            # value for next layer to avoid overwriting existing versions
3899
            if layer.next_version <= next_version:
1✔
3900
                # We don't need to update layer.next_version if the created version is lower than the "next in line"
3901
                layer.next_version = max(next_version, layer.next_version) + 1
1✔
3902

3903
        # creating a new layer
3904
        if content.get("ZipFile"):
1✔
3905
            code = store_lambda_archive(
1✔
3906
                archive_file=content["ZipFile"],
3907
                function_name=layer_name,
3908
                region_name=region,
3909
                account_id=account,
3910
            )
3911
        else:
3912
            code = store_s3_bucket_archive(
1✔
3913
                archive_bucket=content["S3Bucket"],
3914
                archive_key=content["S3Key"],
3915
                archive_version=content.get("S3ObjectVersion"),
3916
                function_name=layer_name,
3917
                region_name=region,
3918
                account_id=account,
3919
            )
3920

3921
        new_layer_version = LayerVersion(
1✔
3922
            layer_version_arn=api_utils.layer_version_arn(
3923
                layer_name=layer_name,
3924
                account=account,
3925
                region=region,
3926
                version=str(next_version),
3927
            ),
3928
            layer_arn=layer.arn,
3929
            version=next_version,
3930
            description=description or "",
3931
            license_info=license_info,
3932
            compatible_runtimes=compatible_runtimes,
3933
            compatible_architectures=compatible_architectures,
3934
            created=api_utils.generate_lambda_date(),
3935
            code=code,
3936
        )
3937

3938
        layer.layer_versions[str(next_version)] = new_layer_version
1✔
3939

3940
        return api_utils.map_layer_out(new_layer_version)
1✔
3941

3942
    def get_layer_version(
1✔
3943
        self,
3944
        context: RequestContext,
3945
        layer_name: LayerName,
3946
        version_number: LayerVersionNumber,
3947
        **kwargs,
3948
    ) -> GetLayerVersionResponse:
3949
        # TODO: handle layer_name as an ARN
3950

3951
        region_name, account_id, layer_name, _ = LambdaProvider._resolve_layer(layer_name, context)
1✔
3952
        state = lambda_stores[account_id][region_name]
1✔
3953

3954
        layer = state.layers.get(layer_name)
1✔
3955
        if version_number < 1:
1✔
3956
            raise InvalidParameterValueException("Layer Version Cannot be less than 1", Type="User")
1✔
3957
        if layer is None:
1✔
3958
            raise ResourceNotFoundException(
1✔
3959
                "The resource you requested does not exist.", Type="User"
3960
            )
3961
        layer_version = layer.layer_versions.get(str(version_number))
1✔
3962
        if layer_version is None:
1✔
3963
            raise ResourceNotFoundException(
1✔
3964
                "The resource you requested does not exist.", Type="User"
3965
            )
3966
        return api_utils.map_layer_out(layer_version)
1✔
3967

3968
    def get_layer_version_by_arn(
1✔
3969
        self, context: RequestContext, arn: LayerVersionArn, **kwargs
3970
    ) -> GetLayerVersionResponse:
3971
        region_name, account_id, layer_name, layer_version = LambdaProvider._resolve_layer(
1✔
3972
            arn, context
3973
        )
3974

3975
        if not layer_version:
1✔
3976
            raise ValidationException(
1✔
3977
                f"1 validation error detected: Value '{arn}' at 'arn' failed to satisfy constraint: Member must satisfy regular expression pattern: "
3978
                + "(arn:(aws[a-zA-Z-]*)?:lambda:(eusc-)?[a-z]{2}((-gov)|(-iso([a-z]?)))?-[a-z]+-\\d{1}:\\d{12}:layer:[a-zA-Z0-9-_]+:[0-9]+)|(arn:[a-zA-Z0-9-]+:lambda:::awslayer:[a-zA-Z0-9-_]+)"
3979
            )
3980

3981
        store = lambda_stores[account_id][region_name]
1✔
3982
        if not (layers := store.layers.get(layer_name)):
1✔
UNCOV
3983
            raise ResourceNotFoundException(
×
3984
                "The resource you requested does not exist.", Type="User"
3985
            )
3986

3987
        layer_version = layers.layer_versions.get(layer_version)
1✔
3988

3989
        if not layer_version:
1✔
3990
            raise ResourceNotFoundException(
1✔
3991
                "The resource you requested does not exist.", Type="User"
3992
            )
3993

3994
        return api_utils.map_layer_out(layer_version)
1✔
3995

3996
    def list_layers(
1✔
3997
        self,
3998
        context: RequestContext,
3999
        compatible_runtime: Runtime | None = None,
4000
        marker: String | None = None,
4001
        max_items: MaxLayerListItems | None = None,
4002
        compatible_architecture: Architecture | None = None,
4003
        **kwargs,
4004
    ) -> ListLayersResponse:
4005
        validation_errors = []
1✔
4006

4007
        validation_error_arch = api_utils.validate_layer_architecture(compatible_architecture)
1✔
4008
        if validation_error_arch:
1✔
4009
            validation_errors.append(validation_error_arch)
1✔
4010

4011
        validation_error_runtime = api_utils.validate_layer_runtime(compatible_runtime)
1✔
4012
        if validation_error_runtime:
1✔
4013
            validation_errors.append(validation_error_runtime)
1✔
4014

4015
        if validation_errors:
1✔
4016
            raise ValidationException(
1✔
4017
                f"{len(validation_errors)} validation error{'s' if len(validation_errors) > 1 else ''} detected: {';'.join(validation_errors)}"
4018
            )
4019
        # TODO: handle filter: compatible_runtime
4020
        # TODO: handle filter: compatible_architecture
4021

UNCOV
4022
        state = lambda_stores[context.account_id][context.region]
×
4023
        layers = state.layers
×
4024

4025
        # TODO: test how filters behave together with only returning layers here? Does it return the latest "matching" layer, i.e. does it ignore later layer versions that don't match?
4026

4027
        responses: list[LayersListItem] = []
×
4028
        for layer_name, layer in layers.items():
×
4029
            # fetch latest version
UNCOV
4030
            layer_versions = list(layer.layer_versions.values())
×
UNCOV
4031
            sorted(layer_versions, key=lambda x: x.version)
×
UNCOV
4032
            latest_layer_version = layer_versions[-1]
×
UNCOV
4033
            responses.append(
×
4034
                LayersListItem(
4035
                    LayerName=layer_name,
4036
                    LayerArn=layer.arn,
4037
                    LatestMatchingVersion=api_utils.map_layer_out(latest_layer_version),
4038
                )
4039
            )
4040

UNCOV
4041
        responses = PaginatedList(responses)
×
UNCOV
4042
        page, token = responses.get_page(
×
4043
            lambda version: version,
4044
            marker,
4045
            max_items,
4046
        )
4047

UNCOV
4048
        return ListLayersResponse(NextMarker=token, Layers=page)
×
4049

4050
    def list_layer_versions(
1✔
4051
        self,
4052
        context: RequestContext,
4053
        layer_name: LayerName,
4054
        compatible_runtime: Runtime | None = None,
4055
        marker: String | None = None,
4056
        max_items: MaxLayerListItems | None = None,
4057
        compatible_architecture: Architecture | None = None,
4058
        **kwargs,
4059
    ) -> ListLayerVersionsResponse:
4060
        validation_errors = api_utils.validate_layer_runtimes_and_architectures(
1✔
4061
            [compatible_runtime] if compatible_runtime else [],
4062
            [compatible_architecture] if compatible_architecture else [],
4063
        )
4064
        if validation_errors:
1✔
UNCOV
4065
            raise ValidationException(
×
4066
                f"{len(validation_errors)} validation error{'s' if len(validation_errors) > 1 else ''} detected: {';'.join(validation_errors)}"
4067
            )
4068

4069
        region_name, account_id, layer_name, layer_version = LambdaProvider._resolve_layer(
1✔
4070
            layer_name, context
4071
        )
4072
        state = lambda_stores[account_id][region_name]
1✔
4073

4074
        # TODO: Test & handle filter: compatible_runtime
4075
        # TODO: Test & handle filter: compatible_architecture
4076
        all_layer_versions = []
1✔
4077
        layer = state.layers.get(layer_name)
1✔
4078
        if layer is not None:
1✔
4079
            for layer_version in layer.layer_versions.values():
1✔
4080
                all_layer_versions.append(api_utils.map_layer_out(layer_version))
1✔
4081

4082
        all_layer_versions.sort(key=lambda x: x["Version"], reverse=True)
1✔
4083
        all_layer_versions = PaginatedList(all_layer_versions)
1✔
4084
        page, token = all_layer_versions.get_page(
1✔
4085
            lambda version: version["LayerVersionArn"],
4086
            marker,
4087
            max_items,
4088
        )
4089
        return ListLayerVersionsResponse(NextMarker=token, LayerVersions=page)
1✔
4090

4091
    def delete_layer_version(
1✔
4092
        self,
4093
        context: RequestContext,
4094
        layer_name: LayerName,
4095
        version_number: LayerVersionNumber,
4096
        **kwargs,
4097
    ) -> None:
4098
        if version_number < 1:
1✔
4099
            raise InvalidParameterValueException("Layer Version Cannot be less than 1", Type="User")
1✔
4100

4101
        region_name, account_id, layer_name, layer_version = LambdaProvider._resolve_layer(
1✔
4102
            layer_name, context
4103
        )
4104

4105
        store = lambda_stores[account_id][region_name]
1✔
4106
        layer = store.layers.get(layer_name, {})
1✔
4107
        if layer:
1✔
4108
            layer.layer_versions.pop(str(version_number), None)
1✔
4109

4110
    # =======================================
4111
    # =====  Layer Version Permissions  =====
4112
    # =======================================
4113
    # TODO: lock updates that change revision IDs
4114

4115
    def add_layer_version_permission(
1✔
4116
        self,
4117
        context: RequestContext,
4118
        layer_name: LayerName,
4119
        version_number: LayerVersionNumber,
4120
        statement_id: StatementId,
4121
        action: LayerPermissionAllowedAction,
4122
        principal: LayerPermissionAllowedPrincipal,
4123
        organization_id: OrganizationId = None,
4124
        revision_id: String = None,
4125
        **kwargs,
4126
    ) -> AddLayerVersionPermissionResponse:
4127
        # `layer_name` can either be layer name or ARN. It is used to generate error messages.
4128
        # `layer_n` contains the layer name.
4129
        region_name, account_id, layer_n, _ = LambdaProvider._resolve_layer(layer_name, context)
1✔
4130

4131
        if action != "lambda:GetLayerVersion":
1✔
4132
            raise ValidationException(
1✔
4133
                f"1 validation error detected: Value '{action}' at 'action' failed to satisfy constraint: Member must satisfy regular expression pattern: lambda:GetLayerVersion"
4134
            )
4135

4136
        store = lambda_stores[account_id][region_name]
1✔
4137
        layer = store.layers.get(layer_n)
1✔
4138

4139
        layer_version_arn = api_utils.layer_version_arn(
1✔
4140
            layer_name, account_id, region_name, str(version_number)
4141
        )
4142

4143
        if layer is None:
1✔
4144
            raise ResourceNotFoundException(
1✔
4145
                f"Layer version {layer_version_arn} does not exist.", Type="User"
4146
            )
4147
        layer_version = layer.layer_versions.get(str(version_number))
1✔
4148
        if layer_version is None:
1✔
4149
            raise ResourceNotFoundException(
1✔
4150
                f"Layer version {layer_version_arn} does not exist.", Type="User"
4151
            )
4152
        # do we have a policy? if not set one
4153
        if layer_version.policy is None:
1✔
4154
            layer_version.policy = LayerPolicy()
1✔
4155

4156
        if statement_id in layer_version.policy.statements:
1✔
4157
            raise ResourceConflictException(
1✔
4158
                f"The statement id ({statement_id}) provided already exists. Please provide a new statement id, or remove the existing statement.",
4159
                Type="User",
4160
            )
4161

4162
        if revision_id and layer_version.policy.revision_id != revision_id:
1✔
4163
            raise PreconditionFailedException(
1✔
4164
                "The Revision Id provided does not match the latest Revision Id. "
4165
                "Call the GetLayerPolicy API to retrieve the latest Revision Id",
4166
                Type="User",
4167
            )
4168

4169
        statement = LayerPolicyStatement(
1✔
4170
            sid=statement_id, action=action, principal=principal, organization_id=organization_id
4171
        )
4172

4173
        old_statements = layer_version.policy.statements
1✔
4174
        layer_version.policy = dataclasses.replace(
1✔
4175
            layer_version.policy, statements={**old_statements, statement_id: statement}
4176
        )
4177

4178
        return AddLayerVersionPermissionResponse(
1✔
4179
            Statement=json.dumps(
4180
                {
4181
                    "Sid": statement.sid,
4182
                    "Effect": "Allow",
4183
                    "Principal": statement.principal,
4184
                    "Action": statement.action,
4185
                    "Resource": layer_version.layer_version_arn,
4186
                }
4187
            ),
4188
            RevisionId=layer_version.policy.revision_id,
4189
        )
4190

4191
    def remove_layer_version_permission(
1✔
4192
        self,
4193
        context: RequestContext,
4194
        layer_name: LayerName,
4195
        version_number: LayerVersionNumber,
4196
        statement_id: StatementId,
4197
        revision_id: String = None,
4198
        **kwargs,
4199
    ) -> None:
4200
        # `layer_name` can either be layer name or ARN. It is used to generate error messages.
4201
        # `layer_n` contains the layer name.
4202
        region_name, account_id, layer_n, layer_version = LambdaProvider._resolve_layer(
1✔
4203
            layer_name, context
4204
        )
4205

4206
        layer_version_arn = api_utils.layer_version_arn(
1✔
4207
            layer_name, account_id, region_name, str(version_number)
4208
        )
4209

4210
        state = lambda_stores[account_id][region_name]
1✔
4211
        layer = state.layers.get(layer_n)
1✔
4212
        if layer is None:
1✔
4213
            raise ResourceNotFoundException(
1✔
4214
                f"Layer version {layer_version_arn} does not exist.", Type="User"
4215
            )
4216
        layer_version = layer.layer_versions.get(str(version_number))
1✔
4217
        if layer_version is None:
1✔
4218
            raise ResourceNotFoundException(
1✔
4219
                f"Layer version {layer_version_arn} does not exist.", Type="User"
4220
            )
4221

4222
        if revision_id and layer_version.policy.revision_id != revision_id:
1✔
4223
            raise PreconditionFailedException(
1✔
4224
                "The Revision Id provided does not match the latest Revision Id. "
4225
                "Call the GetLayerPolicy API to retrieve the latest Revision Id",
4226
                Type="User",
4227
            )
4228

4229
        if statement_id not in layer_version.policy.statements:
1✔
4230
            raise ResourceNotFoundException(
1✔
4231
                f"Statement {statement_id} is not found in resource policy.", Type="User"
4232
            )
4233

4234
        old_statements = layer_version.policy.statements
1✔
4235
        layer_version.policy = dataclasses.replace(
1✔
4236
            layer_version.policy,
4237
            statements={k: v for k, v in old_statements.items() if k != statement_id},
4238
        )
4239

4240
    def get_layer_version_policy(
1✔
4241
        self,
4242
        context: RequestContext,
4243
        layer_name: LayerName,
4244
        version_number: LayerVersionNumber,
4245
        **kwargs,
4246
    ) -> GetLayerVersionPolicyResponse:
4247
        # `layer_name` can either be layer name or ARN. It is used to generate error messages.
4248
        # `layer_n` contains the layer name.
4249
        region_name, account_id, layer_n, _ = LambdaProvider._resolve_layer(layer_name, context)
1✔
4250

4251
        layer_version_arn = api_utils.layer_version_arn(
1✔
4252
            layer_name, account_id, region_name, str(version_number)
4253
        )
4254

4255
        store = lambda_stores[account_id][region_name]
1✔
4256
        layer = store.layers.get(layer_n)
1✔
4257

4258
        if layer is None:
1✔
4259
            raise ResourceNotFoundException(
1✔
4260
                f"Layer version {layer_version_arn} does not exist.", Type="User"
4261
            )
4262

4263
        layer_version = layer.layer_versions.get(str(version_number))
1✔
4264
        if layer_version is None:
1✔
4265
            raise ResourceNotFoundException(
1✔
4266
                f"Layer version {layer_version_arn} does not exist.", Type="User"
4267
            )
4268

4269
        if layer_version.policy is None:
1✔
4270
            raise ResourceNotFoundException(
1✔
4271
                "No policy is associated with the given resource.", Type="User"
4272
            )
4273

4274
        return GetLayerVersionPolicyResponse(
1✔
4275
            Policy=json.dumps(
4276
                {
4277
                    "Version": layer_version.policy.version,
4278
                    "Id": layer_version.policy.id,
4279
                    "Statement": [
4280
                        {
4281
                            "Sid": ps.sid,
4282
                            "Effect": "Allow",
4283
                            "Principal": ps.principal,
4284
                            "Action": ps.action,
4285
                            "Resource": layer_version.layer_version_arn,
4286
                        }
4287
                        for ps in layer_version.policy.statements.values()
4288
                    ],
4289
                }
4290
            ),
4291
            RevisionId=layer_version.policy.revision_id,
4292
        )
4293

4294
    # =======================================
4295
    # =======  Function Concurrency  ========
4296
    # =======================================
4297
    # (Reserved) function concurrency is scoped to the whole function
4298

4299
    def get_function_concurrency(
1✔
4300
        self, context: RequestContext, function_name: FunctionName, **kwargs
4301
    ) -> GetFunctionConcurrencyResponse:
4302
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
4303
        function_name = api_utils.get_function_name(function_name, context)
1✔
4304
        fn = self._get_function(function_name=function_name, region=region, account_id=account_id)
1✔
4305
        return GetFunctionConcurrencyResponse(
1✔
4306
            ReservedConcurrentExecutions=fn.reserved_concurrent_executions
4307
        )
4308

4309
    def put_function_concurrency(
1✔
4310
        self,
4311
        context: RequestContext,
4312
        function_name: FunctionName,
4313
        reserved_concurrent_executions: ReservedConcurrentExecutions,
4314
        **kwargs,
4315
    ) -> Concurrency:
4316
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
4317

4318
        function_name, qualifier = api_utils.get_name_and_qualifier(function_name, None, context)
1✔
4319
        if qualifier:
1✔
4320
            raise InvalidParameterValueException(
1✔
4321
                "This operation is permitted on Lambda functions only. Aliases and versions do not support this operation. Please specify either a function name or an unqualified function ARN.",
4322
                Type="User",
4323
            )
4324

4325
        store = lambda_stores[account_id][region]
1✔
4326
        fn = store.functions.get(function_name)
1✔
4327
        if not fn:
1✔
4328
            fn_arn = api_utils.qualified_lambda_arn(
1✔
4329
                function_name,
4330
                qualifier="$LATEST",
4331
                account=account_id,
4332
                region=region,
4333
            )
4334
            raise ResourceNotFoundException(f"Function not found: {fn_arn}", Type="User")
1✔
4335

4336
        settings = self.get_account_settings(context)
1✔
4337
        unreserved_concurrent_executions = settings["AccountLimit"][
1✔
4338
            "UnreservedConcurrentExecutions"
4339
        ]
4340

4341
        # The existing reserved concurrent executions for the same function are already deduced in
4342
        # unreserved_concurrent_executions but must not count because the new one will replace the existing one.
4343
        # Joel tested this behavior manually against AWS (2023-11-28).
4344
        existing_reserved_concurrent_executions = (
1✔
4345
            fn.reserved_concurrent_executions if fn.reserved_concurrent_executions else 0
4346
        )
4347
        if (
1✔
4348
            unreserved_concurrent_executions
4349
            - reserved_concurrent_executions
4350
            + existing_reserved_concurrent_executions
4351
        ) < config.LAMBDA_LIMITS_MINIMUM_UNRESERVED_CONCURRENCY:
4352
            raise InvalidParameterValueException(
1✔
4353
                f"Specified ReservedConcurrentExecutions for function decreases account's UnreservedConcurrentExecution below its minimum value of [{config.LAMBDA_LIMITS_MINIMUM_UNRESERVED_CONCURRENCY}]."
4354
            )
4355

4356
        total_provisioned_concurrency = sum(
1✔
4357
            [
4358
                provisioned_configs.provisioned_concurrent_executions
4359
                for provisioned_configs in fn.provisioned_concurrency_configs.values()
4360
            ]
4361
        )
4362
        if total_provisioned_concurrency > reserved_concurrent_executions:
1✔
4363
            raise InvalidParameterValueException(
1✔
4364
                f" ReservedConcurrentExecutions  {reserved_concurrent_executions} should not be lower than function's total provisioned concurrency [{total_provisioned_concurrency}]."
4365
            )
4366

4367
        fn.reserved_concurrent_executions = reserved_concurrent_executions
1✔
4368

4369
        return Concurrency(ReservedConcurrentExecutions=fn.reserved_concurrent_executions)
1✔
4370

4371
    def delete_function_concurrency(
1✔
4372
        self, context: RequestContext, function_name: FunctionName, **kwargs
4373
    ) -> None:
4374
        account_id, region = api_utils.get_account_and_region(function_name, context)
1✔
4375
        function_name, qualifier = api_utils.get_name_and_qualifier(function_name, None, context)
1✔
4376
        store = lambda_stores[account_id][region]
1✔
4377
        fn = store.functions.get(function_name)
1✔
4378
        fn.reserved_concurrent_executions = None
1✔
4379

4380
    # =======================================
4381
    # ===============  TAGS   ===============
4382
    # =======================================
4383
    # only Function, Event Source Mapping, and Code Signing Config (not currently supported by LocalStack) ARNs an are available for tagging in AWS
4384

4385
    def _get_tags(self, resource: TaggableResource) -> dict[str, str]:
1✔
4386
        state = self.fetch_lambda_store_for_tagging(resource)
1✔
4387
        lambda_adapted_tags = {
1✔
4388
            tag["Key"]: tag["Value"]
4389
            for tag in state.TAGS.list_tags_for_resource(resource).get("Tags")
4390
        }
4391
        return lambda_adapted_tags
1✔
4392

4393
    def _store_tags(self, resource: TaggableResource, tags: dict[str, str]):
1✔
4394
        state = self.fetch_lambda_store_for_tagging(resource)
1✔
4395
        if len(state.TAGS.tags.get(resource, {}) | tags) > LAMBDA_TAG_LIMIT_PER_RESOURCE:
1✔
4396
            raise InvalidParameterValueException(
1✔
4397
                "Number of tags exceeds resource tag limit.", Type="User"
4398
            )
4399

4400
        tag_svc_adapted_tags = [{"Key": key, "Value": value} for key, value in tags.items()]
1✔
4401
        state.TAGS.tag_resource(resource, tag_svc_adapted_tags)
1✔
4402

4403
    def fetch_lambda_store_for_tagging(self, resource: TaggableResource) -> LambdaStore:
1✔
4404
        """
4405
        Takes a resource ARN for a TaggableResource (Lambda Function, Event Source Mapping, Code Signing Config, or Capacity Provider) and returns a corresponding
4406
        LambdaStore for its region and account.
4407

4408
        In addition, this function validates that the ARN is a valid TaggableResource type, and that the TaggableResource exists.
4409

4410
        Raises:
4411
            ValidationException: If the resource ARN is not a full ARN for a TaggableResource.
4412
            ResourceNotFoundException: If the specified resource does not exist.
4413
            InvalidParameterValueException: If the resource ARN is a qualified Lambda Function.
4414
        """
4415

4416
        def _raise_validation_exception():
1✔
4417
            raise ValidationException(
1✔
4418
                f"1 validation error detected: Value '{resource}' at 'resource' failed to satisfy constraint: Member must satisfy regular expression pattern: {api_utils.TAGGABLE_RESOURCE_ARN_PATTERN}"
4419
            )
4420

4421
        # Check whether the ARN we have been passed is correctly formatted
4422
        parsed_resource_arn: ArnData = None
1✔
4423
        try:
1✔
4424
            parsed_resource_arn = parse_arn(resource)
1✔
4425
        except Exception:
1✔
4426
            _raise_validation_exception()
1✔
4427

4428
        # TODO: Should we be checking whether this is a full ARN?
4429
        region, account_id, resource_type = map(
1✔
4430
            parsed_resource_arn.get, ("region", "account", "resource")
4431
        )
4432

4433
        if not all((region, account_id, resource_type)):
1✔
UNCOV
4434
            _raise_validation_exception()
×
4435

4436
        if not (parts := resource_type.split(":")):
1✔
UNCOV
4437
            _raise_validation_exception()
×
4438

4439
        resource_type, resource_identifier, *qualifier = parts
1✔
4440

4441
        # Qualifier validation raises before checking for NotFound
4442
        if qualifier:
1✔
4443
            if resource_type == "function":
1✔
4444
                raise InvalidParameterValueException(
1✔
4445
                    "Tags on function aliases and versions are not supported. Please specify a function ARN.",
4446
                    Type="User",
4447
                )
4448
            _raise_validation_exception()
1✔
4449

4450
        if resource_type == "event-source-mapping":
1✔
4451
            self._get_esm(resource_identifier, account_id, region)
1✔
4452
        elif resource_type == "code-signing-config":
1✔
4453
            raise NotImplementedError("Resource tagging on CSC not yet implemented.")
4454
        elif resource_type == "function":
1✔
4455
            self._get_function(
1✔
4456
                function_name=resource_identifier, account_id=account_id, region=region
4457
            )
4458
        elif resource_type == "capacity-provider":
1✔
4459
            self._get_capacity_provider(resource_identifier, account_id, region)
1✔
4460
        else:
4461
            _raise_validation_exception()
1✔
4462

4463
        # If no exceptions are raised, assume ARN and referenced resource is valid for tag operations
4464
        return lambda_stores[account_id][region]
1✔
4465

4466
    def tag_resource(
1✔
4467
        self, context: RequestContext, resource: TaggableResource, tags: Tags, **kwargs
4468
    ) -> None:
4469
        if not tags:
1✔
4470
            raise InvalidParameterValueException(
1✔
4471
                "An error occurred and the request cannot be processed.", Type="User"
4472
            )
4473
        self._store_tags(resource, tags)
1✔
4474

4475
        if (resource_id := extract_resource_from_arn(resource)) and resource_id.startswith(
1✔
4476
            "function"
4477
        ):
4478
            name, _, account, region = function_locators_from_arn(resource)
1✔
4479
            function = self._get_function(name, account, region)
1✔
4480
            with function.lock:
1✔
4481
                # dirty hack for changed revision id, should reevaluate model to prevent this:
4482
                latest_version = function.versions["$LATEST"]
1✔
4483
                function.versions["$LATEST"] = dataclasses.replace(
1✔
4484
                    latest_version, config=dataclasses.replace(latest_version.config)
4485
                )
4486

4487
    def list_tags(
1✔
4488
        self, context: RequestContext, resource: TaggableResource, **kwargs
4489
    ) -> ListTagsResponse:
4490
        tags = self._get_tags(resource)
1✔
4491
        return ListTagsResponse(Tags=tags)
1✔
4492

4493
    def untag_resource(
1✔
4494
        self, context: RequestContext, resource: TaggableResource, tag_keys: TagKeyList, **kwargs
4495
    ) -> None:
4496
        if not tag_keys:
1✔
4497
            raise ValidationException(
1✔
4498
                "1 validation error detected: Value null at 'tagKeys' failed to satisfy constraint: Member must not be null"
4499
            )  # should probably be generalized a bit
4500

4501
        state = self.fetch_lambda_store_for_tagging(resource)
1✔
4502
        state.TAGS.untag_resource(resource, tag_keys)
1✔
4503

4504
        if (resource_id := extract_resource_from_arn(resource)) and resource_id.startswith(
1✔
4505
            "function"
4506
        ):
4507
            name, _, account, region = function_locators_from_arn(resource)
1✔
4508
            function = self._get_function(name, account, region)
1✔
4509
            # TODO: Potential race condition
4510
            with function.lock:
1✔
4511
                # dirty hack for changed revision id, should reevaluate model to prevent this:
4512
                latest_version = function.versions["$LATEST"]
1✔
4513
                function.versions["$LATEST"] = dataclasses.replace(
1✔
4514
                    latest_version, config=dataclasses.replace(latest_version.config)
4515
                )
4516

4517
    # =======================================
4518
    # =======  LEGACY / DEPRECATED   ========
4519
    # =======================================
4520

4521
    def invoke_async(
1✔
4522
        self,
4523
        context: RequestContext,
4524
        function_name: NamespacedFunctionName,
4525
        invoke_args: IO[BlobStream],
4526
        **kwargs,
4527
    ) -> InvokeAsyncResponse:
4528
        """LEGACY API endpoint. Even AWS heavily discourages its usage."""
4529
        raise NotImplementedError
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc