7956692684

pending completion

Build # 7956692684

Build Type

push

github

Committed by

datakurre

Commit Message

ci: Install sphinxcontrib-httpexample in readthedocs

Run Details

418 of 445 relevant lines covered (93.93%)

0.94 hits per line

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

97.56

/src/sphinxcontrib/httpexample/parsers.py

# -*- coding: utf-8 -*-
from io import BytesIO
from sphinxcontrib.httpexample.utils import add_url_params
from sphinxcontrib.httpexample.utils import capitalize_keys
from sphinxcontrib.httpexample.utils import is_json
from sphinxcontrib.httpexample.utils import ordered

import base64
import json
import re


try:
    from http.server import BaseHTTPRequestHandler
except ImportError:
    from BaseHTTPServer import BaseHTTPRequestHandler


AVAILABLE_FIELDS = [
    'query'
]


class HTTPRequest(BaseHTTPRequestHandler):
    # http://stackoverflow.com/a/5955949

    scheme = 'http'

    # noinspection PyMissingConstructor
    def __init__(self, request_bytes, scheme):
        assert isinstance(request_bytes, bytes)

        self.scheme = scheme
        self.rfile = BytesIO(request_bytes)
        self.raw_requestline = self.rfile.readline()
        self.error_code = self.error_message = None
        self.parse_request()

        if self.error_message:
            raise Exception(self.error_message)

        # Replace headers with simple dict to coup differences in Py2 and Py3
        self.headers = capitalize_keys(dict(getattr(self, 'headers', {})))

    def send_error(self, code, message=None, explain=None):
        self.error_code = code
        self.error_message = message

    def extract_fields(self, field=None, available_fields=None):
        if available_fields is None:
            available_fields = AVAILABLE_FIELDS

        if (field is not None) and field not in available_fields:
            msg = "Unexpected field '{}'. Expected one of {}."
            msg = msg.format(field, ', '.join(available_fields))
            raise ValueError(msg)

        if field is None:
            field = '|'.join(available_fields)
        is_field = r':({}) (.+): (.+)'.format(field)

        fields = []
        remaining_request = []
        cursor = self.rfile.tell()
        for i, line in enumerate(self.rfile.readlines()):
            line = line.decode('utf-8')
            try:
                field, key, val = re.match(is_field, line).groups()
            except AttributeError:
                remaining_request.append(line)
                continue
            fields.append((field.strip(), key.strip(), val.strip()))

        remaining_request = BytesIO(
            '\n'.join(remaining_request).encode('utf-8').strip())
        remaining_request.seek(0)
        self.rfile.seek(cursor)

        return (fields, remaining_request)

    def auth(self):
        try:
            method, token = self.headers.get('Authorization').split()
        except (AttributeError, KeyError, ValueError):
            return None, None
        if not isinstance(token, bytes):
            token = token.encode('utf-8')
        if method == 'Basic':
            return method, base64.b64decode(token).decode('utf-8')
        else:
            return method, token

    def url(self):
        base_url = '{}://{}{}'.format(
            self.scheme,
            self.headers.get('Host', 'nohost'),
            self.path
        )

        params, _ = self.extract_fields('query')
        params = [(p[1], p[2]) for p in params]

        if params:
            new_url = add_url_params(base_url, params)
        else:
            new_url = base_url

        return new_url

    def data(self):
        _, payload_bytes = self.extract_fields(None)
        payload_bytes = payload_bytes.read()
        if payload_bytes:
            if is_json(self.headers.get('Content-Type', '')):
                assert isinstance(payload_bytes, bytes)
                payload_str = payload_bytes.decode('utf-8')
                return ordered(json.loads(payload_str))
            else:
                return payload_bytes


def parse_request(request_bytes, scheme='http'):
    return HTTPRequest(request_bytes, scheme)

1	# -- coding: utf-8 --
2	from io import BytesIO	4✔
3	from sphinxcontrib.httpexample.utils import add_url_params	4✔
4	from sphinxcontrib.httpexample.utils import capitalize_keys	4✔
5	from sphinxcontrib.httpexample.utils import is_json	4✔
6	from sphinxcontrib.httpexample.utils import ordered	4✔
7
8	import base64	4✔
9	import json	4✔
10	import re	4✔
11
12
13	try:	4✔
14	from http.server import BaseHTTPRequestHandler	4✔
15	except ImportError:	×
16	from BaseHTTPServer import BaseHTTPRequestHandler	×
17
18
19	AVAILABLE_FIELDS = [	4✔
20	'query'
21	]
22
23
24	class HTTPRequest(BaseHTTPRequestHandler):	4✔
25	# http://stackoverflow.com/a/5955949
26
27	scheme = 'http'	4✔
28
29	# noinspection PyMissingConstructor
30	def __init__(self, request_bytes, scheme):	4✔
31	assert isinstance(request_bytes, bytes)	4✔
32
33	self.scheme = scheme	4✔
34	self.rfile = BytesIO(request_bytes)	4✔
35	self.raw_requestline = self.rfile.readline()	4✔
36	self.error_code = self.error_message = None	4✔
37	self.parse_request()	4✔
38
39	if self.error_message:	4✔
40	raise Exception(self.error_message)	4✔
41
42	# Replace headers with simple dict to coup differences in Py2 and Py3
43	self.headers = capitalize_keys(dict(getattr(self, 'headers', {})))	4✔
44
45	def send_error(self, code, message=None, explain=None):	4✔
46	self.error_code = code	4✔
47	self.error_message = message	4✔
48
49	def extract_fields(self, field=None, available_fields=None):	4✔
50	if available_fields is None:	4✔
51	available_fields = AVAILABLE_FIELDS	4✔
52
53	if (field is not None) and field not in available_fields:	4✔
54	msg = "Unexpected field '{}'. Expected one of {}."	4✔
55	msg = msg.format(field, ', '.join(available_fields))	4✔
56	raise ValueError(msg)	4✔
57
58	if field is None:	4✔
59	field = '\|'.join(available_fields)	4✔
60	is_field = r':({}) (.+): (.+)'.format(field)	4✔
61
62	fields = []	4✔
63	remaining_request = []	4✔
64	cursor = self.rfile.tell()	4✔
65	for i, line in enumerate(self.rfile.readlines()):	4✔
66	line = line.decode('utf-8')	4✔
67	try:	4✔
68	field, key, val = re.match(is_field, line).groups()	4✔
69	except AttributeError:	4✔
70	remaining_request.append(line)	4✔
71	continue	4✔
72	fields.append((field.strip(), key.strip(), val.strip()))	4✔
73
74	remaining_request = BytesIO(	4✔
75	'\n'.join(remaining_request).encode('utf-8').strip())
76	remaining_request.seek(0)	4✔
77	self.rfile.seek(cursor)	4✔
78
79	return (fields, remaining_request)	4✔
80
81	def auth(self):	4✔
82	try:	4✔
83	method, token = self.headers.get('Authorization').split()	4✔
84	except (AttributeError, KeyError, ValueError):	4✔
85	return None, None	4✔
86	if not isinstance(token, bytes):	4✔
87	token = token.encode('utf-8')	4✔
88	if method == 'Basic':	4✔
89	return method, base64.b64decode(token).decode('utf-8')	4✔
90	else:
91	return method, token	4✔
92
93	def url(self):	4✔
94	base_url = '{}://{}{}'.format(	4✔
95	self.scheme,
96	self.headers.get('Host', 'nohost'),
97	self.path
98	)
99
100	params, _ = self.extract_fields('query')	4✔
101	params = [(p[1], p[2]) for p in params]	4✔
102
103	if params:	4✔
104	new_url = add_url_params(base_url, params)	4✔
105	else:
106	new_url = base_url	4✔
107
108	return new_url	4✔
109
110	def data(self):	4✔
111	_, payload_bytes = self.extract_fields(None)	4✔
112	payload_bytes = payload_bytes.read()	4✔
113	if payload_bytes:	4✔
114	if is_json(self.headers.get('Content-Type', '')):	4✔
115	assert isinstance(payload_bytes, bytes)	4✔
116	payload_str = payload_bytes.decode('utf-8')	4✔
117	return ordered(json.loads(payload_str))	4✔
118	else:
119	return payload_bytes	4✔
120
121
122	def parse_request(request_bytes, scheme='http'):	4✔
123	return HTTPRequest(request_bytes, scheme)	4✔

collective / sphinxcontrib-httpexample / 7956692684

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