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

85.0

/localstack-core/localstack/dns/models.py

import dataclasses
from collections.abc import Callable
from enum import Enum, auto
from typing import Protocol


class RecordType(Enum):
    A = auto()
    AAAA = auto()
    CNAME = auto()
    TXT = auto()
    MX = auto()
    SOA = auto()
    NS = auto()
    SRV = auto()


@dataclasses.dataclass(frozen=True)
class NameRecord:
    """
    Dataclass of a stored record
    """

    record_type: RecordType
    record_id: str | None = None


@dataclasses.dataclass(frozen=True)
class _TargetRecordBase:
    """
    Dataclass of a stored record
    """

    target: str


@dataclasses.dataclass(frozen=True)
class TargetRecord(NameRecord, _TargetRecordBase):
    pass


@dataclasses.dataclass(frozen=True)
class _SOARecordBase:
    m_name: str
    r_name: str


@dataclasses.dataclass(frozen=True)
class SOARecord(NameRecord, _SOARecordBase):
    pass


@dataclasses.dataclass(frozen=True)
class AliasTarget:
    target: str
    alias_id: str | None = None
    health_check: Callable[[], bool] | None = None


@dataclasses.dataclass(frozen=True)
class _DynamicRecordBase:
    """
    Dataclass of a record that is dynamically determined at query time to return the IP address
    of the LocalStack container
    """

    record_type: RecordType


@dataclasses.dataclass(frozen=True)
class DynamicRecord(NameRecord, _DynamicRecordBase):
    pass


# TODO decide if we need the whole concept of multiple zones in our DNS implementation
class DnsServerProtocol(Protocol):
    def add_host(self, name: str, record: NameRecord) -> None:
        """
        Add a host resolution to the DNS server.
        This will resolve the given host to the record provided, if it matches.

        :param name: Name pattern to add resolution for. Can be arbitrary regex.
        :param record: Record, consisting of a record type, an optional record id, and the attached data.
            Has to be a subclass of a NameRecord, not a NameRecord itself to contain some data.
        """
        pass

    def delete_host(self, name: str, record: NameRecord) -> None:
        """
        Deletes a host resolution from the DNS server.
        Only the name, the record type, and optionally the given record id will be used to find entries to delete.
        All matching entries will be deleted.

        :param name: Name pattern, identically to the one registered with `add_host`
        :param record: Record, ideally identically to the one registered with add_host but only record_type and
            record_id have to match to find the record.

        :raises ValueError: If no record that was previously registered with `add_host` was found which matches the provided record
        """
        pass

    def add_host_pointing_to_localstack(self, name: str) -> None:
        """
        Add a dns name which should be pointing to LocalStack when resolved.

        :param name: Name which should be pointing to LocalStack when resolved
        """
        pass

    def delete_host_pointing_to_localstack(self, name: str) -> None:
        """
        Removes a dns name from pointing to LocalStack

        :param name: Name to be removed
        :raises ValueError: If the host pointing to LocalStack was not previously registered using `add_host_pointing_to_localstack`
        """
        pass

    def add_alias(self, source_name: str, record_type: RecordType, target: AliasTarget) -> None:
        """
        Adds an alias to the DNS, with an optional healthcheck callback.
        When a request which matches `source_name` comes in, the DNS will check the aliases, and if the healthcheck
        (if provided) succeeds, the resolution result for the `target_name` will be returned instead.
        If multiple aliases are registered for the same source_name record_type tuple, and no health checks interfere,
        the server will process requests with the first added alias

        :param source_name: Alias name
        :param record_type: Record type of the alias
        :param target: Target of the alias
        """
        pass

    def delete_alias(self, source_name: str, record_type: RecordType, target: AliasTarget) -> None:
        """
        Removes an alias from the DNS.
        Only the name, the record type, and optionally the given alias id will be used to find entries to delete.
        All matching entries will be deleted.

        :param source_name: Alias name
        :param record_type: Record type of the alias to remove
        :param target: Target of the alias. Only relevant data for deletion will be its id.
        :raises ValueError: If the alias was not previously registered using `add_alias`
        """
        pass

    # TODO: support regex or wildcard?
    # need to update when custom cloudpod destination is enabled
    # has standard list of skips: localstack.services.dns_server.SKIP_PATTERNS
    def add_skip(self, skip_pattern: str) -> None:
        """
        Add a skip pattern to the DNS server.

        A skip pattern will prevent the DNS server from resolving a matching request against it's internal zones or
        aliases, and will directly contact an upstream DNS for resolution.

        This is usually helpful if AWS endpoints are overwritten by internal entries, but we have to reach AWS for
        some reason. (Often used for cloudpods or installers).

        :param skip_pattern: Skip pattern to add. Can be a valid regex.
        """
        pass

    def delete_skip(self, skip_pattern: str) -> None:
        """
        Removes a skip pattern from the DNS server.

        :param skip_pattern: Skip pattern to remove
        :raises ValueError: If the skip pattern was not previously registered using `add_skip`
        """
        pass

    def clear(self):
        """
        Removes all runtime configurations.
        """
        pass

1	import dataclasses	1✔
2	from collections.abc import Callable	1✔
3	from enum import Enum, auto	1✔
4	from typing import Protocol	1✔
5
6
7	class RecordType(Enum):	1✔
8	A = auto()	1✔
9	AAAA = auto()	1✔
10	CNAME = auto()	1✔
11	TXT = auto()	1✔
12	MX = auto()	1✔
13	SOA = auto()	1✔
14	NS = auto()	1✔
15	SRV = auto()	1✔
16
17
18	@dataclasses.dataclass(frozen=True)	1✔
19	class NameRecord:	1✔
20	"""
21	Dataclass of a stored record
22	"""
23
24	record_type: RecordType	1✔
25	record_id: str \| None = None	1✔
26
27
28	@dataclasses.dataclass(frozen=True)	1✔
29	class _TargetRecordBase:	1✔
30	"""
31	Dataclass of a stored record
32	"""
33
34	target: str	1✔
35
36
37	@dataclasses.dataclass(frozen=True)	1✔
38	class TargetRecord(NameRecord, _TargetRecordBase):	1✔
39	pass	1✔
40
41
42	@dataclasses.dataclass(frozen=True)	1✔
43	class _SOARecordBase:	1✔
44	m_name: str	1✔
45	r_name: str	1✔
46
47
48	@dataclasses.dataclass(frozen=True)	1✔
49	class SOARecord(NameRecord, _SOARecordBase):	1✔
50	pass	1✔
51
52
53	@dataclasses.dataclass(frozen=True)	1✔
54	class AliasTarget:	1✔
55	target: str	1✔
56	alias_id: str \| None = None	1✔
57	health_check: Callable[[], bool] \| None = None	1✔
58
59
60	@dataclasses.dataclass(frozen=True)	1✔
61	class _DynamicRecordBase:	1✔
62	"""
63	Dataclass of a record that is dynamically determined at query time to return the IP address
64	of the LocalStack container
65	"""
66
67	record_type: RecordType	1✔
68
69
70	@dataclasses.dataclass(frozen=True)	1✔
71	class DynamicRecord(NameRecord, _DynamicRecordBase):	1✔
72	pass	1✔
73
74
75	# TODO decide if we need the whole concept of multiple zones in our DNS implementation
76	class DnsServerProtocol(Protocol):	1✔
77	def add_host(self, name: str, record: NameRecord) -> None:	1✔
78	"""
79	Add a host resolution to the DNS server.
80	This will resolve the given host to the record provided, if it matches.
81
82	:param name: Name pattern to add resolution for. Can be arbitrary regex.
83	:param record: Record, consisting of a record type, an optional record id, and the attached data.
84	Has to be a subclass of a NameRecord, not a NameRecord itself to contain some data.
85	"""
UNCOV 86	pass	×
87
88	def delete_host(self, name: str, record: NameRecord) -> None:	1✔
89	"""
90	Deletes a host resolution from the DNS server.
91	Only the name, the record type, and optionally the given record id will be used to find entries to delete.
92	All matching entries will be deleted.
93
94	:param name: Name pattern, identically to the one registered with `add_host`
95	:param record: Record, ideally identically to the one registered with add_host but only record_type and
96	record_id have to match to find the record.
97
98	:raises ValueError: If no record that was previously registered with `add_host` was found which matches the provided record
99	"""
UNCOV 100	pass	×
101
102	def add_host_pointing_to_localstack(self, name: str) -> None:	1✔
103	"""
104	Add a dns name which should be pointing to LocalStack when resolved.
105
106	:param name: Name which should be pointing to LocalStack when resolved
107	"""
UNCOV 108	pass	×
109
110	def delete_host_pointing_to_localstack(self, name: str) -> None:	1✔
111	"""
112	Removes a dns name from pointing to LocalStack
113
114	:param name: Name to be removed
115	:raises ValueError: If the host pointing to LocalStack was not previously registered using `add_host_pointing_to_localstack`
116	"""
UNCOV 117	pass	×
118
119	def add_alias(self, source_name: str, record_type: RecordType, target: AliasTarget) -> None:	1✔
120	"""
121	Adds an alias to the DNS, with an optional healthcheck callback.
122	When a request which matches `source_name` comes in, the DNS will check the aliases, and if the healthcheck
123	(if provided) succeeds, the resolution result for the `target_name` will be returned instead.
124	If multiple aliases are registered for the same source_name record_type tuple, and no health checks interfere,
125	the server will process requests with the first added alias
126
127	:param source_name: Alias name
128	:param record_type: Record type of the alias
129	:param target: Target of the alias
130	"""
UNCOV 131	pass	×
132
133	def delete_alias(self, source_name: str, record_type: RecordType, target: AliasTarget) -> None:	1✔
134	"""
135	Removes an alias from the DNS.
136	Only the name, the record type, and optionally the given alias id will be used to find entries to delete.
137	All matching entries will be deleted.
138
139	:param source_name: Alias name
140	:param record_type: Record type of the alias to remove
141	:param target: Target of the alias. Only relevant data for deletion will be its id.
142	:raises ValueError: If the alias was not previously registered using `add_alias`
143	"""
UNCOV 144	pass	×
145
146	# TODO: support regex or wildcard?
147	# need to update when custom cloudpod destination is enabled
148	# has standard list of skips: localstack.services.dns_server.SKIP_PATTERNS
149	def add_skip(self, skip_pattern: str) -> None:	1✔
150	"""
151	Add a skip pattern to the DNS server.
152
153	A skip pattern will prevent the DNS server from resolving a matching request against it's internal zones or
154	aliases, and will directly contact an upstream DNS for resolution.
155
156	This is usually helpful if AWS endpoints are overwritten by internal entries, but we have to reach AWS for
157	some reason. (Often used for cloudpods or installers).
158
159	:param skip_pattern: Skip pattern to add. Can be a valid regex.
160	"""
UNCOV 161	pass	×
162
163	def delete_skip(self, skip_pattern: str) -> None:	1✔
164	"""
165	Removes a skip pattern from the DNS server.
166
167	:param skip_pattern: Skip pattern to remove
168	:raises ValueError: If the skip pattern was not previously registered using `add_skip`
169	"""
UNCOV 170	pass	×
171
172	def clear(self):	1✔
173	"""
174	Removes all runtime configurations.
175	"""
UNCOV 176	pass	×

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