a43cb023e0d04fe59852315c1ffb8a25dd1ff96e

Committed 17 Aug 2024 05:41PM UTC coverage: 96.629% (+2.7%) from 93.933%

Build # a43cb023e0d04fe59852315c1ffb8a25dd1ff96e

Build Type

push

github

Committed by

datakurre

Commit Message

Back to development: 1.3

Run Details

430 of 445 relevant lines covered (96.63%)

1.93 hits per line

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

97.27

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

collective / sphinxcontrib-httpexample / a43cb023e0d04fe59852315c1ffb8a25dd1ff96e

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