22334798432

Committed 23 Feb 2026 06:42PM UTC coverage: 86.956% (-0.02%) from 86.973%

Build # 22334798432

Build Type

push

github

Committed by

web-flow

Commit Message

S3: regenerate test snapshots & parity fixes (#13824)

Run Details

69831 of 80306 relevant lines covered (86.96%)

0.87 hits per line

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

93.1

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

from localstack.aws.api import RequestContext
from localstack.aws.api.s3control import (
    AccountId,
    ListTagsForResourceResult,
    S3ControlApi,
    S3ResourceArn,
    Tag,
    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_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)

        store = self.get_s3_store(account_id, context.region)
        store.tags.update_tags(resource_arn, {tag["Key"]: tag["Value"] for tag in 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)

        store = self.get_s3_store(account_id, context.region)
        store.tags.delete_tags(resource_arn, 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)

        store = self.get_s3_store(account_id, context.region)
        tags = store.tags.get_tags(resource_arn)
        return ListTagsForResourceResult(
            Tags=[Tag(Key=key, Value=value) for key, value in tags.items()]
        )

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	Tag,
8	TagKeyList,
9	TagList,
10	TagResourceResult,
11	UntagResourceResult,
12	)
13	from localstack.services.s3.models import S3Store, s3_stores	1✔
14	from localstack.services.s3control.validation import validate_arn_for_tagging, validate_tags	1✔
15	from localstack.state import StateVisitor	1✔
16
17
18	class S3ControlProvider(S3ControlApi):	1✔
19	def accept_state_visitor(self, visitor: StateVisitor):	1✔
20	from moto.s3control.models import s3control_backends	×
21
22	visitor.visit(s3control_backends)	×
23
24	"""	1✔
25	S3Control is a management interface for S3, and can access some of its internals with no public API
26	This requires us to access the s3 stores directly
27	"""
28
29	@staticmethod	1✔
30	def get_s3_store(account_id: str, region: str) -> S3Store:	1✔
31	return s3_stores[account_id][region]	1✔
32
33	def tag_resource(	1✔
34	self,
35	context: RequestContext,
36	account_id: AccountId,
37	resource_arn: S3ResourceArn,
38	tags: TagList,
39	**kwargs,
40	) -> TagResourceResult:
41	# Currently S3Control only supports tagging buckets
42	validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)	1✔
43	validate_tags(tags)	1✔
44
45	store = self.get_s3_store(account_id, context.region)	1✔
46	store.tags.update_tags(resource_arn, {tag["Key"]: tag["Value"] for tag in tags})	1✔
47	return TagResourceResult()	1✔
48
49	def untag_resource(	1✔
50	self,
51	context: RequestContext,
52	account_id: AccountId,
53	resource_arn: S3ResourceArn,
54	tag_keys: TagKeyList,
55	**kwargs,
56	) -> UntagResourceResult:
57	# Currently S3Control only supports tagging buckets
58	validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)	1✔
59
60	store = self.get_s3_store(account_id, context.region)	1✔
61	store.tags.delete_tags(resource_arn, tag_keys)	1✔
62	return TagResourceResult()	1✔
63
64	def list_tags_for_resource(	1✔
65	self, context: RequestContext, account_id: AccountId, resource_arn: S3ResourceArn, **kwargs
66	) -> ListTagsForResourceResult:
67	# Currently S3Control only supports tagging buckets
68	validate_arn_for_tagging(resource_arn, context.partition, account_id, context.region)	1✔
69
70	store = self.get_s3_store(account_id, context.region)	1✔
71	tags = store.tags.get_tags(resource_arn)	1✔
72	return ListTagsForResourceResult(	1✔
73	Tags=[Tag(Key=key, Value=value) for key, value in tags.items()]
74	)

localstack / localstack / 22334798432

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