19880423371

Committed 02 Dec 2025 08:25PM UTC coverage: 86.905% (-0.04%) from 86.945%

Build # 19880423371

Build Type

push

github

Committed by

web-flow

Commit Message

fix/external client CA bundle (#13451)

Run Details

1 of 5 new or added lines in 1 file covered. (20.0%)

414 existing lines in 19 files now uncovered.

69738 of 80246 relevant lines covered (86.91%)

0.87 hits per line

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

82.61

/localstack-core/localstack/services/route53/provider.py

from datetime import datetime

import moto.route53.models as route53_models
from botocore.exceptions import ClientError
from moto.route53.models import route53_backends

from localstack.aws.api import RequestContext
from localstack.aws.api.route53 import (
    VPC,
    ChangeInfo,
    ChangeStatus,
    CreateHostedZoneResponse,
    DeleteHealthCheckResponse,
    DNSName,
    GetChangeResponse,
    GetHealthCheckResponse,
    HealthCheck,
    HealthCheckId,
    HostedZoneConfig,
    InvalidVPCId,
    Nonce,
    NoSuchHealthCheck,
    ResourceId,
    Route53Api,
)
from localstack.aws.connect import connect_to
from localstack.services.moto import call_moto
from localstack.services.plugins import ServiceLifecycleHook
from localstack.state import StateVisitor


class Route53Provider(Route53Api, ServiceLifecycleHook):
    def accept_state_visitor(self, visitor: StateVisitor):
        from localstack.services.route53.models import route53_stores

        visitor.visit(route53_backends)
        visitor.visit(route53_stores)

    def create_hosted_zone(
        self,
        context: RequestContext,
        name: DNSName,
        caller_reference: Nonce,
        vpc: VPC = None,
        hosted_zone_config: HostedZoneConfig = None,
        delegation_set_id: ResourceId = None,
        **kwargs,
    ) -> CreateHostedZoneResponse:
        # private hosted zones cannot be created in a VPC that does not exist
        # check that the VPC exists
        if vpc:
            vpc_id = vpc.get("VPCId")
            vpc_region = vpc.get("VPCRegion")
            if not vpc_id or not vpc_region:
                raise Exception(
                    "VPCId and VPCRegion must be specified when creating a private hosted zone"
                )
            try:
                connect_to(
                    aws_access_key_id=context.account_id, region_name=vpc_region
                ).ec2.describe_vpcs(VpcIds=[vpc_id])
            except ClientError as e:
                if e.response.get("Error", {}).get("Code") == "InvalidVpcID.NotFound":
                    raise InvalidVPCId("The VPC ID is invalid.", sender_fault=True) from e
                raise e

        response = call_moto(context)

        # moto does not populate the VPC struct of the response if creating a private hosted zone
        if (
            hosted_zone_config
            and hosted_zone_config.get("PrivateZone", False)
            and "VPC" in response
            and vpc
        ):
            response["VPC"]["VPCId"] = response["VPC"]["VPCId"] or vpc.get("VPCId", "")
            response["VPC"]["VPCRegion"] = response["VPC"]["VPCRegion"] or vpc.get("VPCRegion", "")

        return response

    def get_change(self, context: RequestContext, id: ResourceId, **kwargs) -> GetChangeResponse:
        change_info = ChangeInfo(Id=id, Status=ChangeStatus.INSYNC, SubmittedAt=datetime.now())
        return GetChangeResponse(ChangeInfo=change_info)

    def get_health_check(
        self, context: RequestContext, health_check_id: HealthCheckId, **kwargs
    ) -> GetHealthCheckResponse:
        health_check: route53_models.HealthCheck | None = route53_backends[context.account_id][
            context.partition
        ].health_checks.get(health_check_id, None)
        if not health_check:
            raise NoSuchHealthCheck(
                f"No health check exists with the specified ID {health_check_id}"
            )
        health_check_config = {
            "Disabled": health_check.disabled,
            "EnableSNI": health_check.enable_sni,
            "FailureThreshold": health_check.failure_threshold,
            "FullyQualifiedDomainName": health_check.fqdn,
            "HealthThreshold": health_check.health_threshold,
            "Inverted": health_check.inverted,
            "IPAddress": health_check.ip_address,
            "MeasureLatency": health_check.measure_latency,
            "Port": health_check.port,
            "RequestInterval": health_check.request_interval,
            "ResourcePath": health_check.resource_path,
            "Type": health_check.type_,
        }
        return GetHealthCheckResponse(
            HealthCheck=HealthCheck(
                Id=health_check.id,
                CallerReference=health_check.caller_reference,
                HealthCheckConfig=health_check_config,
            )
        )

    def delete_health_check(
        self, context: RequestContext, health_check_id: HealthCheckId, **kwargs
    ) -> DeleteHealthCheckResponse:
        if (
            health_check_id
            not in route53_backends[context.account_id][context.partition].health_checks
        ):
            raise NoSuchHealthCheck(
                f"No health check exists with the specified ID {health_check_id}"
            )

        route53_backends[context.account_id][context.partition].delete_health_check(health_check_id)
        return {}

1	from datetime import datetime	1✔
2
3	import moto.route53.models as route53_models	1✔
4	from botocore.exceptions import ClientError	1✔
5	from moto.route53.models import route53_backends	1✔
6
7	from localstack.aws.api import RequestContext	1✔
8	from localstack.aws.api.route53 import (	1✔
9	VPC,
10	ChangeInfo,
11	ChangeStatus,
12	CreateHostedZoneResponse,
13	DeleteHealthCheckResponse,
14	DNSName,
15	GetChangeResponse,
16	GetHealthCheckResponse,
17	HealthCheck,
18	HealthCheckId,
19	HostedZoneConfig,
20	InvalidVPCId,
21	Nonce,
22	NoSuchHealthCheck,
23	ResourceId,
24	Route53Api,
25	)
26	from localstack.aws.connect import connect_to	1✔
27	from localstack.services.moto import call_moto	1✔
28	from localstack.services.plugins import ServiceLifecycleHook	1✔
29	from localstack.state import StateVisitor	1✔
30
31
32	class Route53Provider(Route53Api, ServiceLifecycleHook):	1✔
33	def accept_state_visitor(self, visitor: StateVisitor):	1✔
UNCOV 34	from localstack.services.route53.models import route53_stores	×
35
UNCOV 36	visitor.visit(route53_backends)	×
UNCOV 37	visitor.visit(route53_stores)	×
38
39	def create_hosted_zone(	1✔
40	self,
41	context: RequestContext,
42	name: DNSName,
43	caller_reference: Nonce,
44	vpc: VPC = None,
45	hosted_zone_config: HostedZoneConfig = None,
46	delegation_set_id: ResourceId = None,
47	**kwargs,
48	) -> CreateHostedZoneResponse:
49	# private hosted zones cannot be created in a VPC that does not exist
50	# check that the VPC exists
51	if vpc:	1✔
52	vpc_id = vpc.get("VPCId")	1✔
53	vpc_region = vpc.get("VPCRegion")	1✔
54	if not vpc_id or not vpc_region:	1✔
UNCOV 55	raise Exception(	×
56	"VPCId and VPCRegion must be specified when creating a private hosted zone"
57	)
58	try:	1✔
59	connect_to(	1✔
60	aws_access_key_id=context.account_id, region_name=vpc_region
61	).ec2.describe_vpcs(VpcIds=[vpc_id])
62	except ClientError as e:	1✔
63	if e.response.get("Error", {}).get("Code") == "InvalidVpcID.NotFound":	1✔
64	raise InvalidVPCId("The VPC ID is invalid.", sender_fault=True) from e	1✔
UNCOV 65	raise e	×
66
67	response = call_moto(context)	1✔
68
69	# moto does not populate the VPC struct of the response if creating a private hosted zone
70	if (	1✔
71	hosted_zone_config
72	and hosted_zone_config.get("PrivateZone", False)
73	and "VPC" in response
74	and vpc
75	):
76	response["VPC"]["VPCId"] = response["VPC"]["VPCId"] or vpc.get("VPCId", "")	1✔
77	response["VPC"]["VPCRegion"] = response["VPC"]["VPCRegion"] or vpc.get("VPCRegion", "")	1✔
78
79	return response	1✔
80
81	def get_change(self, context: RequestContext, id: ResourceId, **kwargs) -> GetChangeResponse:	1✔
UNCOV 82	change_info = ChangeInfo(Id=id, Status=ChangeStatus.INSYNC, SubmittedAt=datetime.now())	×
UNCOV 83	return GetChangeResponse(ChangeInfo=change_info)	×
84
85	def get_health_check(	1✔
86	self, context: RequestContext, health_check_id: HealthCheckId, **kwargs
87	) -> GetHealthCheckResponse:
88	health_check: route53_models.HealthCheck \| None = route53_backends[context.account_id][	1✔
89	context.partition
90	].health_checks.get(health_check_id, None)
91	if not health_check:	1✔
UNCOV 92	raise NoSuchHealthCheck(	×
93	f"No health check exists with the specified ID {health_check_id}"
94	)
95	health_check_config = {	1✔
96	"Disabled": health_check.disabled,
97	"EnableSNI": health_check.enable_sni,
98	"FailureThreshold": health_check.failure_threshold,
99	"FullyQualifiedDomainName": health_check.fqdn,
100	"HealthThreshold": health_check.health_threshold,
101	"Inverted": health_check.inverted,
102	"IPAddress": health_check.ip_address,
103	"MeasureLatency": health_check.measure_latency,
104	"Port": health_check.port,
105	"RequestInterval": health_check.request_interval,
106	"ResourcePath": health_check.resource_path,
107	"Type": health_check.type_,
108	}
109	return GetHealthCheckResponse(	1✔
110	HealthCheck=HealthCheck(
111	Id=health_check.id,
112	CallerReference=health_check.caller_reference,
113	HealthCheckConfig=health_check_config,
114	)
115	)
116
117	def delete_health_check(	1✔
118	self, context: RequestContext, health_check_id: HealthCheckId, **kwargs
119	) -> DeleteHealthCheckResponse:
120	if (	1✔
121	health_check_id
122	not in route53_backends[context.account_id][context.partition].health_checks
123	):
124	raise NoSuchHealthCheck(	1✔
125	f"No health check exists with the specified ID {health_check_id}"
126	)
127
128	route53_backends[context.account_id][context.partition].delete_health_check(health_check_id)	1✔
129	return {}	1✔

localstack / localstack / 19880423371

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