• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

collective / sphinxcontrib-httpexample / ff8e68703114cb30bdfb941181f89b6fc41e7c08

18 Aug 2024 11:09AM UTC coverage: 95.94% (+2.0%) from 93.933%
ff8e68703114cb30bdfb941181f89b6fc41e7c08

Pull #96

github

datakurre
Add sphinx-design and sphinx-inline-tabs into test lock files
Pull Request #96: Merge use with other tab libraries

50 of 54 new or added lines in 2 files covered. (92.59%)

9 existing lines in 2 files now uncovered.

449 of 468 relevant lines covered (95.94%)

1.92 hits per line

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

98.78
/src/sphinxcontrib/httpexample/parsers.py
1
# -*- coding: utf-8 -*-
2
from io import BytesIO
2✔
3
from sphinxcontrib.httpexample.utils import add_url_params
2✔
4
from sphinxcontrib.httpexample.utils import capitalize_keys
2✔
5
from sphinxcontrib.httpexample.utils import is_json
2✔
6
from sphinxcontrib.httpexample.utils import ordered
2✔
7

8
import base64
2✔
9
import json
2✔
10
import re
2✔
11

12

13
try:
2✔
14
    from http.server import BaseHTTPRequestHandler
2✔
15
except ImportError:
2✔
16
    from BaseHTTPServer import BaseHTTPRequestHandler
2✔
17

18

19
AVAILABLE_FIELDS = [
2✔
20
    'query'
21
]
22

23

24
class HTTPRequest(BaseHTTPRequestHandler):
2✔
25
    # http://stackoverflow.com/a/5955949
26

27
    scheme = 'http'
2✔
28

29
    # noinspection PyMissingConstructor
30
    def __init__(self, request_bytes, scheme):
2✔
31
        assert isinstance(request_bytes, bytes)
2✔
32

33
        self.scheme = scheme
2✔
34
        self.rfile = BytesIO(request_bytes)
2✔
35
        self.raw_requestline = self.rfile.readline()
2✔
36
        self.error_code = self.error_message = None
2✔
37
        self.parse_request()
2✔
38

39
        if self.error_message:
2✔
40
            raise Exception(self.error_message)
2✔
41

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

45
    def send_error(self, code, message=None, explain=None):
2✔
46
        self.error_code = code
2✔
47
        self.error_message = message
2✔
48

49
    def extract_fields(self, field=None, available_fields=None):
2✔
50
        if available_fields is None:
2✔
51
            available_fields = AVAILABLE_FIELDS
2✔
52

53
        if (field is not None) and field not in available_fields:
2✔
54
            msg = "Unexpected field '{}'. Expected one of {}."
2✔
55
            msg = msg.format(field, ', '.join(available_fields))
2✔
56
            raise ValueError(msg)
2✔
57

58
        if field is None:
2✔
59
            field = '|'.join(available_fields)
2✔
60
        is_field = r':({}) (.+): (.+)'.format(field)
2✔
61

62
        fields = []
2✔
63
        remaining_request = []
2✔
64
        cursor = self.rfile.tell()
2✔
65
        for i, line in enumerate(self.rfile.readlines()):
2✔
66
            line = line.decode('utf-8')
2✔
67
            try:
2✔
68
                field, key, val = re.match(is_field, line).groups()
2✔
69
            except AttributeError:
2✔
70
                remaining_request.append(line)
2✔
71
                continue
2✔
72
            fields.append((field.strip(), key.strip(), val.strip()))
2✔
73

74
        remaining_request = BytesIO(
2✔
75
            '\n'.join(remaining_request).encode('utf-8').strip())
76
        remaining_request.seek(0)
2✔
77
        self.rfile.seek(cursor)
2✔
78

79
        return (fields, remaining_request)
2✔
80

81
    def auth(self):
2✔
82
        try:
2✔
83
            method, token = self.headers.get('Authorization').split()
2✔
84
        except (AttributeError, KeyError, ValueError):
2✔
85
            return None, None
2✔
86
        if not isinstance(token, bytes):
2✔
UNCOV
87
            token = token.encode('utf-8')
×
88
        if method == 'Basic':
2✔
89
            return method, base64.b64decode(token).decode('utf-8')
2✔
90
        else:
91
            return method, token
2✔
92

93
    def url(self):
2✔
94
        base_url = '{}://{}{}'.format(
2✔
95
            self.scheme,
96
            self.headers.get('Host', 'nohost'),
97
            self.path
98
        )
99

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

103
        if params:
2✔
104
            new_url = add_url_params(base_url, params)
2✔
105
        else:
106
            new_url = base_url
2✔
107

108
        return new_url
2✔
109

110
    def data(self):
2✔
111
        _, payload_bytes = self.extract_fields(None)
2✔
112
        payload_bytes = payload_bytes.read()
2✔
113
        if payload_bytes:
2✔
114
            if is_json(self.headers.get('Content-Type', '')):
2✔
115
                assert isinstance(payload_bytes, bytes)
2✔
116
                payload_str = payload_bytes.decode('utf-8')
2✔
117
                return ordered(json.loads(payload_str))
2✔
118
            else:
119
                return payload_bytes
2✔
120

121

122
def parse_request(request_bytes, scheme='http'):
2✔
123
    return HTTPRequest(request_bytes, scheme)
2✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc