11038896747

Committed 25 Sep 2024 06:24PM UTC coverage: 96.228% (-0.09%) from 96.317%

Build # 11038896747

Build Type

Pull #1397

github

Committed by

web-flow

Commit Message

Merge 8bf2cf94a into 3587026b8

Pull Request Pull Request #1397: Next major infra

Run Details

1331 of 1967 branches covered (67.67%)

Branch coverage included in aggregate %.

509 of 584 new or added lines in 40 files covered. (87.16%)

2 existing lines in 2 files now uncovered.

67645 of 69713 relevant lines covered (97.03%)

0.97 hits per line

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

93.42

/stripe/_multipart_data_generator.py

import random
import io

from stripe._encode import _api_encode


class MultipartDataGenerator(object):
    data: io.BytesIO
    line_break: str
    boundary: int
    chunk_size: int

    def __init__(self, chunk_size: int = 1028):
        self.data = io.BytesIO()
        self.line_break = "\r\n"
        self.boundary = self._initialize_boundary()
        self.chunk_size = chunk_size

    def add_params(self, params):
        # Flatten parameters first

        params = dict(_api_encode(params, "V1"))

        for key, value in params.items():
            if value is None:
                continue

            self._write(self.param_header())
            self._write(self.line_break)
            if hasattr(value, "read"):
                filename = "blob"
                if hasattr(value, "name"):
                    # Convert the filename to string, just in case it's not
                    # already one. E.g. `tempfile.TemporaryFile` has a `name`
                    # attribute but it's an `int`.
                    filename = str(value.name)

                self._write('Content-Disposition: form-data; name="')
                self._write(key)
                self._write('"; filename="')
                self._write(filename)
                self._write('"')
                self._write(self.line_break)
                self._write("Content-Type: application/octet-stream")
                self._write(self.line_break)
                self._write(self.line_break)

                self._write_file(value)
            else:
                self._write('Content-Disposition: form-data; name="')
                self._write(key)
                self._write('"')
                self._write(self.line_break)
                self._write(self.line_break)
                self._write(str(value))

            self._write(self.line_break)

    def param_header(self):
        return "--%s" % self.boundary

    def get_post_data(self):
        self._write("--%s--" % (self.boundary,))
        self._write(self.line_break)
        return self.data.getvalue()

    def _write(self, value):
        if isinstance(value, bytes):
            array = bytearray(value)
        elif isinstance(value, str):
            array = bytearray(value, encoding="utf-8")
        else:
            raise TypeError(
                "unexpected type: {value_type}".format(value_type=type(value))
            )

        self.data.write(array)

    def _write_file(self, f):
        while True:
            file_contents = f.read(self.chunk_size)
            if not file_contents:
                break
            self._write(file_contents)

    def _initialize_boundary(self):
        return random.randint(0, 2**63)

1	import random	1✔
2	import io	1✔
3
4	from stripe._encode import _api_encode	1✔
5
6
7	class MultipartDataGenerator(object):	1✔
8	data: io.BytesIO	1✔
9	line_break: str	1✔
10	boundary: int	1✔
11	chunk_size: int	1✔
12
13	def __init__(self, chunk_size: int = 1028):	1✔
14	self.data = io.BytesIO()	1✔
15	self.line_break = "\r\n"	1✔
16	self.boundary = self._initialize_boundary()	1✔
17	self.chunk_size = chunk_size	1✔
18
19	def add_params(self, params):	1✔
20	# Flatten parameters first
21
22	params = dict(_api_encode(params, "V1"))	1✔
23
24	for key, value in params.items():	1✔
25	if value is None:	1!
26	continue	×
27
28	self._write(self.param_header())	1✔
29	self._write(self.line_break)	1✔
30	if hasattr(value, "read"):	1✔
31	filename = "blob"	1✔
32	if hasattr(value, "name"):	1✔
33	# Convert the filename to string, just in case it's not
34	# already one. E.g. `tempfile.TemporaryFile` has a `name`
35	# attribute but it's an `int`.
36	filename = str(value.name)	1✔
37
38	self._write('Content-Disposition: form-data; name="')	1✔
39	self._write(key)	1✔
40	self._write('"; filename="')	1✔
41	self._write(filename)	1✔
42	self._write('"')	1✔
43	self._write(self.line_break)	1✔
44	self._write("Content-Type: application/octet-stream")	1✔
45	self._write(self.line_break)	1✔
46	self._write(self.line_break)	1✔
47
48	self._write_file(value)	1✔
49	else:
50	self._write('Content-Disposition: form-data; name="')	1✔
51	self._write(key)	1✔
52	self._write('"')	1✔
53	self._write(self.line_break)	1✔
54	self._write(self.line_break)	1✔
55	self._write(str(value))	1✔
56
57	self._write(self.line_break)	1✔
58
59	def param_header(self):	1✔
60	return "--%s" % self.boundary	1✔
61
62	def get_post_data(self):	1✔
63	self._write("--%s--" % (self.boundary,))	1✔
64	self._write(self.line_break)	1✔
65	return self.data.getvalue()	1✔
66
67	def _write(self, value):	1✔
68	if isinstance(value, bytes):	1✔
69	array = bytearray(value)	1✔
70	elif isinstance(value, str):	1!
71	array = bytearray(value, encoding="utf-8")	1✔
72	else:
73	raise TypeError(	×
74	"unexpected type: {value_type}".format(value_type=type(value))
75	)
76
77	self.data.write(array)	1✔
78
79	def _write_file(self, f):	1✔
80	while True:	1✔
81	file_contents = f.read(self.chunk_size)	1✔
82	if not file_contents:	1✔
83	break	1✔
84	self._write(file_contents)	1✔
85
86	def _initialize_boundary(self):	1✔
UNCOV 87	return random.randint(0, 2**63)	×

stripe / stripe-python / 11038896747

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