21765279428

Committed 06 Feb 2026 08:43PM UTC coverage: 73.535% (-13.3%) from 86.871%

Build # 21765279428

Build Type

Pull #13716

github

Committed by

web-flow

Commit Message

Merge 0f5988375 into 20cc1b384

Pull Request Pull Request #13716: Events: improve Store IAM Statement typing

Run Details

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

9920 existing lines in 273 files now uncovered.

56187 of 76409 relevant lines covered (73.53%)

0.74 hits per line

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

52.38

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

import os

from moto.redshift.models import redshift_backends

from localstack import config
from localstack.aws.api import RequestContext, handler
from localstack.aws.api.redshift import (
    ClusterSecurityGroupMessage,
    DescribeClusterSecurityGroupsMessage,
    RedshiftApi,
)
from localstack.services.moto import call_moto
from localstack.state import AssetDirectory, StateVisitor


class RedshiftProvider(RedshiftApi):
    def accept_state_visitor(self, visitor: StateVisitor):
        visitor.visit(redshift_backends)
        visitor.visit(AssetDirectory(self.service, os.path.join(config.dirs.data, "redshift")))

    @handler("DescribeClusterSecurityGroups", expand=False)
    def describe_cluster_security_groups(
        self,
        context: RequestContext,
        request: DescribeClusterSecurityGroupsMessage,
    ) -> ClusterSecurityGroupMessage:
        result = call_moto(context)
        backend = redshift_backends[context.account_id][context.region]
        for group in result.get("ClusterSecurityGroups", []):
            if group.get("IPRanges"):
                continue
            sgroup = backend.security_groups.get(group["ClusterSecurityGroupName"])
            group["IPRanges"] = [
                {"Status": "authorized", "CIDRIP": ip} for ip in sgroup.ingress_rules
            ]
        return result

1	import os	1✔
2
3	from moto.redshift.models import redshift_backends	1✔
4
5	from localstack import config	1✔
6	from localstack.aws.api import RequestContext, handler	1✔
7	from localstack.aws.api.redshift import (	1✔
8	ClusterSecurityGroupMessage,
9	DescribeClusterSecurityGroupsMessage,
10	RedshiftApi,
11	)
12	from localstack.services.moto import call_moto	1✔
13	from localstack.state import AssetDirectory, StateVisitor	1✔
14
15
16	class RedshiftProvider(RedshiftApi):	1✔
17	def accept_state_visitor(self, visitor: StateVisitor):	1✔
18	visitor.visit(redshift_backends)	×
19	visitor.visit(AssetDirectory(self.service, os.path.join(config.dirs.data, "redshift")))	×
20
21	@handler("DescribeClusterSecurityGroups", expand=False)	1✔
22	def describe_cluster_security_groups(	1✔
23	self,
24	context: RequestContext,
25	request: DescribeClusterSecurityGroupsMessage,
26	) -> ClusterSecurityGroupMessage:
UNCOV 27	result = call_moto(context)	×
UNCOV 28	backend = redshift_backends[context.account_id][context.region]	×
UNCOV 29	for group in result.get("ClusterSecurityGroups", []):	×
UNCOV 30	if group.get("IPRanges"):	×
31	continue	×
UNCOV 32	sgroup = backend.security_groups.get(group["ClusterSecurityGroupName"])	×
UNCOV 33	group["IPRanges"] = [	×
34	{"Status": "authorized", "CIDRIP": ip} for ip in sgroup.ingress_rules
35	]
UNCOV 36	return result	×

localstack / localstack / 21765279428

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