10442703330

Committed 18 Aug 2024 06:06PM UTC coverage: 94.03% (-1.9%) from 95.94%

Build # 10442703330

Build Type

push

github

Committed by

datakurre

Commit Message

Back to development: 1.4

Run Details

441 of 469 relevant lines covered (94.03%)

14.07 hits per line

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

90.16

/src/sphinxcontrib/httpexample/builders.py

# -*- coding: utf-8 -*-
from sphinxcontrib.httpexample.utils import is_json
from sphinxcontrib.httpexample.utils import maybe_str

import ast
import json
import re
import string


try:
    from urllib.parse import parse_qs
except ImportError:
    from urlparse import parse_qs

try:
    from ast import unparse
except ImportError:
    from six.moves import StringIO

    import astunparse

    # Fix: https://github.com/simonpercivall/astunparse/issues/43
    # See: https://github.com/juanlao7/codeclose/commit/0f145f53a3253f9c593c537e1a25c9ef445f30d1  # noqa: E501
    class FixUnparser(astunparse.Unparser):
        def _Constant(self, t):  # noqa:  N802
            if not hasattr(t, 'kind'):
                setattr(t, 'kind', None)
            super()._Constant(t)

    def unparse(tree):
        v = StringIO()
        FixUnparser(tree, file=v)
        return v.getvalue()


try:
    from shlex import quote as shlex_quote
except ImportError:
    from pipes import quote as shlex_quote


_find_unsafe = re.compile(
    r'[^\w@%+=:,./-' + string.ascii_letters + string.digits + ']',
).search


def shlex_double_quote(s):
    """Return a shell-escaped version of the string *s*."""
    if not s:
        return '""'
    if _find_unsafe(s) is None:
        return s

    # use double quotes, and put double quotes into single quotes
    # the string $"b is then quoted as "$"'"'"b"
    return re.sub(r'^""|""$', '', ('"' + s.replace('"', "\"'\"'\"") + '"'))


EXCLUDE_HEADERS = [
    'Authorization',
    'Host',
]
EXCLUDE_HEADERS_HTTP = EXCLUDE_HEADERS + []
EXCLUDE_HEADERS_REQUESTS = EXCLUDE_HEADERS + []


def build_curl_command(request):
    parts = ['curl', '-i']

    # Method
    parts.append('-X {}'.format(request.command))

    # URL
    parts.append(shlex_quote(request.url()))

    # Authorization (prepare)
    method, token = request.auth()

    # Headers
    for header in sorted(request.headers):
        if header in EXCLUDE_HEADERS:
            continue
        header_line = shlex_double_quote(
            '{}: {}'.format(header, request.headers[header]),
        )
        parts.append('-H {}'.format(header_line))

    if method != 'Basic' and 'Authorization' in request.headers:
        header = 'Authorization'
        header_line = shlex_double_quote(
            '{}: {}'.format(header, request.headers[header]),
        )
        parts.append('-H {}'.format(header_line))

    # JSON
    data = maybe_str(request.data())
    if data:
        if is_json(request.headers.get('Content-Type', '')):
            data = json.dumps(data)
        parts.append("--data-raw '{}'".format(data))

    # Authorization
    if method == 'Basic':
        parts.append('--user {}'.format(token))

    return ' '.join(parts)


def build_wget_command(request):
    parts = ['wget', '-S', '-O-']

    # Method
    if request.command not in ['GET', 'POST']:
        parts.append('--method={}'.format(request.command))

    # URL
    parts.append(shlex_quote(request.url()))

    # Authorization (prepare)
    method, token = request.auth()

    # Headers
    for header in sorted(request.headers):
        if header in EXCLUDE_HEADERS:
            continue
        header_line = shlex_double_quote(
            '{}: {}'.format(header, request.headers[header]),
        )
        parts.append('--header={}'.format(header_line))

    if method != 'Basic' and 'Authorization' in request.headers:
        header = 'Authorization'
        header_line = shlex_double_quote(
            '{}: {}'.format(header, request.headers[header])
        )
        parts.append('--header={}'.format(header_line))

    # JSON or raw data
    data = maybe_str(request.data())
    if data:
        if is_json(request.headers.get('Content-Type', '')):
            data = json.dumps(data)
        if request.command == 'POST':
            parts.append("--post-data='{}'".format(data))
        elif request.command != 'POST':
            parts.append("--body-data='{}'".format(data))

    # Authorization
    if method == 'Basic':
        user, password = token.split(':')
        parts.append('--auth-no-challenge')
        parts.append('--user={}'.format(user))
        parts.append('--password={}'.format(password))

    return ' '.join(parts)


def build_httpie_command(request):
    parts = ['http']
    redir_input = ''

    # Method
    if request.command != 'GET':
        parts.append(request.command)

    # URL
    parts.append(shlex_quote(request.url()))

    # Authorization (prepare)
    method, token = request.auth()

    # Headers
    for header in sorted(request.headers):
        if header in EXCLUDE_HEADERS_HTTP:
            continue
        parts.append(
            '{}:{}'.format(
                header,
                shlex_double_quote(request.headers[header]),
            )
        )

    if method != 'Basic' and 'Authorization' in request.headers:
        header = 'Authorization'
        parts.append(
            '{}:{}'.format(
                header,
                shlex_double_quote(request.headers[header]),
            )
        )

    # JSON or raw data
    data = maybe_str(request.data())
    if data:

        if is_json(request.headers.get('Content-Type', '')):
            # We need to explicitly set the separators to get consistent
            # whitespace handling across Python 2 and 3. See
            # https://bugs.python.org/issue16333 for details.
            redir_input = shlex_quote(
                json.dumps(data, indent=2, sort_keys=True, separators=(',', ': '))
            )
        else:
            redir_input = shlex_quote(data)

    # Authorization
    if method == 'Basic':
        parts.append('-a {}'.format(token))

    cmd = ' '.join(parts)

    if not redir_input:
        return cmd

    else:
        return 'echo {} | {}'.format(redir_input, cmd)


def build_plone_javascript_command(request):
    javascript_code = 'createAliasesMutation'
    redir_input2 = ''

    # Request body
    data = maybe_str(request.data())
    if data:
        if is_json(request.headers.get('Content-Type', '')):
            redir_input2 = json.dumps(
                data,
                indent=2,
                sort_keys=True,
                separators=(',', ': '),
            ).encode('utf-8')
        else:
            redir_input2 = data

    # Output string
    output_string = "{}\n|\nconst aliasesData = '{}';".format(
        maybe_str(javascript_code),
        maybe_str(redir_input2),
    )

    return output_string


def flatten_parsed_qs(data):
    """Flatten single value lists in parse_qs results."""
    for key, value in data.items():
        if isinstance(value, list) and len(value) == 1:
            data[key] = value[0]
    return data


def build_requests_command(request):
    # Method
    tree = ast.parse('requests.{}()'.format(request.command.lower()))
    call = tree.body[0].value
    call.keywords = []

    # URL
    call.args.append(ast.Str(request.url()))

    # Authorization (prepare)
    method, token = request.auth()

    # Headers
    header_keys = []
    header_values = []
    for header in sorted(request.headers):
        if header in EXCLUDE_HEADERS_REQUESTS:
            continue
        header_keys.append(ast.Str(header))
        header_values.append(ast.Str(request.headers[header]))
    if method != 'Basic' and 'Authorization' in request.headers:
        header_keys.append(ast.Str('Authorization'))
        header_values.append(ast.Str(request.headers['Authorization']))
    if header_keys and header_values:
        call.keywords.append(
            ast.keyword('headers', ast.Dict(header_keys, header_values))
        )

    # JSON or raw data
    data = maybe_str(request.data())

    # Form data
    content_type = request.headers.get('Content-Type')
    if content_type == 'application/x-www-form-urlencoded':
        if not isinstance(data, dict):
            data = flatten_parsed_qs(parse_qs(data))

    def astify_json_obj(obj):
        obj = maybe_str(obj)
        if isinstance(obj, str):
            return ast.Str(obj)
        elif isinstance(obj, bool):
            return ast.Name(str(obj), ast.Load())
        elif isinstance(obj, int):
            return ast.Name(str(obj), ast.Load())
        elif isinstance(obj, float):
            return ast.Name(str(obj), ast.Load())
        elif isinstance(obj, list):
            json_values = []
            for v in obj:
                json_values.append(astify_json_obj(v))
            return ast.List(json_values, ast.Load())
        elif isinstance(obj, dict):
            json_values = []
            json_keys = []
            for k, v in obj.items():
                json_keys.append(ast.Str(maybe_str(k)))
                json_values.append(astify_json_obj(v))
            return ast.Dict(json_keys, json_values)
        else:
            raise Exception('Cannot astify {0:s}'.format(str(obj)))

    if data:
        if is_json(request.headers.get('Content-Type', '')):
            call.keywords.append(ast.keyword('json', astify_json_obj(data)))
        else:
            call.keywords.append(ast.keyword('data', ast.Str(data)))

    # Authorization
    if method == 'Basic':
        token = maybe_str(token)
        call.keywords.append(
            ast.keyword('auth', ast.Tuple(tuple(map(ast.Str, token.split(':'))), None))
        )

    return unparse(tree).strip()

1	# -- coding: utf-8 --
2	from sphinxcontrib.httpexample.utils import is_json	15✔
3	from sphinxcontrib.httpexample.utils import maybe_str	15✔
4
5	import ast	15✔
6	import json	15✔
7	import re	15✔
8	import string	15✔
9
10
11	try:	15✔
12	from urllib.parse import parse_qs	15✔
13	except ImportError:	×
14	from urlparse import parse_qs	×
15
16	try:	15✔
17	from ast import unparse	15✔
18	except ImportError:	×
19	from six.moves import StringIO	×
20
21	import astunparse	×
22
23	# Fix: https://github.com/simonpercivall/astunparse/issues/43
24	# See: https://github.com/juanlao7/codeclose/commit/0f145f53a3253f9c593c537e1a25c9ef445f30d1 # noqa: E501
25	class FixUnparser(astunparse.Unparser):	×
26	def _Constant(self, t): # noqa: N802	×
27	if not hasattr(t, 'kind'):	×
28	setattr(t, 'kind', None)	×
29	super()._Constant(t)	×
30
31	def unparse(tree):	×
32	v = StringIO()	×
33	FixUnparser(tree, file=v)	×
34	return v.getvalue()	×
35
36
37	try:	15✔
38	from shlex import quote as shlex_quote	15✔
39	except ImportError:	×
40	from pipes import quote as shlex_quote	×
41
42
43	_find_unsafe = re.compile(	15✔
44	r'[^\w@%+=:,./-' + string.ascii_letters + string.digits + ']',
45	).search
46
47
48	def shlex_double_quote(s):	15✔
49	"""Return a shell-escaped version of the string s."""
50	if not s:	15✔
51	return '""'	×
52	if _find_unsafe(s) is None:	15✔
53	return s	15✔
54
55	# use double quotes, and put double quotes into single quotes
56	# the string $"b is then quoted as "$"'"'"b"
57	return re.sub(r'^""\|""$', '', ('"' + s.replace('"', "\"'\"'\"") + '"'))	15✔
58
59
60	EXCLUDE_HEADERS = [	15✔
61	'Authorization',
62	'Host',
63	]
64	EXCLUDE_HEADERS_HTTP = EXCLUDE_HEADERS + []	15✔
65	EXCLUDE_HEADERS_REQUESTS = EXCLUDE_HEADERS + []	15✔
66
67
68	def build_curl_command(request):	15✔
69	parts = ['curl', '-i']	15✔
70
71	# Method
72	parts.append('-X {}'.format(request.command))	15✔
73
74	# URL
75	parts.append(shlex_quote(request.url()))	15✔
76
77	# Authorization (prepare)
78	method, token = request.auth()	15✔
79
80	# Headers
81	for header in sorted(request.headers):	15✔
82	if header in EXCLUDE_HEADERS:	15✔
83	continue	15✔
84	header_line = shlex_double_quote(	15✔
85	'{}: {}'.format(header, request.headers[header]),
86	)
87	parts.append('-H {}'.format(header_line))	15✔
88
89	if method != 'Basic' and 'Authorization' in request.headers:	15✔
90	header = 'Authorization'	15✔
91	header_line = shlex_double_quote(	15✔
92	'{}: {}'.format(header, request.headers[header]),
93	)
94	parts.append('-H {}'.format(header_line))	15✔
95
96	# JSON
97	data = maybe_str(request.data())	15✔
98	if data:	15✔
99	if is_json(request.headers.get('Content-Type', '')):	15✔
100	data = json.dumps(data)	15✔
101	parts.append("--data-raw '{}'".format(data))	15✔
102
103	# Authorization
104	if method == 'Basic':	15✔
105	parts.append('--user {}'.format(token))	15✔
106
107	return ' '.join(parts)	15✔
108
109
110	def build_wget_command(request):	15✔
111	parts = ['wget', '-S', '-O-']	15✔
112
113	# Method
114	if request.command not in ['GET', 'POST']:	15✔
115	parts.append('--method={}'.format(request.command))	15✔
116
117	# URL
118	parts.append(shlex_quote(request.url()))	15✔
119
120	# Authorization (prepare)
121	method, token = request.auth()	15✔
122
123	# Headers
124	for header in sorted(request.headers):	15✔
125	if header in EXCLUDE_HEADERS:	15✔
126	continue	15✔
127	header_line = shlex_double_quote(	15✔
128	'{}: {}'.format(header, request.headers[header]),
129	)
130	parts.append('--header={}'.format(header_line))	15✔
131
132	if method != 'Basic' and 'Authorization' in request.headers:	15✔
133	header = 'Authorization'	15✔
134	header_line = shlex_double_quote(	15✔
135	'{}: {}'.format(header, request.headers[header])
136	)
137	parts.append('--header={}'.format(header_line))	15✔
138
139	# JSON or raw data
140	data = maybe_str(request.data())	15✔
141	if data:	15✔
142	if is_json(request.headers.get('Content-Type', '')):	15✔
143	data = json.dumps(data)	15✔
144	if request.command == 'POST':	15✔
145	parts.append("--post-data='{}'".format(data))	15✔
146	elif request.command != 'POST':	15✔
147	parts.append("--body-data='{}'".format(data))	15✔
148
149	# Authorization
150	if method == 'Basic':	15✔
151	user, password = token.split(':')	15✔
152	parts.append('--auth-no-challenge')	15✔
153	parts.append('--user={}'.format(user))	15✔
154	parts.append('--password={}'.format(password))	15✔
155
156	return ' '.join(parts)	15✔
157
158
159	def build_httpie_command(request):	15✔
160	parts = ['http']	15✔
161	redir_input = ''	15✔
162
163	# Method
164	if request.command != 'GET':	15✔
165	parts.append(request.command)	15✔
166
167	# URL
168	parts.append(shlex_quote(request.url()))	15✔
169
170	# Authorization (prepare)
171	method, token = request.auth()	15✔
172
173	# Headers
174	for header in sorted(request.headers):	15✔
175	if header in EXCLUDE_HEADERS_HTTP:	15✔
176	continue	15✔
177	parts.append(	15✔
178	'{}:{}'.format(
179	header,
180	shlex_double_quote(request.headers[header]),
181	)
182	)
183
184	if method != 'Basic' and 'Authorization' in request.headers:	15✔
185	header = 'Authorization'	15✔
186	parts.append(	15✔
187	'{}:{}'.format(
188	header,
189	shlex_double_quote(request.headers[header]),
190	)
191	)
192
193	# JSON or raw data
194	data = maybe_str(request.data())	15✔
195	if data:	15✔
196
197	if is_json(request.headers.get('Content-Type', '')):	15✔
198	# We need to explicitly set the separators to get consistent
199	# whitespace handling across Python 2 and 3. See
200	# https://bugs.python.org/issue16333 for details.
201	redir_input = shlex_quote(	15✔
202	json.dumps(data, indent=2, sort_keys=True, separators=(',', ': '))
203	)
204	else:
205	redir_input = shlex_quote(data)	15✔
206
207	# Authorization
208	if method == 'Basic':	15✔
209	parts.append('-a {}'.format(token))	15✔
210
211	cmd = ' '.join(parts)	15✔
212
213	if not redir_input:	15✔
214	return cmd	15✔
215
216	else:
217	return 'echo {} \| {}'.format(redir_input, cmd)	15✔
218
219
220	def build_plone_javascript_command(request):	15✔
221	javascript_code = 'createAliasesMutation'	15✔
222	redir_input2 = ''	15✔
223
224	# Request body
225	data = maybe_str(request.data())	15✔
226	if data:	15✔
227	if is_json(request.headers.get('Content-Type', '')):	15✔
228	redir_input2 = json.dumps(	15✔
229	data,
230	indent=2,
231	sort_keys=True,
232	separators=(',', ': '),
233	).encode('utf-8')
234	else:
235	redir_input2 = data	15✔
236
237	# Output string
238	output_string = "{}\n\|\nconst aliasesData = '{}';".format(	15✔
239	maybe_str(javascript_code),
240	maybe_str(redir_input2),
241	)
242
243	return output_string	15✔
244
245
246	def flatten_parsed_qs(data):	15✔
247	"""Flatten single value lists in parse_qs results."""
248	for key, value in data.items():	15✔
249	if isinstance(value, list) and len(value) == 1:	15✔
250	data[key] = value[0]	15✔
251	return data	15✔
252
253
254	def build_requests_command(request):	15✔
255	# Method
256	tree = ast.parse('requests.{}()'.format(request.command.lower()))	15✔
257	call = tree.body[0].value	15✔
258	call.keywords = []	15✔
259
260	# URL
261	call.args.append(ast.Str(request.url()))	15✔
262
263	# Authorization (prepare)
264	method, token = request.auth()	15✔
265
266	# Headers
267	header_keys = []	15✔
268	header_values = []	15✔
269	for header in sorted(request.headers):	15✔
270	if header in EXCLUDE_HEADERS_REQUESTS:	15✔
271	continue	15✔
272	header_keys.append(ast.Str(header))	15✔
273	header_values.append(ast.Str(request.headers[header]))	15✔
274	if method != 'Basic' and 'Authorization' in request.headers:	15✔
275	header_keys.append(ast.Str('Authorization'))	15✔
276	header_values.append(ast.Str(request.headers['Authorization']))	15✔
277	if header_keys and header_values:	15✔
278	call.keywords.append(	15✔
279	ast.keyword('headers', ast.Dict(header_keys, header_values))
280	)
281
282	# JSON or raw data
283	data = maybe_str(request.data())	15✔
284
285	# Form data
286	content_type = request.headers.get('Content-Type')	15✔
287	if content_type == 'application/x-www-form-urlencoded':	15✔
288	if not isinstance(data, dict):	15✔
289	data = flatten_parsed_qs(parse_qs(data))	15✔
290
291	def astify_json_obj(obj):	15✔
292	obj = maybe_str(obj)	15✔
293	if isinstance(obj, str):	15✔
294	return ast.Str(obj)	15✔
295	elif isinstance(obj, bool):	15✔
296	return ast.Name(str(obj), ast.Load())	15✔
297	elif isinstance(obj, int):	15✔
298	return ast.Name(str(obj), ast.Load())	15✔
299	elif isinstance(obj, float):	15✔
300	return ast.Name(str(obj), ast.Load())	15✔
301	elif isinstance(obj, list):	15✔
302	json_values = []	15✔
303	for v in obj:	15✔
304	json_values.append(astify_json_obj(v))	15✔
305	return ast.List(json_values, ast.Load())	15✔
306	elif isinstance(obj, dict):	15✔
307	json_values = []	15✔
308	json_keys = []	15✔
309	for k, v in obj.items():	15✔
310	json_keys.append(ast.Str(maybe_str(k)))	15✔
311	json_values.append(astify_json_obj(v))	15✔
312	return ast.Dict(json_keys, json_values)	15✔
313	else:
314	raise Exception('Cannot astify {0:s}'.format(str(obj)))	×
315
316	if data:	15✔
317	if is_json(request.headers.get('Content-Type', '')):	15✔
318	call.keywords.append(ast.keyword('json', astify_json_obj(data)))	15✔
319	else:
320	call.keywords.append(ast.keyword('data', ast.Str(data)))	15✔
321
322	# Authorization
323	if method == 'Basic':	15✔
324	token = maybe_str(token)	15✔
325	call.keywords.append(	15✔
326	ast.keyword('auth', ast.Tuple(tuple(map(ast.Str, token.split(':'))), None))
327	)
328
329	return unparse(tree).strip()	15✔

collective / sphinxcontrib-httpexample / 10442703330

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