20357994067

Committed 18 Dec 2025 10:01PM UTC coverage: 86.913% (-0.02%) from 86.929%

Build # 20357994067

Build Type

push

github

Committed by

web-flow

Commit Message

Fix CloudWatch model annotation (#13545)

Run Details

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

1391 existing lines in 33 files now uncovered.

70000 of 80540 relevant lines covered (86.91%)

1.72 hits per line

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

96.0

/localstack-core/localstack/services/lambda_/resource_providers/aws_lambda_permission.py

# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import json
from pathlib import Path
from typing import TypedDict

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceProvider,
    ResourceRequest,
)


class LambdaPermissionProperties(TypedDict):
    Action: str | None
    FunctionName: str | None
    Principal: str | None
    EventSourceToken: str | None
    FunctionUrlAuthType: str | None
    Id: str | None
    PrincipalOrgID: str | None
    SourceAccount: str | None
    SourceArn: str | None


REPEATED_INVOCATION = "repeated_invocation"


class LambdaPermissionProvider(ResourceProvider[LambdaPermissionProperties]):
    TYPE = "AWS::Lambda::Permission"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

    def create(
        self,
        request: ResourceRequest[LambdaPermissionProperties],
    ) -> ProgressEvent[LambdaPermissionProperties]:
        """
        Create a new resource.

        Primary identifier fields:
          - /properties/Id

        Required properties:
          - FunctionName
          - Action
          - Principal

        Create-only properties:
          - /properties/SourceAccount
          - /properties/FunctionUrlAuthType
          - /properties/SourceArn
          - /properties/Principal
          - /properties/Action
          - /properties/FunctionName
          - /properties/EventSourceToken
          - /properties/PrincipalOrgID

        Read-only properties:
          - /properties/Id


        """
        model = request.desired_state
        lambda_client = request.aws_client_factory.lambda_

        params = util.select_attributes(
            model=model, params=["FunctionName", "Action", "Principal", "SourceArn"]
        )

        params["StatementId"] = util.generate_default_name(
            request.stack_name, request.logical_resource_id
        )

        response = lambda_client.add_permission(**params)

        parsed_statement = json.loads(response["Statement"])
        model["Id"] = parsed_statement["Sid"]

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

    def read(
        self,
        request: ResourceRequest[LambdaPermissionProperties],
    ) -> ProgressEvent[LambdaPermissionProperties]:
        """
        Fetch resource information


        """
        raise NotImplementedError

    def delete(
        self,
        request: ResourceRequest[LambdaPermissionProperties],
    ) -> ProgressEvent[LambdaPermissionProperties]:
        """
        Delete a resource


        """
        model = request.desired_state
        lambda_client = request.aws_client_factory.lambda_
        try:
            lambda_client.remove_permission(
                FunctionName=model.get("FunctionName"), StatementId=model["Id"]
            )
        except lambda_client.exceptions.ResourceNotFoundException:
            pass

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

    def update(
        self,
        request: ResourceRequest[LambdaPermissionProperties],
    ) -> ProgressEvent[LambdaPermissionProperties]:
        """
        Update a resource


        """
        model = request.desired_state
        lambda_client = request.aws_client_factory.lambda_

        if not model.get("Id"):
            model["Id"] = request.previous_state["Id"]

        params = util.select_attributes(
            model=model, params=["FunctionName", "Action", "Principal", "SourceArn"]
        )

        try:
            lambda_client.remove_permission(
                FunctionName=model.get("FunctionName"), StatementId=model["Id"]
            )
        except lambda_client.exceptions.ResourceNotFoundException:
            pass

        lambda_client.add_permission(StatementId=model["Id"], **params)

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

1	# LocalStack Resource Provider Scaffolding v2
2	from __future__ import annotations	2✔
3
4	import json	2✔
5	from pathlib import Path	2✔
6	from typing import TypedDict	2✔
7
8	import localstack.services.cloudformation.provider_utils as util	2✔
9	from localstack.services.cloudformation.resource_provider import (	2✔
10	OperationStatus,
11	ProgressEvent,
12	ResourceProvider,
13	ResourceRequest,
14	)
15
16
17	class LambdaPermissionProperties(TypedDict):	2✔
18	Action: str \| None	2✔
19	FunctionName: str \| None	2✔
20	Principal: str \| None	2✔
21	EventSourceToken: str \| None	2✔
22	FunctionUrlAuthType: str \| None	2✔
23	Id: str \| None	2✔
24	PrincipalOrgID: str \| None	2✔
25	SourceAccount: str \| None	2✔
26	SourceArn: str \| None	2✔
27
28
29	REPEATED_INVOCATION = "repeated_invocation"	2✔
30
31
32	class LambdaPermissionProvider(ResourceProvider[LambdaPermissionProperties]):	2✔
33	TYPE = "AWS::Lambda::Permission" # Autogenerated. Don't change	2✔
34	SCHEMA = util.get_schema_path(Path(__file__)) # Autogenerated. Don't change	2✔
35
36	def create(	2✔
37	self,
38	request: ResourceRequest[LambdaPermissionProperties],
39	) -> ProgressEvent[LambdaPermissionProperties]:
40	"""
41	Create a new resource.
42
43	Primary identifier fields:
44	- /properties/Id
45
46	Required properties:
47	- FunctionName
48	- Action
49	- Principal
50
51	Create-only properties:
52	- /properties/SourceAccount
53	- /properties/FunctionUrlAuthType
54	- /properties/SourceArn
55	- /properties/Principal
56	- /properties/Action
57	- /properties/FunctionName
58	- /properties/EventSourceToken
59	- /properties/PrincipalOrgID
60
61	Read-only properties:
62	- /properties/Id
63
64
65	"""
66	model = request.desired_state	2✔
67	lambda_client = request.aws_client_factory.lambda_	2✔
68
69	params = util.select_attributes(	2✔
70	model=model, params=["FunctionName", "Action", "Principal", "SourceArn"]
71	)
72
73	params["StatementId"] = util.generate_default_name(	2✔
74	request.stack_name, request.logical_resource_id
75	)
76
77	response = lambda_client.add_permission(**params)	2✔
78
79	parsed_statement = json.loads(response["Statement"])	2✔
80	model["Id"] = parsed_statement["Sid"]	2✔
81
82	return ProgressEvent(	2✔
83	status=OperationStatus.SUCCESS,
84	resource_model=model,
85	custom_context=request.custom_context,
86	)
87
88	def read(	2✔
89	self,
90	request: ResourceRequest[LambdaPermissionProperties],
91	) -> ProgressEvent[LambdaPermissionProperties]:
92	"""
93	Fetch resource information
94
95
96	"""
97	raise NotImplementedError
98
99	def delete(	2✔
100	self,
101	request: ResourceRequest[LambdaPermissionProperties],
102	) -> ProgressEvent[LambdaPermissionProperties]:
103	"""
104	Delete a resource
105
106
107	"""
108	model = request.desired_state	2✔
109	lambda_client = request.aws_client_factory.lambda_	2✔
110	try:	2✔
111	lambda_client.remove_permission(	2✔
112	FunctionName=model.get("FunctionName"), StatementId=model["Id"]
113	)
114	except lambda_client.exceptions.ResourceNotFoundException:	2✔
115	pass	2✔
116
117	return ProgressEvent(	2✔
118	status=OperationStatus.SUCCESS,
119	resource_model=model,
120	custom_context=request.custom_context,
121	)
122
123	def update(	2✔
124	self,
125	request: ResourceRequest[LambdaPermissionProperties],
126	) -> ProgressEvent[LambdaPermissionProperties]:
127	"""
128	Update a resource
129
130
131	"""
UNCOV 132	model = request.desired_state	1✔
UNCOV 133	lambda_client = request.aws_client_factory.lambda_	1✔
134
UNCOV 135	if not model.get("Id"):	1✔
UNCOV 136	model["Id"] = request.previous_state["Id"]	1✔
137
UNCOV 138	params = util.select_attributes(	1✔
139	model=model, params=["FunctionName", "Action", "Principal", "SourceArn"]
140	)
141
UNCOV 142	try:	1✔
UNCOV 143	lambda_client.remove_permission(	1✔
144	FunctionName=model.get("FunctionName"), StatementId=model["Id"]
145	)
146	except lambda_client.exceptions.ResourceNotFoundException:	×
147	pass	×
148
UNCOV 149	lambda_client.add_permission(StatementId=model["Id"], **params)	1✔
150
UNCOV 151	return ProgressEvent(	1✔
152	status=OperationStatus.SUCCESS,
153	resource_model=model,
154	custom_context=request.custom_context,
155	)

localstack / localstack / 20357994067

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

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