21697093787

Committed 04 Feb 2026 09:56PM UTC coverage: 86.962% (-0.004%) from 86.966%

Build # 21697093787

Build Type

push

github

Committed by

web-flow

Commit Message

improve system information sent in session and container_info (#13680)

Run Details

10 of 17 new or added lines in 2 files covered. (58.82%)

222 existing lines in 17 files now uncovered.

70560 of 81139 relevant lines covered (86.96%)

0.87 hits per line

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

94.29

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

from localstack.aws.api import RequestContext
from localstack.aws.api.s3control import (
    AccountId,
    ListTagsForResourceResult,
    S3ControlApi,
    S3ResourceArn,
    TagKeyList,
    TagList,
    TagResourceResult,
    UntagResourceResult,
)
from localstack.services.s3.models import S3Store, s3_stores
from localstack.services.s3control.validation import validate_arn_for_tagging, validate_tags
from localstack.state import StateVisitor


class S3ControlProvider(S3ControlApi):
    def accept_state_visitor(self, visitor: StateVisitor):
        from moto.s3control.models import s3control_backends

        visitor.visit(s3control_backends)

    """
    S3Control is a management interface for S3, and can access some of its internals with no public API
    This requires us to access the s3 stores directly
    """

    @staticmethod
    def get_s3_store(account_id: str, region: str) -> S3Store:
        return s3_stores[account_id][region]

    def _tag_bucket_resource(
        self, resource_arn: str, partition: str, region: str, account_id: str, tags: TagList
    ) -> None:
        tagging_service = self.get_s3_store(account_id, region).TAGS
        tagging_service.tag_resource(resource_arn, tags)

    def _untag_bucket_resource(
        self, resource_arn: str, partition: str, region: str, account_id: str, tag_keys: TagKeyList
    ) -> None:
        tagging_service = self.get_s3_store(account_id, region).TAGS
        tagging_service.untag_resource(resource_arn, tag_keys)

    def _list_bucket_tags(
        self, resource_arn: str, partition: str, region: str, account_id: str
    ) -> TagList:
        tagging_service = self.get_s3_store(account_id, region).TAGS
        return tagging_service.list_tags_for_resource(resource_arn)["Tags"]

    def tag_resource(
        self,
        context: RequestContext,
        account_id: AccountId,
        resource_arn: S3ResourceArn,
        tags: TagList,
        **kwargs,
    ) -> TagResourceResult:
        # Currently S3Control only supports tagging buckets
        validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)
        validate_tags(tags)

        self._tag_bucket_resource(resource_arn, context.partition, context.region, account_id, tags)
        return TagResourceResult()

    def untag_resource(
        self,
        context: RequestContext,
        account_id: AccountId,
        resource_arn: S3ResourceArn,
        tag_keys: TagKeyList,
        **kwargs,
    ) -> UntagResourceResult:
        # Currently S3Control only supports tagging buckets
        validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)

        self._untag_bucket_resource(
            resource_arn, context.partition, context.region, account_id, tag_keys
        )
        return TagResourceResult()

    def list_tags_for_resource(
        self, context: RequestContext, account_id: AccountId, resource_arn: S3ResourceArn, **kwargs
    ) -> ListTagsForResourceResult:
        # Currently S3Control only supports tagging buckets
        validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)

        tags = self._list_bucket_tags(resource_arn, context.partition, context.region, account_id)
        return ListTagsForResourceResult(Tags=tags)

1	from localstack.aws.api import RequestContext	1✔
2	from localstack.aws.api.s3control import (	1✔
3	AccountId,
4	ListTagsForResourceResult,
5	S3ControlApi,
6	S3ResourceArn,
7	TagKeyList,
8	TagList,
9	TagResourceResult,
10	UntagResourceResult,
11	)
12	from localstack.services.s3.models import S3Store, s3_stores	1✔
13	from localstack.services.s3control.validation import validate_arn_for_tagging, validate_tags	1✔
14	from localstack.state import StateVisitor	1✔
15
16
17	class S3ControlProvider(S3ControlApi):	1✔
18	def accept_state_visitor(self, visitor: StateVisitor):	1✔
UNCOV 19	from moto.s3control.models import s3control_backends	×
20
UNCOV 21	visitor.visit(s3control_backends)	×
22
23	"""	1✔
24	S3Control is a management interface for S3, and can access some of its internals with no public API
25	This requires us to access the s3 stores directly
26	"""
27
28	@staticmethod	1✔
29	def get_s3_store(account_id: str, region: str) -> S3Store:	1✔
30	return s3_stores[account_id][region]	1✔
31
32	def _tag_bucket_resource(	1✔
33	self, resource_arn: str, partition: str, region: str, account_id: str, tags: TagList
34	) -> None:
35	tagging_service = self.get_s3_store(account_id, region).TAGS	1✔
36	tagging_service.tag_resource(resource_arn, tags)	1✔
37
38	def _untag_bucket_resource(	1✔
39	self, resource_arn: str, partition: str, region: str, account_id: str, tag_keys: TagKeyList
40	) -> None:
41	tagging_service = self.get_s3_store(account_id, region).TAGS	1✔
42	tagging_service.untag_resource(resource_arn, tag_keys)	1✔
43
44	def _list_bucket_tags(	1✔
45	self, resource_arn: str, partition: str, region: str, account_id: str
46	) -> TagList:
47	tagging_service = self.get_s3_store(account_id, region).TAGS	1✔
48	return tagging_service.list_tags_for_resource(resource_arn)["Tags"]	1✔
49
50	def tag_resource(	1✔
51	self,
52	context: RequestContext,
53	account_id: AccountId,
54	resource_arn: S3ResourceArn,
55	tags: TagList,
56	**kwargs,
57	) -> TagResourceResult:
58	# Currently S3Control only supports tagging buckets
59	validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)	1✔
60	validate_tags(tags)	1✔
61
62	self._tag_bucket_resource(resource_arn, context.partition, context.region, account_id, tags)	1✔
63	return TagResourceResult()	1✔
64
65	def untag_resource(	1✔
66	self,
67	context: RequestContext,
68	account_id: AccountId,
69	resource_arn: S3ResourceArn,
70	tag_keys: TagKeyList,
71	**kwargs,
72	) -> UntagResourceResult:
73	# Currently S3Control only supports tagging buckets
74	validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)	1✔
75
76	self._untag_bucket_resource(	1✔
77	resource_arn, context.partition, context.region, account_id, tag_keys
78	)
79	return TagResourceResult()	1✔
80
81	def list_tags_for_resource(	1✔
82	self, context: RequestContext, account_id: AccountId, resource_arn: S3ResourceArn, **kwargs
83	) -> ListTagsForResourceResult:
84	# Currently S3Control only supports tagging buckets
85	validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)	1✔
86
87	tags = self._list_bucket_tags(resource_arn, context.partition, context.region, account_id)	1✔
88	return ListTagsForResourceResult(Tags=tags)	1✔

localstack / localstack / 21697093787

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