22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

28.13

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

from localstack.aws.api.s3 import InvalidTag
from localstack.aws.api.s3control import Tag, TagList
from localstack.aws.forwarder import NotImplementedAvoidFallbackError
from localstack.services.s3.exceptions import MalformedXML
from localstack.services.s3.models import s3_stores
from localstack.services.s3.utils import TAG_REGEX
from localstack.services.s3control.exceptions import NoSuchResource


def validate_arn_for_tagging(
    resource_arn: str, partition: str, account_id: str, region: str
) -> None:
    """
    Validates the resource ARN for the resource being tagged.

    :param resource_arn: The ARN of the resource being tagged.
    :param partition: The partition the request is originating from.
    :param account_id: The account ID of the target resource.
    :param region: The region the request is originating from.
    :return: None
    """

    s3_prefix = f"arn:{partition}:s3:::"
    if not resource_arn.startswith(s3_prefix):
        # Moto does not support Tagging operations for S3 Control, so we should not forward those operations back
        # to it
        raise NotImplementedAvoidFallbackError(
            "LocalStack only support Bucket tagging operations for S3Control"
        )

    store = s3_stores[account_id][region]
    bucket_name = resource_arn.removeprefix(s3_prefix)
    if bucket_name not in store.global_bucket_map:
        raise NoSuchResource("The specified resource doesn't exist.")


def validate_tags(tags: TagList):
    """
    Validate the tags provided. This is the same function as S3, but with different error messages
    :param tags: a TagList object
    :raises MalformedXML if the object does not conform to the schema
    :raises InvalidTag if the tag key or value are outside the set of validations defined by S3 and S3Control
    :return: None
    """
    keys = set()
    for tag in tags:
        tag: Tag
        if set(tag) != {"Key", "Value"}:
            raise MalformedXML()

        key = tag["Key"]
        value = tag["Value"]

        if key is None or value is None:
            raise MalformedXML()

        if key in keys:
            raise InvalidTag(
                "There are duplicate tag keys in your request. Remove the duplicate tag keys and try again.",
                TagKey=key,
            )

        if key.startswith("aws:"):
            raise InvalidTag(
                'User-defined tag keys can\'t start with "aws:". This prefix is reserved for system tags. Remove "aws:" from your tag keys and try again.',
            )

        if not TAG_REGEX.match(key):
            raise InvalidTag(
                "This request contains a tag key or value that isn't valid. Valid characters include the following: [a-zA-Z+-=._:/]. Tag keys can contain up to 128 characters. Tag values can contain up to 256 characters.",
                TagKey=key,
            )
        elif not TAG_REGEX.match(value):
            raise InvalidTag(
                "This request contains a tag key or value that isn't valid. Valid characters include the following: [a-zA-Z+-=._:/]. Tag keys can contain up to 128 characters. Tag values can contain up to 256 characters.",
                TagKey=key,
                TagValue=value,
            )

        keys.add(key)

1	from localstack.aws.api.s3 import InvalidTag	1✔
2	from localstack.aws.api.s3control import Tag, TagList	1✔
3	from localstack.aws.forwarder import NotImplementedAvoidFallbackError	1✔
4	from localstack.services.s3.exceptions import MalformedXML	1✔
5	from localstack.services.s3.models import s3_stores	1✔
6	from localstack.services.s3.utils import TAG_REGEX	1✔
7	from localstack.services.s3control.exceptions import NoSuchResource	1✔
8
9
10	def validate_arn_for_tagging(	1✔
11	resource_arn: str, partition: str, account_id: str, region: str
12	) -> None:
13	"""
14	Validates the resource ARN for the resource being tagged.
15
16	:param resource_arn: The ARN of the resource being tagged.
17	:param partition: The partition the request is originating from.
18	:param account_id: The account ID of the target resource.
19	:param region: The region the request is originating from.
20	:return: None
21	"""
22
UNCOV 23	s3_prefix = f"arn:{partition}:s3:::"	×
UNCOV 24	if not resource_arn.startswith(s3_prefix):	×
25	# Moto does not support Tagging operations for S3 Control, so we should not forward those operations back
26	# to it
27	raise NotImplementedAvoidFallbackError(
28	"LocalStack only support Bucket tagging operations for S3Control"
29	)
30
UNCOV 31	store = s3_stores[account_id][region]	×
UNCOV 32	bucket_name = resource_arn.removeprefix(s3_prefix)	×
UNCOV 33	if bucket_name not in store.global_bucket_map:	×
UNCOV 34	raise NoSuchResource("The specified resource doesn't exist.")	×
35
36
37	def validate_tags(tags: TagList):	1✔
38	"""
39	Validate the tags provided. This is the same function as S3, but with different error messages
40	:param tags: a TagList object
41	:raises MalformedXML if the object does not conform to the schema
42	:raises InvalidTag if the tag key or value are outside the set of validations defined by S3 and S3Control
43	:return: None
44	"""
UNCOV 45	keys = set()	×
UNCOV 46	for tag in tags:	×
47	tag: Tag
UNCOV 48	if set(tag) != {"Key", "Value"}:	×
49	raise MalformedXML()	×
50
UNCOV 51	key = tag["Key"]	×
UNCOV 52	value = tag["Value"]	×
53
UNCOV 54	if key is None or value is None:	×
UNCOV 55	raise MalformedXML()	×
56
UNCOV 57	if key in keys:	×
UNCOV 58	raise InvalidTag(	×
59	"There are duplicate tag keys in your request. Remove the duplicate tag keys and try again.",
60	TagKey=key,
61	)
62
UNCOV 63	if key.startswith("aws:"):	×
UNCOV 64	raise InvalidTag(	×
65	'User-defined tag keys can\'t start with "aws:". This prefix is reserved for system tags. Remove "aws:" from your tag keys and try again.',
66	)
67
UNCOV 68	if not TAG_REGEX.match(key):	×
UNCOV 69	raise InvalidTag(	×
70	"This request contains a tag key or value that isn't valid. Valid characters include the following: [a-zA-Z+-=._:/]. Tag keys can contain up to 128 characters. Tag values can contain up to 256 characters.",
71	TagKey=key,
72	)
UNCOV 73	elif not TAG_REGEX.match(value):	×
UNCOV 74	raise InvalidTag(	×
75	"This request contains a tag key or value that isn't valid. Valid characters include the following: [a-zA-Z+-=._:/]. Tag keys can contain up to 128 characters. Tag values can contain up to 256 characters.",
76	TagKey=key,
77	TagValue=value,
78	)
79
UNCOV 80	keys.add(key)	×

localstack / localstack / 22709357475

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