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

82.61

/localstack-core/localstack/services/s3/codec.py

import io
from typing import IO, Any


class AwsChunkedDecoder(io.RawIOBase):
    """
    This helper class takes a IO[bytes] stream, and decodes it on the fly, so that S3 can directly access the stream
    without worrying about implementation details of `aws-chunked`.
    It needs to expose the trailing headers, which will be available once the stream is fully read.
    You can also directly pass the S3 Object, so the stream would set the checksum value once it's done.
    See `aws-chunked` format here: https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html
    """

    def readable(self):
        return True

    def __init__(
        self, stream: IO[bytes], decoded_content_length: int, s3_object: Any | None = None
    ):
        self._stream = stream

        self._decoded_length = decoded_content_length  # Length of the encoded object
        self._new_chunk = True
        self._end_chunk = False
        self._trailing_set = False
        self._chunk_size = 0
        self._trailing_headers = {}
        self.s3_object = s3_object

    @property
    def trailing_headers(self):
        if not self._trailing_set:
            raise AttributeError(
                "The stream has not been fully read yet, the trailing headers are not available."
            )
        return self._trailing_headers

    def seekable(self):
        return self._stream.seekable()

    def readinto(self, b):
        with memoryview(b) as view, view.cast("B") as byte_view:
            data = self.read(len(byte_view))
            byte_view[: len(data)] = data
        return len(data)

    def read(self, size=-1):
        """
        Read from the underlying stream, and return at most `size` decoded bytes.
        If a chunk is smaller than `size`, we will return less than asked, but we will always return data if there
        are chunks left
        :param size: amount to read, please note that it can return less than asked
        :return: bytes from the underlying stream
        """
        if size < 0:
            return self.readall()

        if not size:
            return b""

        if self._end_chunk:
            # if it's the end of a chunk we need to strip the newline at the end of the chunk
            # before jumping to the new one
            self._strip_chunk_new_lines()
            self._new_chunk = True
            self._end_chunk = False

        if self._new_chunk:
            # If the _new_chunk flag is set, we have to jump to the next chunk, if there's one
            self._get_next_chunk_length()
            self._new_chunk = False

        if self._chunk_size == 0 and self._decoded_length <= 0:
            # If the next chunk is 0, and we decoded everything, try to get the trailing headers
            self._get_trailing_headers()
            if self.s3_object:
                self._set_checksum_value()
            return b""

        # take the minimum account between the requested size, and the left chunk size
        # (to not over read from the chunk)
        amount = min(self._chunk_size, size)
        data = self._stream.read(amount)

        if data == b"":
            raise EOFError("Encoded file ended before the end-of-stream marker was reached")

        read = len(data)
        self._chunk_size -= read
        if self._chunk_size <= 0:
            self._end_chunk = True

        self._decoded_length -= read

        return data

    def _strip_chunk_new_lines(self):
        self._stream.read(2)

    def _get_next_chunk_length(self):
        line = self._stream.readline()
        chunk_length = int(line.split(b";")[0], 16)
        self._chunk_size = chunk_length

    def _get_trailing_headers(self):
        """
        Once the stream content is read, we try to parse the trailing headers.
        """
        # try to get all trailing headers until the end of the stream
        while line := self._stream.readline():
            if trailing_header := line.strip():
                header_key, header_value = trailing_header.decode("utf-8").split(":", maxsplit=1)
                self._trailing_headers[header_key.lower()] = header_value.strip()
        self._trailing_set = True

    def _set_checksum_value(self):
        """
        If an S3 object was passed, we check the presence of the `checksum_algorithm` field, so that we can properly
        get the right checksum header value, and set it directly to the object. It allows us to transparently access
        the provided checksum value by the client in the S3 logic.
        """
        if checksum_algorithm := getattr(self.s3_object, "checksum_algorithm", None):
            if checksum_value := self._trailing_headers.get(
                f"x-amz-checksum-{checksum_algorithm.lower()}"
            ):
                self.s3_object.checksum_value = checksum_value

1	import io	1✔
2	from typing import IO, Any	1✔
3
4
5	class AwsChunkedDecoder(io.RawIOBase):	1✔
6	"""
7	This helper class takes a IO[bytes] stream, and decodes it on the fly, so that S3 can directly access the stream
8	without worrying about implementation details of `aws-chunked`.
9	It needs to expose the trailing headers, which will be available once the stream is fully read.
10	You can also directly pass the S3 Object, so the stream would set the checksum value once it's done.
11	See `aws-chunked` format here: https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html
12	"""
13
14	def readable(self):	1✔
15	return True	×
16
17	def __init__(	1✔
18	self, stream: IO[bytes], decoded_content_length: int, s3_object: Any \| None = None
19	):
20	self._stream = stream	1✔
21
22	self._decoded_length = decoded_content_length # Length of the encoded object	1✔
23	self._new_chunk = True	1✔
24	self._end_chunk = False	1✔
25	self._trailing_set = False	1✔
26	self._chunk_size = 0	1✔
27	self._trailing_headers = {}	1✔
28	self.s3_object = s3_object	1✔
29
30	@property	1✔
31	def trailing_headers(self):	1✔
32	if not self._trailing_set:	1✔
33	raise AttributeError(	1✔
34	"The stream has not been fully read yet, the trailing headers are not available."
35	)
36	return self._trailing_headers	1✔
37
38	def seekable(self):	1✔
39	return self._stream.seekable()	×
40
41	def readinto(self, b):	1✔
42	with memoryview(b) as view, view.cast("B") as byte_view:	×
43	data = self.read(len(byte_view))	×
44	byte_view[: len(data)] = data	×
45	return len(data)	×
46
47	def read(self, size=-1):	1✔
48	"""
49	Read from the underlying stream, and return at most `size` decoded bytes.
50	If a chunk is smaller than `size`, we will return less than asked, but we will always return data if there
51	are chunks left
52	:param size: amount to read, please note that it can return less than asked
53	:return: bytes from the underlying stream
54	"""
55	if size < 0:	1✔
56	return self.readall()	1✔
57
58	if not size:	1✔
59	return b""	×
60
61	if self._end_chunk:	1✔
62	# if it's the end of a chunk we need to strip the newline at the end of the chunk
63	# before jumping to the new one
64	self._strip_chunk_new_lines()	1✔
65	self._new_chunk = True	1✔
66	self._end_chunk = False	1✔
67
68	if self._new_chunk:	1✔
69	# If the _new_chunk flag is set, we have to jump to the next chunk, if there's one
70	self._get_next_chunk_length()	1✔
71	self._new_chunk = False	1✔
72
73	if self._chunk_size == 0 and self._decoded_length <= 0:	1✔
74	# If the next chunk is 0, and we decoded everything, try to get the trailing headers
75	self._get_trailing_headers()	1✔
76	if self.s3_object:	1✔
UNCOV 77	self._set_checksum_value()	×
78	return b""	1✔
79
80	# take the minimum account between the requested size, and the left chunk size
81	# (to not over read from the chunk)
82	amount = min(self._chunk_size, size)	1✔
83	data = self._stream.read(amount)	1✔
84
85	if data == b"":	1✔
86	raise EOFError("Encoded file ended before the end-of-stream marker was reached")	×
87
88	read = len(data)	1✔
89	self._chunk_size -= read	1✔
90	if self._chunk_size <= 0:	1✔
91	self._end_chunk = True	1✔
92
93	self._decoded_length -= read	1✔
94
95	return data	1✔
96
97	def _strip_chunk_new_lines(self):	1✔
98	self._stream.read(2)	1✔
99
100	def _get_next_chunk_length(self):	1✔
101	line = self._stream.readline()	1✔
102	chunk_length = int(line.split(b";")[0], 16)	1✔
103	self._chunk_size = chunk_length	1✔
104
105	def _get_trailing_headers(self):	1✔
106	"""
107	Once the stream content is read, we try to parse the trailing headers.
108	"""
109	# try to get all trailing headers until the end of the stream
110	while line := self._stream.readline():	1✔
111	if trailing_header := line.strip():	1✔
112	header_key, header_value = trailing_header.decode("utf-8").split(":", maxsplit=1)	1✔
113	self._trailing_headers[header_key.lower()] = header_value.strip()	1✔
114	self._trailing_set = True	1✔
115
116	def _set_checksum_value(self):	1✔
117	"""
118	If an S3 object was passed, we check the presence of the `checksum_algorithm` field, so that we can properly
119	get the right checksum header value, and set it directly to the object. It allows us to transparently access
120	the provided checksum value by the client in the S3 logic.
121	"""
UNCOV 122	if checksum_algorithm := getattr(self.s3_object, "checksum_algorithm", None):	×
UNCOV 123	if checksum_value := self._trailing_headers.get(	×
124	f"x-amz-checksum-{checksum_algorithm.lower()}"
125	):
UNCOV 126	self.s3_object.checksum_value = checksum_value	×

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