16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

89.88

/localstack-core/localstack/aws/mocking.py

import logging
import math
import random
import re
from datetime import date, datetime
from functools import lru_cache, singledispatch
from typing import cast

import botocore
import networkx
import rstr
from botocore.model import ListShape, MapShape, OperationModel, Shape, StringShape, StructureShape

from localstack.aws.api import RequestContext, ServiceRequest, ServiceResponse
from localstack.aws.skeleton import DispatchTable, ServiceRequestDispatcher, Skeleton
from localstack.aws.spec import load_service
from localstack.utils.sync import retry

LOG = logging.getLogger(__name__)

types = {
    "timestamp",
    "string",
    "blob",
    "map",
    "list",
    "long",
    "structure",
    "integer",
    "double",
    "float",
    "boolean",
}

Instance = (
    dict[str, "Instance"] | list["Instance"] | str | bytes | map | list | float | int | bool | date
)

# https://github.com/boto/botocore/issues/2623
StringShape.METADATA_ATTRS.append("pattern")

words = [
    # a few snazzy six-letter words
    "snazzy",
    "mohawk",
    "poncho",
    "proton",
    "foobar",
    "python",
    "umlaut",
    "except",
    "global",
    "latest",
]

DEFAULT_ARN = "arn:aws:ec2:us-east-1:1234567890123:instance/i-abcde0123456789f"


class ShapeGraph(networkx.DiGraph):
    root: ListShape | StructureShape | MapShape
    cycle: list[tuple[str, str]]
    cycle_shapes: list[str]


def populate_graph(graph: networkx.DiGraph, root: Shape):
    stack: list[Shape] = [root]
    visited: set[str] = set()

    while stack:
        cur = stack.pop()
        if cur is None:
            continue

        if cur.name in visited:
            continue

        visited.add(cur.name)
        graph.add_node(cur.name, shape=cur)

        if isinstance(cur, ListShape):
            graph.add_edge(cur.name, cur.member.name)
            stack.append(cur.member)
        elif isinstance(cur, StructureShape):
            for member in cur.members.values():
                stack.append(member)
                graph.add_edge(cur.name, member.name)
        elif isinstance(cur, MapShape):
            stack.append(cur.key)
            stack.append(cur.value)
            graph.add_edge(cur.name, cur.key.name)
            graph.add_edge(cur.name, cur.value.name)

        else:  # leaf types (int, string, bool, ...)
            pass


def shape_graph(root: Shape) -> ShapeGraph:
    graph = networkx.DiGraph()
    graph.root = root
    populate_graph(graph, root)

    cycles = list()
    shapes = set()
    for node in graph.nodes:
        try:
            cycle = networkx.find_cycle(graph, source=node)
            for k, v in cycle:
                shapes.add(k)
                shapes.add(v)

            if cycle not in cycles:
                cycles.append(cycle)
        except networkx.NetworkXNoCycle:
            pass

    graph.cycles = cycles
    graph.cycle_shapes = list(shapes)

    return cast(ShapeGraph, graph)


def sanitize_pattern(pattern: str) -> str:
    if pattern == "^(https|s3)://([^/]+)/?(.*)$":
        pattern = "^(https|s3)://(\\w+)$"
    pattern = pattern.replace("\\p{XDigit}", "[A-Fa-f0-9]")
    pattern = pattern.replace("\\p{P}", "[.,;]")
    pattern = pattern.replace("\\p{Punct}", "[.,;]")
    pattern = pattern.replace("\\p{N}", "[0-9]")
    pattern = pattern.replace("\\p{L}", "[A-Z]")
    pattern = pattern.replace("\\p{LD}", "[A-Z]")
    pattern = pattern.replace("\\p{Z}", "[ ]")
    pattern = pattern.replace("\\p{S}", "[+\\u-*]")
    pattern = pattern.replace("\\p{M}", "[`]")
    pattern = pattern.replace("\\p{IsLetter}", "[a-zA-Z]")
    pattern = pattern.replace("[:alnum:]", "[a-zA-Z0-9]")
    pattern = pattern.replace("\\p{ASCII}*", "[a-zA-Z0-9]")
    pattern = pattern.replace("\\p{Alnum}", "[a-zA-Z0-9]")

    if "\\p{" in pattern:
        LOG.warning("Find potential additional pattern that need to be sanitized: %s", pattern)
    return pattern


def sanitize_arn_pattern(pattern: str) -> str:
    # clown emoji

    # some devs were just lazy ...
    if pattern in [
        ".*",
        "arn:.*",
        "arn:.+",
        "^arn:.+",
        "arn:aws.*:*",
        "^arn:aws.*",
        "^arn:.*",
        "arn:\\S+",
        ".*\\S.*",
        "^[A-Za-z0-9:\\/_-]*$",
        "^arn[\\/\\:\\-\\_\\.a-zA-Z0-9]+$",
        ".{0,1600}",
        "^arn:[!-~]+$",
        "[\\S]+",
        "[\\s\\S]*",
        "^([\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]*)$",
        "[a-zA-Z0-9_:\\-\\/]+",
    ]:
        pattern = "arn:aws:[a-z]{4}:us-east-1:[0-9]{12}:[a-z]{8}"

    # common pattern to describe a partition
    pattern = pattern.replace("arn:[^:]*:", "arn:aws:")
    pattern = pattern.replace("arn:[a-z\\d-]+", "arn:aws")
    pattern = pattern.replace("arn:[\\w+=\\/,.@-]+", "arn:aws")
    pattern = pattern.replace("arn:[a-z-]+?", "arn:aws")
    pattern = pattern.replace("arn:[a-z0-9][-.a-z0-9]{0,62}", "arn:aws")
    pattern = pattern.replace(":aws(-\\w+)*", ":aws")
    pattern = pattern.replace(":aws[a-z\\-]*", ":aws")
    pattern = pattern.replace(":aws(-[\\w]+)*", ":aws")
    pattern = pattern.replace(":aws[^:\\s]*", ":aws")
    pattern = pattern.replace(":aws[A-Za-z0-9-]{0,64}", ":aws")
    # often the account-id
    pattern = pattern.replace(":[0-9]+:", ":[0-9]{13}:")
    pattern = pattern.replace(":\\w{12}:", ":[0-9]{13}:")
    # substitutions
    pattern = pattern.replace("[a-z\\-\\d]", "[a-z0-9]")
    pattern = pattern.replace(
        "[\\u0020-\\uD7FF\\uE000-\\uFFFD\\uD800\\uDC00-\\uDBFF\\uDFFF\\r\\n\\t]", "[a-z0-9]"
    )
    pattern = pattern.replace("[\\w\\d-]", "[a-z0-9]")
    pattern = pattern.replace("[\\w+=/,.@-]", "[a-z]")
    pattern = pattern.replace("[^:]", "[a-z]")
    pattern = pattern.replace("[^/]", "[a-z]")
    pattern = pattern.replace("\\d+", "[0-9]+")
    pattern = pattern.replace("\\d*", "[0-9]*")
    pattern = pattern.replace("\\S+", "[a-z]{4}")
    pattern = pattern.replace("\\d]", "0-9]")
    pattern = pattern.replace("[a-z\\d", "[a-z0-9")
    pattern = pattern.replace("[a-zA-Z\\d", "[a-z0-9")
    pattern = pattern.replace("^$|", "")
    pattern = pattern.replace("(^$)|", "")
    pattern = pattern.replace("[:/]", "[a-z]")
    pattern = pattern.replace("/.{", "/[a-z]{")
    pattern = pattern.replace(".{", "[a-z]{")
    pattern = pattern.replace("-*", "-")
    pattern = pattern.replace("\\n", "")
    pattern = pattern.replace("\\r", "")
    # quantifiers
    pattern = pattern.replace("{11}{0,1011}", "{11}")
    pattern = pattern.replace("}+", "}")
    pattern = pattern.replace("]*", "]{6}")
    pattern = pattern.replace("]+", "]{6}")
    pattern = pattern.replace(".*", "[a-z]{6}")
    pattern = pattern.replace(".+", "[a-z]{6}")

    return pattern


custom_arns = {
    "DeviceFarmArn": "arn:aws:devicefarm:us-east-1:1234567890123:mydevicefarm",
    "KmsKeyArn": "arn:aws:kms:us-east-1:1234567890123:key/somekmskeythatisawesome",
}


@singledispatch
def generate_instance(shape: Shape, graph: ShapeGraph) -> Instance | None:
    if shape is None:
        return None
    raise ValueError("could not generate shape for type %s" % shape.type_name)


@generate_instance.register
def _(shape: StructureShape, graph: ShapeGraph) -> dict[str, Instance]:
    if shape.is_tagged_union:
        k, v = random.choice(list(shape.members.items()))
        members = {k: v}
    else:
        members = shape.members

    if shape.name in graph.cycle_shapes:
        return {}

    return {
        name: generate_instance(member_shape, graph)
        for name, member_shape in members.items()
        if member_shape.name != shape.name
    }


@generate_instance.register
def _(shape: ListShape, graph: ShapeGraph) -> list[Instance]:
    if shape.name in graph.cycle_shapes:
        return []
    return [generate_instance(shape.member, graph) for _ in range(shape.metadata.get("min", 1))]


@generate_instance.register
def _(shape: MapShape, graph: ShapeGraph) -> dict[str, Instance]:
    if shape.name in graph.cycle_shapes:
        return {}
    return {generate_instance(shape.key, graph): generate_instance(shape.value, graph)}


def generate_arn(shape: StringShape):
    if not shape.metadata:
        return DEFAULT_ARN

    def _generate_arn():
        # some custom hacks
        if shape.name in custom_arns:
            return custom_arns[shape.name]

        max_len = shape.metadata.get("max") or math.inf
        min_len = shape.metadata.get("min") or 0

        pattern = shape.metadata.get("pattern")
        if pattern:
            # FIXME: also conforming to length may be difficult
            pattern = sanitize_arn_pattern(pattern)
            pattern = sanitize_pattern(pattern)
            arn = rstr.xeger(pattern)
        else:
            arn = DEFAULT_ARN

        # if there's a value set for the region, replace with a randomly picked region
        # TODO: splitting the ARNs here by ":" sometimes fails for some reason (e.g. or dynamodb for some reason)
        arn_parts = arn.split(":")
        if len(arn_parts) >= 4:
            region = arn_parts[3]
            if region:
                # TODO: check service in ARN and try to get the actual region for the service
                regions = botocore.session.Session().get_available_regions("lambda")
                picked_region = random.choice(regions)
                arn_parts[3] = picked_region
                arn = ":".join(arn_parts)

        if len(arn) > max_len:
            arn = arn[:max_len]

        if len(arn) < min_len or len(arn) > max_len:
            raise ValueError(
                f"generated arn {arn} for shape {shape.name} does not match constraints {shape.metadata}"
            )

        return arn

    return retry(_generate_arn, retries=10, sleep_before=0, sleep=0)


custom_strings = {"DailyTime": "12:10", "WeeklyTime": "1:12:10"}


@generate_instance.register
def _(shape: StringShape, graph: ShapeGraph) -> str:
    if shape.enum:
        return shape.enum[0]

    if shape.name in custom_strings:
        return custom_strings[shape.name]

    if (
        shape.name.endswith("ARN")
        or shape.name.endswith("Arn")
        or shape.name.endswith("ArnString")
        or shape.name == "AmazonResourceName"
    ):
        try:
            return generate_arn(shape)
        except re.error:
            LOG.error(
                "Could not generate arn pattern for %s, with pattern %s",
                shape.name,
                shape.metadata.get("pattern", "(no pattern set)"),
            )
            return DEFAULT_ARN

    max_len: int = shape.metadata.get("max") or 256
    min_len: int = shape.metadata.get("min") or 0
    str_len = min(min_len or 6, max_len)

    pattern = shape.metadata.get("pattern")

    if not pattern or pattern in [".*", "^.*$", ".+"]:
        if min_len <= 6 and max_len >= 6:
            # pick a random six-letter word, to spice things up. this will be the case most of the time.
            return random.choice(words)
        else:
            return "a" * str_len
    if shape.name == "EndpointId" and pattern == "^[A-Za-z0-9\\-]+[\\.][A-Za-z0-9\\-]+$":
        # there are sometimes issues with this pattern, because it could create invalid host labels, e.g. b6NOZqj5rIMdcta4IKyKRHvZakH90r.-wzuX6tQ-pB-pTNePY2
        # for simplification we just remove the dash for now
        pattern = "^[A-Za-z0-9]+[\\.][A-Za-z0-9]+$"
    pattern = sanitize_pattern(pattern)

    try:
        # try to return something simple first
        random_string = "a" * str_len
        if re.match(pattern, random_string):
            return random_string

        val = rstr.xeger(pattern)
        # TODO: this will break the pattern if the string needs to end with something that we may cut off.
        return val[: min(max_len, len(val))]
    except re.error:
        # TODO: this will likely break the pattern
        LOG.error(
            "Could not generate pattern for %s, with pattern %s",
            shape.name,
            shape.metadata.get("pattern", "(no pattern set)"),
        )
        return "0" * str_len


@generate_instance.register
def _(shape: Shape, graph: ShapeGraph) -> int | float | bool | bytes | date:
    if shape.type_name in ["integer", "long"]:
        return shape.metadata.get("min", 1)
    if shape.type_name in ["float", "double"]:
        return shape.metadata.get("min", 1.0)
    if shape.type_name == "boolean":
        return True
    if shape.type_name == "blob":
        # TODO: better blob generator
        return b"0" * shape.metadata.get("min", 1)
    if shape.type_name == "timestamp":
        return datetime.now()

    raise ValueError("unknown type %s" % shape.type_name)


def generate_response(operation: OperationModel):
    graph = shape_graph(operation.output_shape)
    response = generate_instance(graph.root, graph)
    response.pop("nextToken", None)
    return response


def generate_request(operation: OperationModel):
    graph = shape_graph(operation.input_shape)
    return generate_instance(graph.root, graph)


def return_mock_response(context: RequestContext, request: ServiceRequest) -> ServiceResponse:
    return generate_response(context.operation)


def create_mocking_dispatch_table(service) -> DispatchTable:
    dispatch_table = {}

    for operation in service.operation_names:
        # resolve the bound function of the delegate
        # create a dispatcher
        dispatch_table[operation] = ServiceRequestDispatcher(
            return_mock_response,
            operation=operation,
            pass_context=True,
            expand_parameters=False,
        )

    return dispatch_table


@lru_cache
def get_mocking_skeleton(service: str) -> Skeleton:
    service = load_service(service)
    return Skeleton(service, create_mocking_dispatch_table(service))

1	import logging	1✔
2	import math	1✔
3	import random	1✔
4	import re	1✔
5	from datetime import date, datetime	1✔
6	from functools import lru_cache, singledispatch	1✔
7	from typing import cast	1✔
8
9	import botocore	1✔
10	import networkx	1✔
11	import rstr	1✔
12	from botocore.model import ListShape, MapShape, OperationModel, Shape, StringShape, StructureShape	1✔
13
14	from localstack.aws.api import RequestContext, ServiceRequest, ServiceResponse	1✔
15	from localstack.aws.skeleton import DispatchTable, ServiceRequestDispatcher, Skeleton	1✔
16	from localstack.aws.spec import load_service	1✔
17	from localstack.utils.sync import retry	1✔
18
19	LOG = logging.getLogger(__name__)	1✔
20
21	types = {	1✔
22	"timestamp",
23	"string",
24	"blob",
25	"map",
26	"list",
27	"long",
28	"structure",
29	"integer",
30	"double",
31	"float",
32	"boolean",
33	}
34
35	Instance = (	1✔
36	dict[str, "Instance"] \| list["Instance"] \| str \| bytes \| map \| list \| float \| int \| bool \| date
37	)
38
39	# https://github.com/boto/botocore/issues/2623
40	StringShape.METADATA_ATTRS.append("pattern")	1✔
41
42	words = [	1✔
43	# a few snazzy six-letter words
44	"snazzy",
45	"mohawk",
46	"poncho",
47	"proton",
48	"foobar",
49	"python",
50	"umlaut",
51	"except",
52	"global",
53	"latest",
54	]
55
56	DEFAULT_ARN = "arn:aws:ec2:us-east-1:1234567890123:instance/i-abcde0123456789f"	1✔
57
58
59	class ShapeGraph(networkx.DiGraph):	1✔
60	root: ListShape \| StructureShape \| MapShape	1✔
61	cycle: list[tuple[str, str]]	1✔
62	cycle_shapes: list[str]	1✔
63
64
65	def populate_graph(graph: networkx.DiGraph, root: Shape):	1✔
66	stack: list[Shape] = [root]	1✔
67	visited: set[str] = set()	1✔
68
69	while stack:	1✔
70	cur = stack.pop()	1✔
71	if cur is None:	1✔
UNCOV 72	continue	×
73
74	if cur.name in visited:	1✔
75	continue	1✔
76
77	visited.add(cur.name)	1✔
78	graph.add_node(cur.name, shape=cur)	1✔
79
80	if isinstance(cur, ListShape):	1✔
81	graph.add_edge(cur.name, cur.member.name)	1✔
82	stack.append(cur.member)	1✔
83	elif isinstance(cur, StructureShape):	1✔
84	for member in cur.members.values():	1✔
85	stack.append(member)	1✔
86	graph.add_edge(cur.name, member.name)	1✔
87	elif isinstance(cur, MapShape):	1✔
88	stack.append(cur.key)	1✔
89	stack.append(cur.value)	1✔
90	graph.add_edge(cur.name, cur.key.name)	1✔
91	graph.add_edge(cur.name, cur.value.name)	1✔
92
93	else: # leaf types (int, string, bool, ...)
94	pass	1✔
95
96
97	def shape_graph(root: Shape) -> ShapeGraph:	1✔
98	graph = networkx.DiGraph()	1✔
99	graph.root = root	1✔
100	populate_graph(graph, root)	1✔
101
102	cycles = list()	1✔
103	shapes = set()	1✔
104	for node in graph.nodes:	1✔
105	try:	1✔
106	cycle = networkx.find_cycle(graph, source=node)	1✔
107	for k, v in cycle:	1✔
108	shapes.add(k)	1✔
109	shapes.add(v)	1✔
110
111	if cycle not in cycles:	1✔
112	cycles.append(cycle)	1✔
113	except networkx.NetworkXNoCycle:	1✔
114	pass	1✔
115
116	graph.cycles = cycles	1✔
117	graph.cycle_shapes = list(shapes)	1✔
118
119	return cast(ShapeGraph, graph)	1✔
120
121
122	def sanitize_pattern(pattern: str) -> str:	1✔
123	if pattern == "^(https\|s3)://([^/]+)/?(.*)$":	1✔
UNCOV 124	pattern = "^(https\|s3)://(\\w+)$"	×
125	pattern = pattern.replace("\\p{XDigit}", "[A-Fa-f0-9]")	1✔
126	pattern = pattern.replace("\\p{P}", "[.,;]")	1✔
127	pattern = pattern.replace("\\p{Punct}", "[.,;]")	1✔
128	pattern = pattern.replace("\\p{N}", "[0-9]")	1✔
129	pattern = pattern.replace("\\p{L}", "[A-Z]")	1✔
130	pattern = pattern.replace("\\p{LD}", "[A-Z]")	1✔
131	pattern = pattern.replace("\\p{Z}", "[ ]")	1✔
132	pattern = pattern.replace("\\p{S}", "[+\\u-*]")	1✔
133	pattern = pattern.replace("\\p{M}", "[`]")	1✔
134	pattern = pattern.replace("\\p{IsLetter}", "[a-zA-Z]")	1✔
135	pattern = pattern.replace("[:alnum:]", "[a-zA-Z0-9]")	1✔
136	pattern = pattern.replace("\\p{ASCII}*", "[a-zA-Z0-9]")	1✔
137	pattern = pattern.replace("\\p{Alnum}", "[a-zA-Z0-9]")	1✔
138
139	if "\\p{" in pattern:	1✔
UNCOV 140	LOG.warning("Find potential additional pattern that need to be sanitized: %s", pattern)	×
141	return pattern	1✔
142
143
144	def sanitize_arn_pattern(pattern: str) -> str:	1✔
145	# clown emoji
146
147	# some devs were just lazy ...
148	if pattern in [	1✔
149	".*",
150	"arn:.*",
151	"arn:.+",
152	"^arn:.+",
153	"arn:aws.:",
154	"^arn:aws.*",
155	"^arn:.*",
156	"arn:\\S+",
157	".\\S.",
158	"^[A-Za-z0-9:\\/_-]*$",
159	"^arn[\\/\\:\\-\\_\\.a-zA-Z0-9]+$",
160	".{0,1600}",
161	"^arn:[!-~]+$",
162	"[\\S]+",
163	"[\\s\\S]*",
164	"^([\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]*)$",
165	"[a-zA-Z0-9_:\\-\\/]+",
166	]:
UNCOV 167	pattern = "arn:aws:[a-z]{4}:us-east-1:[0-9]{12}:[a-z]{8}"	×
168
169	# common pattern to describe a partition
170	pattern = pattern.replace("arn:[^:]*:", "arn:aws:")	1✔
171	pattern = pattern.replace("arn:[a-z\\d-]+", "arn:aws")	1✔
172	pattern = pattern.replace("arn:[\\w+=\\/,.@-]+", "arn:aws")	1✔
173	pattern = pattern.replace("arn:[a-z-]+?", "arn:aws")	1✔
174	pattern = pattern.replace("arn:[a-z0-9][-.a-z0-9]{0,62}", "arn:aws")	1✔
175	pattern = pattern.replace(":aws(-\\w+)*", ":aws")	1✔
176	pattern = pattern.replace(":aws[a-z\\-]*", ":aws")	1✔
177	pattern = pattern.replace(":aws(-[\\w]+)*", ":aws")	1✔
178	pattern = pattern.replace(":aws[^:\\s]*", ":aws")	1✔
179	pattern = pattern.replace(":aws[A-Za-z0-9-]{0,64}", ":aws")	1✔
180	# often the account-id
181	pattern = pattern.replace(":[0-9]+:", ":[0-9]{13}:")	1✔
182	pattern = pattern.replace(":\\w{12}:", ":[0-9]{13}:")	1✔
183	# substitutions
184	pattern = pattern.replace("[a-z\\-\\d]", "[a-z0-9]")	1✔
185	pattern = pattern.replace(	1✔
186	"[\\u0020-\\uD7FF\\uE000-\\uFFFD\\uD800\\uDC00-\\uDBFF\\uDFFF\\r\\n\\t]", "[a-z0-9]"
187	)
188	pattern = pattern.replace("[\\w\\d-]", "[a-z0-9]")	1✔
189	pattern = pattern.replace("[\\w+=/,.@-]", "[a-z]")	1✔
190	pattern = pattern.replace("[^:]", "[a-z]")	1✔
191	pattern = pattern.replace("[^/]", "[a-z]")	1✔
192	pattern = pattern.replace("\\d+", "[0-9]+")	1✔
193	pattern = pattern.replace("\\d", "[0-9]")	1✔
194	pattern = pattern.replace("\\S+", "[a-z]{4}")	1✔
195	pattern = pattern.replace("\\d]", "0-9]")	1✔
196	pattern = pattern.replace("[a-z\\d", "[a-z0-9")	1✔
197	pattern = pattern.replace("[a-zA-Z\\d", "[a-z0-9")	1✔
198	pattern = pattern.replace("^$\|", "")	1✔
199	pattern = pattern.replace("(^$)\|", "")	1✔
200	pattern = pattern.replace("[:/]", "[a-z]")	1✔
201	pattern = pattern.replace("/.{", "/[a-z]{")	1✔
202	pattern = pattern.replace(".{", "[a-z]{")	1✔
203	pattern = pattern.replace("-*", "-")	1✔
204	pattern = pattern.replace("\\n", "")	1✔
205	pattern = pattern.replace("\\r", "")	1✔
206	# quantifiers
207	pattern = pattern.replace("{11}{0,1011}", "{11}")	1✔
208	pattern = pattern.replace("}+", "}")	1✔
209	pattern = pattern.replace("]*", "]{6}")	1✔
210	pattern = pattern.replace("]+", "]{6}")	1✔
211	pattern = pattern.replace(".*", "[a-z]{6}")	1✔
212	pattern = pattern.replace(".+", "[a-z]{6}")	1✔
213
214	return pattern	1✔
215
216
217	custom_arns = {	1✔
218	"DeviceFarmArn": "arn:aws:devicefarm:us-east-1:1234567890123:mydevicefarm",
219	"KmsKeyArn": "arn:aws:kms:us-east-1:1234567890123:key/somekmskeythatisawesome",
220	}
221
222
223	@singledispatch	1✔
224	def generate_instance(shape: Shape, graph: ShapeGraph) -> Instance \| None:	1✔
UNCOV 225	if shape is None:	×
UNCOV 226	return None	×
UNCOV 227	raise ValueError("could not generate shape for type %s" % shape.type_name)	×
228
229
230	@generate_instance.register	1✔
231	def _(shape: StructureShape, graph: ShapeGraph) -> dict[str, Instance]:	1✔
232	if shape.is_tagged_union:	1✔
UNCOV 233	k, v = random.choice(list(shape.members.items()))	×
234	members = {k: v}	×
235	else:
236	members = shape.members	1✔
237
238	if shape.name in graph.cycle_shapes:	1✔
239	return {}	1✔
240
241	return {	1✔
242	name: generate_instance(member_shape, graph)
243	for name, member_shape in members.items()
244	if member_shape.name != shape.name
245	}
246
247
248	@generate_instance.register	1✔
249	def _(shape: ListShape, graph: ShapeGraph) -> list[Instance]:	1✔
250	if shape.name in graph.cycle_shapes:	1✔
UNCOV 251	return []	×
252	return [generate_instance(shape.member, graph) for _ in range(shape.metadata.get("min", 1))]	1✔
253
254
255	@generate_instance.register	1✔
256	def _(shape: MapShape, graph: ShapeGraph) -> dict[str, Instance]:	1✔
257	if shape.name in graph.cycle_shapes:	1✔
UNCOV 258	return {}	×
259	return {generate_instance(shape.key, graph): generate_instance(shape.value, graph)}	1✔
260
261
262	def generate_arn(shape: StringShape):	1✔
263	if not shape.metadata:	1✔
UNCOV 264	return DEFAULT_ARN	×
265
266	def _generate_arn():	1✔
267	# some custom hacks
268	if shape.name in custom_arns:	1✔
UNCOV 269	return custom_arns[shape.name]	×
270
271	max_len = shape.metadata.get("max") or math.inf	1✔
272	min_len = shape.metadata.get("min") or 0	1✔
273
274	pattern = shape.metadata.get("pattern")	1✔
275	if pattern:	1✔
276	# FIXME: also conforming to length may be difficult
277	pattern = sanitize_arn_pattern(pattern)	1✔
278	pattern = sanitize_pattern(pattern)	1✔
279	arn = rstr.xeger(pattern)	1✔
280	else:
281	arn = DEFAULT_ARN	1✔
282
283	# if there's a value set for the region, replace with a randomly picked region
284	# TODO: splitting the ARNs here by ":" sometimes fails for some reason (e.g. or dynamodb for some reason)
285	arn_parts = arn.split(":")	1✔
286	if len(arn_parts) >= 4:	1✔
287	region = arn_parts[3]	1✔
288	if region:	1✔
289	# TODO: check service in ARN and try to get the actual region for the service
290	regions = botocore.session.Session().get_available_regions("lambda")	1✔
291	picked_region = random.choice(regions)	1✔
292	arn_parts[3] = picked_region	1✔
293	arn = ":".join(arn_parts)	1✔
294
295	if len(arn) > max_len:	1✔
UNCOV 296	arn = arn[:max_len]	×
297
298	if len(arn) < min_len or len(arn) > max_len:	1✔
UNCOV 299	raise ValueError(	×
300	f"generated arn {arn} for shape {shape.name} does not match constraints {shape.metadata}"
301	)
302
303	return arn	1✔
304
305	return retry(_generate_arn, retries=10, sleep_before=0, sleep=0)	1✔
306
307
308	custom_strings = {"DailyTime": "12:10", "WeeklyTime": "1:12:10"}	1✔
309
310
311	@generate_instance.register	1✔
312	def _(shape: StringShape, graph: ShapeGraph) -> str:	1✔
313	if shape.enum:	1✔
314	return shape.enum[0]	1✔
315
316	if shape.name in custom_strings:	1✔
UNCOV 317	return custom_strings[shape.name]	×
318
319	if (	1✔
320	shape.name.endswith("ARN")
321	or shape.name.endswith("Arn")
322	or shape.name.endswith("ArnString")
323	or shape.name == "AmazonResourceName"
324	):
325	try:	1✔
326	return generate_arn(shape)	1✔
UNCOV 327	except re.error:	×
UNCOV 328	LOG.error(	×
329	"Could not generate arn pattern for %s, with pattern %s",
330	shape.name,
331	shape.metadata.get("pattern", "(no pattern set)"),
332	)
UNCOV 333	return DEFAULT_ARN	×
334
335	max_len: int = shape.metadata.get("max") or 256	1✔
336	min_len: int = shape.metadata.get("min") or 0	1✔
337	str_len = min(min_len or 6, max_len)	1✔
338
339	pattern = shape.metadata.get("pattern")	1✔
340
341	if not pattern or pattern in [".", "^.$", ".+"]:	1✔
342	if min_len <= 6 and max_len >= 6:	1✔
343	# pick a random six-letter word, to spice things up. this will be the case most of the time.
344	return random.choice(words)	1✔
345	else:
UNCOV 346	return "a" * str_len	×
347	if shape.name == "EndpointId" and pattern == "^[A-Za-z0-9\\-]+[\\.][A-Za-z0-9\\-]+$":	1✔
348	# there are sometimes issues with this pattern, because it could create invalid host labels, e.g. b6NOZqj5rIMdcta4IKyKRHvZakH90r.-wzuX6tQ-pB-pTNePY2
349	# for simplification we just remove the dash for now
UNCOV 350	pattern = "^[A-Za-z0-9]+[\\.][A-Za-z0-9]+$"	×
351	pattern = sanitize_pattern(pattern)	1✔
352
353	try:	1✔
354	# try to return something simple first
355	random_string = "a" * str_len	1✔
356	if re.match(pattern, random_string):	1✔
357	return random_string	1✔
358
359	val = rstr.xeger(pattern)	1✔
360	# TODO: this will break the pattern if the string needs to end with something that we may cut off.
361	return val[: min(max_len, len(val))]	1✔
UNCOV 362	except re.error:	×
363	# TODO: this will likely break the pattern
UNCOV 364	LOG.error(	×
365	"Could not generate pattern for %s, with pattern %s",
366	shape.name,
367	shape.metadata.get("pattern", "(no pattern set)"),
368	)
UNCOV 369	return "0" * str_len	×
370
371
372	@generate_instance.register	1✔
373	def _(shape: Shape, graph: ShapeGraph) -> int \| float \| bool \| bytes \| date:	1✔
374	if shape.type_name in ["integer", "long"]:	1✔
375	return shape.metadata.get("min", 1)	1✔
376	if shape.type_name in ["float", "double"]:	1✔
377	return shape.metadata.get("min", 1.0)	1✔
378	if shape.type_name == "boolean":	1✔
379	return True	1✔
380	if shape.type_name == "blob":	1✔
381	# TODO: better blob generator
382	return b"0" * shape.metadata.get("min", 1)	1✔
383	if shape.type_name == "timestamp":	1✔
384	return datetime.now()	1✔
385
UNCOV 386	raise ValueError("unknown type %s" % shape.type_name)	×
387
388
389	def generate_response(operation: OperationModel):	1✔
390	graph = shape_graph(operation.output_shape)	1✔
391	response = generate_instance(graph.root, graph)	1✔
392	response.pop("nextToken", None)	1✔
393	return response	1✔
394
395
396	def generate_request(operation: OperationModel):	1✔
397	graph = shape_graph(operation.input_shape)	1✔
398	return generate_instance(graph.root, graph)	1✔
399
400
401	def return_mock_response(context: RequestContext, request: ServiceRequest) -> ServiceResponse:	1✔
402	return generate_response(context.operation)	1✔
403
404
405	def create_mocking_dispatch_table(service) -> DispatchTable:	1✔
406	dispatch_table = {}	1✔
407
408	for operation in service.operation_names:	1✔
409	# resolve the bound function of the delegate
410	# create a dispatcher
411	dispatch_table[operation] = ServiceRequestDispatcher(	1✔
412	return_mock_response,
413	operation=operation,
414	pass_context=True,
415	expand_parameters=False,
416	)
417
418	return dispatch_table	1✔
419
420
421	@lru_cache	1✔
422	def get_mocking_skeleton(service: str) -> Skeleton:	1✔
423	service = load_service(service)	1✔
424	return Skeleton(service, create_mocking_dispatch_table(service))	1✔

localstack / localstack / 16820655284

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